Don't re-enter the consume callback (fixes #388).

lib/index.js

@@ -573,6 +573,8 @@ function Stream(/*optional*/xs, /*optional*/ee, /*optional*/mappingHint) {
     this._nil_seen = false;
     this._delegate = null;
     this._is_observer = false;
+    this._in_consume_cb = false;
+    this._repeat_resume = false;
     this.source = null;
 
     // Old-style node Stream.pipe() checks for this
@@ -887,7 +889,7 @@ Stream.prototype._sendOutgoing = function () {
 
 Stream.prototype.resume = function () {
     //console.log(['resume', this.id]);
-    if (this._resume_running) {
+    if (this._resume_running || this._in_consume_cb) {
         //console.log(['resume already processing _incoming buffer, ignore resume call']);
         // already processing _incoming buffer, ignore resume call
         this._repeat_resume = true;
@@ -1226,12 +1228,22 @@ Stream.prototype.consume = function (f) {
     s._send = function (err, x) {
         async = false;
         next_called = false;
+        s._in_consume_cb = true;
+
         f(err, x, push, next);
+
+        s._in_consume_cb = false;
         async = true;
+
         // Don't pause if x is nil -- as next will never be called after
         if (!next_called && x !== nil) {
             s.pause();
         }
+
+        if (s._repeat_resume) {
+            s._repeat_resume = false;
+            s.resume();
+        }
     };
     self._addConsumer(s);
     return s;

test/test.js

@@ -429,6 +429,75 @@ exports['consume - fork after consume should not throw (issue #366)'] = function
     }
 }
 
+exports['race'] = {
+    setUp: function (callback) {
+        this.clock = sinon.useFakeTimers();
+        callback();
+    },
+    tearDown: function (callback) {
+        this.clock.restore();
+        callback();
+    },
+    'no re-entry of consume callback': function (test) {
+        test.expect(1);
+        // This triggers a race condition. Be careful about changing the source.
+        var stream = _();
+        var i = 0;
+
+        function write() {
+            var cont = false;
+            while ((cont = stream.write(i++)) && i <= 10) {
+            }
+            if (cont) {
+                i++;
+                stream.end();
+            }
+        }
+
+        // This stream mimics the behavior of things like database
+        // drivers.
+        stream.on('drain', function () {
+            if (i === 0) {
+                // The initial read is async.
+                setTimeout(write, 0);
+            } else if (i > 0 && i < 10) {
+                // The driver loads data in batches, so subsequent drains
+                // are sync to mimic pulling from a pre-loaded buffer.
+                write();
+            } else if (i === 10) {
+                i++;
+                setTimeout(stream.end.bind(stream), 0);
+            }
+        });
+
+        var done = false;
+        stream
+            // flatMap to disassociate from the source, since sequence
+            // returns a new stream not a consume stream.
+            .flatMap(function (x) {
+                return _([x]);
+            })
+            // batch(2) to get a transform that sometimes calls next
+            // without calling push beforehand.
+            .batch(2)
+            // Another flatMap to get an async transform.
+            .flatMap(function (x) {
+                return _(function (push) {
+                    setTimeout(function () {
+                        push(null, x);
+                        push(null, _.nil);
+                    }, 100);
+                });
+            })
+            .done(function () {
+                done = true;
+            });
+        this.clock.tick(1000);
+        test.ok(done, 'The stream never completed.');
+        test.done();
+    }
+};
+
 exports['constructor'] = {
     setUp: function (callback) {
         this.createTestIterator = function(array, error, lastVal) {