Fix ending a stream while updating

Author: nordfjord

.gitignore

@@ -6,4 +6,4 @@ logs
 # Dependency directory
 node_modules
 bower_components
-
+.idea

lib/index.js

@@ -10,6 +10,7 @@ function trueFn() { return true; }
 
 // Globals
 var toUpdate = [];
+var toEnd = [];
 var inStream;
 var order = [];
 var orderNextIdx = -1;
@@ -696,6 +697,9 @@ function flushUpdate() {
     var nextUpdateFn = stream.updaters.shift();
     if (nextUpdateFn && stream.shouldUpdate) nextUpdateFn(stream);
   }
+  while (toEnd.length > 0) {
+    toEnd.shift()(true);
+  }
   flushingUpdateQueue = false;
 }
 
@@ -715,6 +719,8 @@ function updateStreamValue(n, s) {
     flushingStreamValue = false;
   } else if (inStream === s) {
     markListeners(s, s.listeners);
+  } else if (inStream.end === s) {
+    toEnd.push(inStream.end)
   } else {
     updateLaterUsing(function(s) { updateStreamValue(n, s); }, s);
   }

test/delayed-stream-end.js

@@ -0,0 +1,40 @@
+var assert = require('assert');
+var R = require('ramda');
+
+var flyd = require('../lib');
+
+function once(s) {
+  return flyd.combine(function(s, self) {
+    self(s.val)
+    self.end(true)
+  }, [s])
+}
+function withLatestFrom() {
+  var streams = arguments
+  return flyd.combine(function() {
+    var self = arguments[arguments.length - 2]
+    var result = []
+    for (var i = 0; i < streams.length; ++i) {
+      if (!streams[i].hasVal) return
+      result.push(streams[i].val)
+    }
+    self(result)
+  }, streams)
+}
+
+describe('ending a stream', function() {
+  it('delays ending the current stream until dependents have been updated', function() {
+    var stream = flyd.stream(1)
+    function doubled() { return stream.map(R.multiply(2)) }
+    var count = 0;
+    stream
+      .map(function() {
+        withLatestFrom(doubled(), doubled())
+          .pipe(once)
+          .map(function() {
+            count++
+          })
+      })
+    assert.equal(count, 1)
+  })
+})