fix: dont drain source while drain in progress

[olizilla]

Misbehaving streams can end up calling drain multiple times while
a drain is in progress. This is bad news for pull streams as it
breaks the contract and you end up with weirdness like end of stream
events getting triggered before data event callbacks are fired.

While tracking this down I applied a similar fix to pull-file-reader
but that modules shouldn't have to guard against bad pullers.

Then I found pull-stream/pull-stream-to-stream#6
which deals with the same issue in pull-stream-to-steam (thanks @mmckegg!)
and recreated it here. The test on this PR is copied from there.
The gist is, as this module operates at the boundary of node-streams
to pull-streams, it should deal with bad node-stream behaviour to
avoid messing up connected pull-streams.

fixes: ipfs-inactive/js-ipfs-http-client#967
fixes: ipfs/ipfs-webui#1010

[codecov-io]

Codecov Report

Merging #1 into master will increase coverage by 1.84%.
The diff coverage is 100%.

@@            Coverage Diff             @@
##           master       #1      +/-   ##
==========================================
+ Coverage   80.85%   82.69%   +1.84%     
==========================================
  Files           1        1              
  Lines          47       52       +5     
==========================================
+ Hits           38       43       +5     
  Misses          9        9

Impacted Files	Coverage Δ
index.js	82.69% <100%> (+1.84%)	⬆️

---

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 487a091...72629c9. Read the comment docs.

[hugomrdias]

thanks @olizilla for tracking this down, looked a little more into this and it's not about misbehaving streams we simply CANNOT always can this.drainPull inside _read.

i would like suggest some changes to make this PR to be more explicit

   constructor(source, sink, options) {
        super(options);
        this.source = source;
        this.drainingSource = false;
        this.sink = sink;
        this.input = [];
        this.writeCallbacks = [];
        this.internalSourceCallbacks = [];
        if (this.sink) {
            this.sink(this._internalSource.bind(this));
        }
    }

    drainPull() {
        const self = this;
        this.drainingSource = true;
        this.source(null, function next(end, data) {
            if (end instanceof Error) {
                return self.emit('error', end);
            }
            
            if (end) {
                return self.push(null);
            }
            
            if (self.push(data)) {
                self.source(null, next);
            } else {
                this.drainingSource = false;
            }
        });
    }

    _read() {
        if (this.source && !this.drainingSource) {
            this.drainPull();
        }
    }

can you please make these changes ? if not ill do it.

[olizilla]

👍 feel free to update the PR!