Merge pull request #564 from vqvu/fix-async-push-nil-deadlock

Fix the deadlock that happens when nil is pushed asynchronously in consume but x !== nil.

CHANGELOG.md

@@ -8,7 +8,13 @@ this library.
 2.10.1
 ------
 ### Bugfix
-* Improved documentation. Examples are now more standalone.
+* Asynchronously pushing a `nil` in `consume` when then input value wasn't a
+  `nil` itself now no longer causes the stream to deadlock.
+  [#564](https://github.com/caolan/highland/pull/564).
+  Fixes [#563](https://github.com/caolan/highland/issues/563).
+  Related to [#558](https://github.com/caolan/highland/issues/558).
+* Much improved documentation. Examples are now more standalone, and more
+  guidance was added for certain common pitfalls.
 
 2.10.0
 ------

lib/index.js

@@ -1420,6 +1420,9 @@ Stream.prototype.consume = function (f) {
 
     // Hack. Not needed in v3.0.
     s._is_consumer = true;
+
+    var async;
+    var next_called;
     var _send = s._send;
     var push = function (err, x) {
         //console.log(['push', err, x, s.paused]);
@@ -1430,6 +1433,12 @@ Stream.prototype.consume = function (f) {
             // ended, remove consumer from source
             s._nil_pushed = true;
             self._removeConsumer(s);
+
+            // We previously paused the stream, but since a nil was pushed,
+            // next won't be called and we must manually resume.
+            if (async) {
+                s.resume();
+            }
         }
         if (s.paused) {
             if (err) {
@@ -1443,8 +1452,6 @@ Stream.prototype.consume = function (f) {
             _send.call(s, err, x);
         }
     };
-    var async;
-    var next_called;
     var next = function (s2) {
         //console.log(['next', async]);
         if (s._nil_pushed) {

test/test.js

@@ -422,6 +422,27 @@ exports['consume - push nil async (issue #173)'] = function (test) {
     });
 };
 
+exports['consume - push nil async when x !== nil (issue #563)'] = function (test) {
+    test.expect(1);
+    _([1, 2, 3, 4]).consume(function(err, x, push, next) {
+        if (err !== null) {
+            push(err);
+            next();
+        }
+        else if (x === _.nil) {
+            push(null, _.nil);
+        }
+        else {
+            _.setImmediate(push.bind(this, null, _.nil));
+        }
+    })
+    .toArray(function (xs) {
+        test.same(xs, []);
+        test.done();
+    });
+};
+
+
 exports['consume - fork after consume should not throw (issue #366)'] = function (test) {
     test.expect(2);
     var arr1, arr2;