Revealing constructor streams

README.md

@@ -33,6 +33,8 @@ converging on a model that unifies at least promises and observables.
 
 However, this description fails to capture all of the varigated concepts of
 reactivity.
+Rather, Rx conflates all reactive primitives into a single Observable type that
+can perform any role.
 As the common paraphrase of Einstein goes, everything should be made as simple
 as possible, but no simpler.
 
@@ -240,25 +242,8 @@ iteration = iterator.next();
 expect(iteration.done).toBe(true);
 ```
 
-We propose that an iteration might also have an optional `index` property.
-This would allow the iterator to provide some additional context from the
-source.
-
-```js
-var iterator = iterate("ABC");
-var iteration = iterator.next();
-expect(iteration.value).toBe("A");
-expect(iteration.index).toBe(0);
-iteration = iterator.next();
-expect(iteration.value).toBe("B");
-expect(iteration.index).toBe(1);
-iteration = iterator.next();
-expect(iteration.value).toBe("C");
-expect(iteration.index).toBe(2);
-```
-
 What distinguishes an iterator from an array is that it is **lazy**.
-They do not necessarily end.
+An iterator does not necessarily end.
 We can have iterators for non-terminating sequences, like counting or the
 fibonacci sequence.
 The `range` function produces a sequence of values within an interval and
@@ -467,18 +452,6 @@ expect(array).toEqual([10, 20, 30]);
 Since ECMAScript 5, at Doug Crockford’s behest, JavaScript allows keywords to be
 used for property names.
 
-Just as iterations can carry an index from an array iterator, `yield` would
-accept an optional index argument.
-
-```js
-var array = [];
-var generator = generate(array);
-generator.yield(30, 2);
-generator.yield(20, 1);
-generator.yield(10, 0);
-expect(array).toEqual([10, 20, 30]);
-```
-
 And just as array iterators are just one implementation of the iterator
 interface, the generator interface could have many interfacets.
 Generator objects foreshadow the existence of stream writers.

package.json

@@ -20,6 +20,6 @@
   },
   "scripts": {
     "test": "jasminum test",
-    "docco": "docco promise-buffer.js task.js behavior.js signal.js iteration.js operators.js clock.js"
+    "docco": "docco stream.js task.js behavior.js signal.js iteration.js operators.js clock.js"
   }
 }

promise-buffer.js

@@ -154,8 +154,14 @@ PromiseGenerator.prototype.throw = function (error) {
 // except that instead of returning iterations, it returns promises for
 // iterations.
 function PromiseIterator(handler, parent) {
-    handlers.set(this, handler);
-    this.parent = parent;
+    if (parent) {
+        handlers.set(this, handler);
+        this.parent = parent;
+    } else if (typeof handler.call === "function") {
+        var buffer = new PromiseBuffer();
+        handler.call(parent, buffer.out.yield, buffer.out.return, buffer.out.throw);
+        handlers.set(this, handlers.get(buffer.out));
+    }
 }
 
 // Requests a promise for the next iteration from the generator side, and sends
@@ -445,5 +451,32 @@ PromiseIterator.prototype.filter = function (callback, thisp, limit) {
 //
 // Yet to be ported.
 
-/* TODO reduce, some, every */
+// ### fork
+
+PromiseIterator.prototype.fork = function (length) {
+    var buffers = new Array(length).map(function () {
+        return new Buffer();
+    });
+    var inputs = buffers.map(function (buffer) {
+        return buffer.in;
+    });
+    var outputs = buffers.map(function (buffer) {
+        return buffer.out;
+    });
+    this.forEach(function (value, index) {
+        return Promise.all(inputs.map(function (input) {
+            return input.yield(value, index);
+        }));
+    }).then(function (value) {
+        return Promise.all(inputs.map(function (input) {
+            return input.return(value);
+        }));
+    }, function (error) {
+        return Promise.all(inputs.map(function (input) {
+            return input.throw(value);
+        }));
+    }).done();
+    return outputs;
+};
 
+/* TODO reduce, some, every */

signal.js

@@ -65,11 +65,11 @@ SignalHandler.prototype.Iteration = Iteration;
 // Note that we track observers in reverse order to take advantage of a small
 // optimization afforded by countdown loops.
 SignalHandler.prototype.yield = function (value, index) {
+    this.value = value;
+    this.index = index;
     if (!this.active) {
         return;
     }
-    this.value = value;
-    this.index = index;
     var observers = this.observers;
     var length = observers.length;
     var observerIndex = observers.length;
@@ -116,6 +116,9 @@ SignalHandler.prototype.call = function () {
                 this.onstart();
             }
             this.active = true;
+            if (Operators.defined(this.value)) {
+                this.yield(this.value, this.index);
+            }
         }
     } else {
         if (!this.observers.length) {