Implement shortest-paths by hand in RxFlow.

Author: JoshRosen

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

@@ -0,0 +1,102 @@
+function Bloom () {
+    var rx = require('rx');
+    var rxflow = require('rxflow');
+
+    var tables = {
+        "path": new rxflow.Table(3) /* table path, [from: string, to: string, nxt: string, cost: int] */,
+        "shortest": new rxflow.Table(3) /* table path, [from: string, to: string, nxt: string, cost: int] */
+    };
+
+    var inputs = {
+        "link": new rx.Subject() /* input link, [from: string, to: string, cost: int] */
+    };
+    this.link = inputs["link"];
+
+    var outputs = {
+        "pathOut": new rx.Subject() /* output pathOut, [from: string, to: string, nxt: string, cost: int] */
+    };
+    this.pathOut = outputs["pathOut"];
+
+    var elements = {
+        6: new rxflow.HashJoin(
+            function(x) { return x[1]; /* link.to */ },
+            function(x) { return x[0]; /* path.from */ },
+            "right"
+        ),
+        5: new rxflow.HashJoin(
+            function(x) { return x[1]; /* link.to */ },
+            function(x) { return x[0]; /* path.from */ },
+            "left"
+        ),
+        0: new rxflow.Map(
+            function(x) { return [x[0], x[1], x[1], x[2]]; /* [l.from, l.to, l.to, l.cost] */ }
+        ),
+        2: new rxflow.ObservableScanner(inputs["link"]),
+        7: new rxflow.Map(
+            function(x) { return [x[0][0], x[1][1], x[0][1], x[0][2] + x[1][3]]; /* [l.from, p.to, l.to, l.cost + p.cost] */ }
+        ),
+        4: new rxflow.TableScanner(tables["path"]),
+        9: new rxflow.TableScanner(tables["shortest"]),
+        8: new rxflow.ArgMin(
+            function(x) { return [x[0], x[1]]; /* [path.from, path.to] */ },
+            function(x) { return x[3]; /* path.cost */ },
+            function(x, y) { return x <= y; }
+        )
+    };
+
+    var invalidationLookupTable = { "link": [2], "path": [7, 5, 0, 6] };
+
+    var rescanLookupTable = { "path": [2, 4] };
+
+    // Wiring, in a roughly topological order
+    // Scanner outputs:
+    elements[2].output.subscribe(elements[5].leftInput);
+    elements[2].output.subscribe(elements[6].leftInput);
+    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(elements[8].input);
+    elements[8].output.subscribe(outputs["pathOut"]);
+
+    // Join outputs:
+    elements[5].output.subscribe(elements[7].input);
+    elements[6].output.subscribe(elements[7].input);
+
+    // Table insertions:
+    elements[0].output.subscribe(tables["path"].insert);  // Link -> Path map
+    elements[7].output.subscribe(tables["path"].insert);  // Join projection
+
+    function tickStratum0() {
+        var tuplesFlushed = 0;
+        tuplesFlushed += elements[2].flush();
+        tuplesFlushed += elements[4].flush();
+        return tuplesFlushed;
+    }
+
+    function tickStratum1() {
+         var tuplesFlushed = 0;
+        tuplesFlushed += elements[8].flush();
+        return tuplesFlushed;
+    }
+
+    this.tick = function() {
+        var atFixpoint = false;
+        while (!atFixpoint) {
+            atFixpoint = tickStratum0() === 0;
+        }
+        tickStratum1()
+    }
+}
+
+var bloom = new Bloom();
+
+bloom.pathOut.subscribe(function(x) { console.log(x)});
+bloom.link.onNext(['a', 'b', 1]);
+bloom.link.onNext(['a', 'b', 4]);
+bloom.link.onNext(['b', 'c', 1]);
+bloom.link.onNext(['c', 'd', 1]);
+bloom.link.onNext(['d', 'e', 1]);
+bloom.tick();
+console.log('----');
+bloom.link.onNext(['e', 'f', 1]);
+bloom.tick();

rxflow/src/Aggregate.js

@@ -13,6 +13,7 @@ var Rx = require('rx');
 function Aggregate(keyFunction, aggregates) {
     'use strict';
 
+    var _this = this;
     var _aggregators = [];
     var _groupKeys = [];
     var _keyToArrayIndex = {};
@@ -40,6 +41,12 @@ function Aggregate(keyFunction, aggregates) {
      */
     this.input = Rx.Observer.create(updateAggs);
 
+    this.output = new Rx.Subject();
+
+    this.flush = function() {
+        this.getCurrentValues().subscribe(_this.output);
+    };
+
     /**
      * Return a stream of groups and values as an Rx observable.
      */
@@ -49,15 +56,14 @@ function Aggregate(keyFunction, aggregates) {
             for (var i = 0; i < _aggregators.length; ++i) {
                 observer.onNext([_groupKeys[i]].concat(_aggregators[i].map(extractValue)));
             }
-            observer.onCompleted();
         });
     };
 
     /**
      * Reset this element by resetting aggregates to their initial values
      * and clearing all groups.
      */
-    this.reset = function() {
+    this.invalidate = function() {
         _aggregators = [];
         _groupKeys = [];
         _keyToArrayIndex = {};

rxflow/src/ArgMin.js

@@ -0,0 +1,31 @@
+var Rx = require('rx');
+var Aggregate = require('./Aggregate');
+
+
+function ArgMin(keyFunction, orderingFields, orderingFunction) {
+    'use strict';
+
+    var _this = this;
+
+    var aggregateFunction = function () {
+        var value = null;
+        this.getValue = function() { return value; };
+        this.next = function(x) {
+            if (value === null || orderingFunction(orderingFields(x), orderingFields(value))) {
+                value = x;
+            }
+        };
+    };
+    var aggregate = new Aggregate(keyFunction, [aggregateFunction]);
+
+    this.input = aggregate.input;
+    this.output = new Rx.Subject();
+
+    this.flush = function () {
+        aggregate.getCurrentValues().map(function(x) { return x[1]; }).subscribe(_this.output);
+    };
+
+    this.invalidate = aggregate.invalidate;
+
+}
+module.exports = ArgMin;

rxflow/src/index.js

@@ -1,5 +1,6 @@
 var rxflow = {};
 rxflow.Aggregate = require('./Aggregate');
+rxflow.ArgMin = require('./ArgMin');
 rxflow.Buffer = require('./Buffer');
 rxflow.HashJoin = require('./HashJoin');
 rxflow.Map = require('./Map');