fix: dont drain source while drain in progress

index.js

@@ -16,9 +16,17 @@ class PullDuplexStream extends Duplex {
     }
 
     drainPull() {
+        // Ensure misbehaving streams wait for a previous drain to finish before
+        // letting them pull again.
+        if (this.busy) {
+            return;
+        }
+        this.busy = true;
+
         const self = this;
 
         this.source(null, function next(end, data) {
+            self.busy = false;
             if (end instanceof Error) {
                 return self.emit('error', end);
             }
@@ -28,6 +36,7 @@ class PullDuplexStream extends Duplex {
             }
 
             if (self.push(data)) {
+                self.busy = true;
                 self.source(null, next);
             }
         });

test.js

@@ -78,6 +78,35 @@ test.cb('source pause/resume', (t) => {
     readable.on('end', () => t.end());
 });
 
+test.cb('source ensure drain not called while drain in progress', (t) => {
+    const values = [0, 1, 2, 3, 4, 5, 6, 7, 8];
+    const readable = toStream.readable(pull(
+        pull.values(values),
+        pull.asyncMap((value, cb) => {
+            setTimeout(() => {
+                // pass through value with delay
+                cb(null, value);
+            }, 10);
+        })
+    ));
+
+    const timer = setInterval(() => {
+        readable.resume();
+    }, 5);
+
+    const output = [];
+
+    readable.on('data', (c) => {
+        output.push(c);
+    });
+
+    readable.once('end', () => {
+        clearInterval(timer);
+        t.deepEqual(output, values, 'End called after all values emitted');
+        t.end();
+    });
+});
+
 test.cb('sink basic', (t) => {
     t.plan(1);