[api test] Added .thru() for piping to multiple independent reactors.

nodejitsu · Mar 9, 2013 · 3fb47e7 · 3fb47e7
1 parent 20059b5
commit 3fb47e7
Show file tree

Hide file tree

Showing 5 changed files with 105 additions and 1 deletion.
diff --git a/lib/godot/reactor/index.js b/lib/godot/reactor/index.js
@@ -67,6 +67,7 @@ reactor.register = function (method, Stream) {
  'sum',
  'tagged',
  'throttle',
+ 'thru',
  'under',
  'where',
  'within'].forEach(function (name) {

diff --git a/lib/godot/reactor/thru.js b/lib/godot/reactor/thru.js
@@ -0,0 +1,40 @@
+/*
+ * thru.js: Stream for piping to multiple independent reactors 
+ *
+ * (C) 2013, Nodejitsu Inc.
+ *
+ */
+
+var util = require('util'),
+    ReadWriteStream = require('../common/read-write-stream'),
+    Reactor = require('./reactor');
+
+//
+// ### function Thru (reactor0, reactor1, ...)
+// #### @reactor0,...reactorN {godot.reactor().type()*} Reactors to be created
+// Constructor function for the thru stream responsible for piping to
+// multiple independent reactors.
+//
+var Thru = module.exports = function () {
+  ReadWriteStream.call(this);
+
+  var reactors = Array.prototype.slice.call(arguments),
+      self     = this;
+
+  reactors.forEach(function (reactor) {
+    if (!(reactor instanceof Reactor)) {
+      throw new Error('This reactor a set of godot.reactor() arguments');
+    }
+  });
+
+  this.reactors = reactors;
+  this.streams  = reactors.map(function (reactor) {
+    var source = self.pipe(new ReadWriteStream());
+    return reactor.createStream(source);
+  });
+};
+
+//
+// Inherit from ReadWriteStream
+//
+util.inherits(Thru, ReadWriteStream);
diff --git a/test/reactor/by-test.js b/test/reactor/by-test.js
@@ -1,5 +1,5 @@
 /*
- * change-test.js: Tests for the Change reactor stream.
+ * by-test.js: Tests for the By reactor stream.
  *
  * (C) 2012, Nodejitsu Inc.
  *

diff --git a/test/reactor/reactor-test.js b/test/reactor/reactor-test.js
@@ -32,6 +32,7 @@ vows.describe('godot/reactor').addBatch({
        'taggedAny',
        'taggedAll',
        'throttle',
+       'thru',
        'timeWindow',
        'under',
        'where',

diff --git a/test/reactor/thru-test.js b/test/reactor/thru-test.js
@@ -0,0 +1,62 @@
+/*
+ * thru-test.js: Tests for the Thru reactor stream.
+ *
+ * (C) 2012, Nodejitsu Inc.
+ *
+ */
+
+var assert = require('assert'),
+    vows = require('vows'),
+    godot = require('../../lib/godot'),
+    macros = require('../macros').reactor;
+
+var counts = [0, 0, 0];
+
+//
+// Helper function to increment the 
+// appropriate count.
+//
+function increment(i) {
+  return function (data) {
+    counts[i]++;
+    return data;
+  };
+}
+
+vows.describe('godot/reactor/thru').addBatch({
+  "Godot thru": {
+    "one reactor": macros.shouldEmitDataSync(
+      godot.reactor()
+        .thru(
+          godot.reactor().map(increment(0))
+        ),
+      'by',
+      6
+    ),
+    "two reactors": macros.shouldEmitDataSync(
+      godot.reactor()
+        .thru(
+          godot.reactor().map(increment(1)),
+          godot.reactor().map(increment(1))
+        ),
+      'by',
+      6
+    ),
+    "three reactors": macros.shouldEmitDataSync(
+      godot.reactor()
+        .thru(
+          godot.reactor().map(increment(2)),
+          godot.reactor().map(increment(2)),
+          godot.reactor().map(increment(2))
+        ),
+      'by',
+      6
+    ),
+  }
+}).addBatch({
+  "should emit pipe the events to the correct pipe-chains": function () {
+    counts.forEach(function (length, i) {
+      assert.equal(length, 6 * (i + 1));
+    });
+  }
+}).export(module);