Start to add end-of-round punctuations.

Author: JoshRosen

compiler/src/test/resources/hand-generated-outputs/shortest-paths-rxflow.js

@@ -79,7 +79,6 @@ function Bloom () {
     function tickStratum0() {
         var tuplesFlushed = 0;
         tuplesFlushed += elements[2].flush();
-        tuplesFlushed += elements[4].flush();
         return tuplesFlushed;
     }
 

compiler/src/test/resources/hand-generated-outputs/transitive-closure-rxflow.js

@@ -12,9 +12,9 @@ function Bloom () {
     this.link = inputs["link"];
 
     var outputs = {
-        "pathOut": new rx.Subject() /* output pathOut, [from: string, to: string, nxt: string, cost: int] */
+        "pathOut": new rxflow.ObservableSink() /* output pathOut, [from: string, to: string, nxt: string, cost: int] */
     };
-    this.pathOut = outputs["pathOut"];
+    this.pathOut = outputs["pathOut"].output;
 
     var elements = {
         6: new rxflow.HashJoin(
@@ -48,7 +48,7 @@ function Bloom () {
     elements[2].output.subscribe(elements[0].input);
     elements[4].output.subscribe(elements[5].rightInput);
     elements[4].output.subscribe(elements[6].rightInput);
-    elements[4].output.subscribe(outputs["pathOut"]);
+    elements[4].output.subscribe(outputs["pathOut"].input);
 
     // Join outputs:
     elements[5].output.subscribe(elements[7].input);
@@ -61,7 +61,6 @@ function Bloom () {
     function tickInternal() {
         var tuplesFlushed = 0;
         tuplesFlushed += elements[2].flush();
-        tuplesFlushed += elements[4].flush();
         return tuplesFlushed;
     }
 

rxflow/benchmark/AggregateBenchmark.js

@@ -60,9 +60,10 @@ module.exports = {
                     var group = function(data) { return data[column]; };
                     var aggregate = new global.rxflow.Aggregate(group, [Count]);
                     var data = global.Rx.Observable.fromArray(testData);
+                    var observableScanner = new global.rxflow.ObservableScanner(data);
                     var sink = new global.rxflow.ObservableSink();
                     aggregate.output.subscribe(sink.input);
-                    data.subscribe(aggregate.input);
+                    observableScanner.output.subscribe(aggregate.input);
                     var results = [];
                     sink.output.forEach(function (x) { results.push(x); });
                     aggregate.flush();

rxflow/src/Aggregate.js

@@ -1,6 +1,3 @@
-/// <reference path="../typings/rx.js/rx.d.ts" />
-
-import Rx = require('rx');
 import DataflowElement = require('./DataflowElement');
 import InputPort = require('./InputPort');
 import OutputPort = require('./OutputPort');
@@ -64,9 +61,9 @@ class Aggregate<T> extends DataflowElement {
     /**
      * An input stream of elements to be aggregated.
      */
-    input = new InputPort(x => this.updateAggs(x));
+    input = new InputPort(x => this.updateAggs(x), this);
 
-    output = new OutputPort();
+    output = new OutputPort(this);
 
     flush() {
         for (var i = 0; i < this.aggregators.length; ++i) {

rxflow/src/Aggregate.ts

@@ -8,7 +8,7 @@ class ArgMin<T> extends DataflowElement {
 
     private aggregate: Aggregate<T>;
     input: InputPort<T>;
-    output = new OutputPort();
+    output = new OutputPort(this);
 
     constructor(keyFunction, orderingFields, orderingFunction) {
         super();
@@ -23,7 +23,7 @@ class ArgMin<T> extends DataflowElement {
         };
         this.aggregate = new Aggregate(keyFunction, [aggregateFunction]);
         this.input = this.aggregate.input;
-        var outputProjector = new InputPort(x => this.output.onNext(x[1]));
+        var outputProjector = new InputPort(x => this.output.onNext(x[1]), null);
         this.aggregate.output.subscribe(outputProjector);
     }
 

rxflow/src/ArgMin.js

@@ -6,8 +6,8 @@ class Buffer<T> extends DataflowElement {
 
     private buffer: Array<T> = [];
 
-    input = new InputPort<T>(x => this.buffer.push(x));
-    output = new OutputPort<T>();
+    input = new InputPort<T>(x => this.buffer.push(x), this);
+    output = new OutputPort<T>(this);
 
 
     invalidate() {