v0.0.28

package.json

@@ -1,6 +1,6 @@
 {
   "name": "eventstore-node",
-  "version": "0.0.27",
+  "version": "0.0.28",
   "description": "A port of the EventStore .Net ClientAPI to Node.js",
   "main": "index.js",
   "types": "index.d.ts",
@@ -9,6 +9,7 @@
     "build": "webpack",
     "pretest": "npm run build",
     "test": "nodeunit",
+    "test-debug": "TESTS_VERBOSE_LOGGING=1 nodeunit",
     "prepublish": "npm run build && npm run gendocs",
     "gendocs": "rm -rf docs && jsdoc src -r -d docs"
   },

src/common/utils/ensure.js

@@ -22,9 +22,10 @@ module.exports.isArrayOf = function(expectedType, value, name) {
     throw new TypeError([name, " should be an array of ", expectedType.name, "."].join(""));
 };
 
-module.exports.isTypeOf = function(expectedType, value, name) {
+module.exports.isTypeOf = function(expectedType, value, name, nullAllowed) {
+  if (nullAllowed && value === null) return;
   if (!(value instanceof expectedType))
-    throw new TypeError([name, " should be of type '", expectedType.name, "'."].join(""));
+    throw new TypeError([name, " should be of type '", expectedType.name, "'", nullAllowed ? " or null": "", "."].join(""));
 };
 
 module.exports.positive = function(value, name) {

src/eventStoreAllCatchUpSubscription.js

@@ -46,8 +46,6 @@ EventStoreAllCatchUpSubscription.prototype._readEventsTill = function(
                 var done = lastCommitPosition === null
                     ? slice.isEndOfStream
                     : slice.nextPosition.compareTo(new results.Position(lastCommitPosition, lastCommitPosition)) >= 0;
-                if (!done && slice.isEndOfStream)
-                    return Promise.resolve(done).delay(10);
                 return Promise.resolve(done);
               });
         })

src/eventStoreNodeConnection.js

@@ -492,7 +492,10 @@ EventStoreNodeConnection.prototype.subscribeToAllFrom = function(
     lastCheckpoint, resolveLinkTos, eventAppeared, liveProcessingStarted, subscriptionDropped,
     userCredentials, readBatchSize
 ) {
-  if (typeof eventAppeared !== 'function') throw new TypeError("eventAppeared must be a function.");
+  ensure.isTypeOf(results.Position, lastCheckpoint, "lastCheckpoint", true);
+  ensure.isTypeOf(Function, eventAppeared, "eventAppeared", false);
+  if (liveProcessingStarted) ensure.isTypeOf(Function, liveProcessingStarted, "liveProcessingStarted", false);
+  if (subscriptionDropped) ensure.isTypeOf(Function, subscriptionDropped, "subscriptionDropped", false);
 
   var catchUpSubscription =
       new EventStoreAllCatchUpSubscription(this, this._settings.log, lastCheckpoint, resolveLinkTos,

test/subscribeToAllFrom_test.js

@@ -8,7 +8,7 @@ function createRandomEvent() {
 }
 
 module.exports = {
-  'Test Subscribe to All From': function(test) {
+  'Test Subscribe to All From (Start)': function(test) {
     test.expect(4);
     var self = this;
     var liveProcessing = false;
@@ -44,6 +44,88 @@ module.exports = {
       done(error);
     }
     var subscription = this.conn.subscribeToAllFrom(null, false, eventAppeared, liveProcessingStarted, subscriptionDropped, allCredentials);
+  },
+  'Test Subscribe to All From (Position)': function(test) {
+    test.expect(5);
+    var self = this;
+    var liveProcessing = false;
+    var catchUpEvents = [];
+    var liveEvents = [];
+    var _doneCount = 0;
+    function done(err) {
+      test.ok(!err, err ? err.stack : '');
+      _doneCount++;
+      if (_doneCount < 2) return;
+      test.done();
+    }
+    function eventAppeared(s, e) {
+      if (liveProcessing) {
+        liveEvents.push(e);
+        s.stop();
+      } else {
+        catchUpEvents.push(e);
+      }
+    }
+    function liveProcessingStarted() {
+      liveProcessing = true;
+      var events = [createRandomEvent()];
+      self.conn.appendToStream(self.testStreamName, client.expectedVersion.any, events)
+        .then(function () {
+          done();
+        })
+        .catch(done);
+    }
+    function subscriptionDropped(connection, reason, error) {
+      test.ok(liveEvents.length === 1, "Expecting 1 live event, got " + liveEvents.length);
+      test.ok(catchUpEvents.length > 1, "Expecting at least 1 catchUp event, got " + catchUpEvents.length);
+      done(error);
+    }
+    this.conn.readAllEventsForward(client.positions.start, 512, true, allCredentials)
+      .then(function (slice) {
+        var subscription = self.conn.subscribeToAllFrom(slice.nextPosition, false, eventAppeared, liveProcessingStarted, subscriptionDropped, allCredentials);
+        test.ok(subscription, "Subscription is null/undefined.");
+      });
+  },
+  'Test Subscribe to All From (End)': function(test) {
+    test.expect(5);
+    var self = this;
+    var liveProcessing = false;
+    var catchUpEvents = [];
+    var liveEvents = [];
+    var _doneCount = 0;
+    function done(err) {
+      test.ok(!err, err ? err.stack : '');
+      _doneCount++;
+      if (_doneCount < 2) return;
+      test.done();
+    }
+    function eventAppeared(s, e) {
+      if (liveProcessing) {
+        liveEvents.push(e);
+        s.stop();
+      } else {
+        catchUpEvents.push(e);
+      }
+    }
+    function liveProcessingStarted() {
+      liveProcessing = true;
+      var events = [createRandomEvent()];
+      self.conn.appendToStream(self.testStreamName, client.expectedVersion.any, events)
+        .then(function () {
+          done();
+        })
+        .catch(done);
+    }
+    function subscriptionDropped(connection, reason, error) {
+      test.ok(liveEvents.length === 1, "Expecting 1 live event, got " + liveEvents.length);
+      test.ok(catchUpEvents.length === 0, "Expecting 0 catchUp event, got " + catchUpEvents.length);
+      done(error);
+    }
+    this.conn.readAllEventsForward(client.positions.end, 512, true, allCredentials)
+      .then(function (slice) {
+        var subscription = self.conn.subscribeToAllFrom(slice.nextPosition, false, eventAppeared, liveProcessingStarted, subscriptionDropped, allCredentials);
+        test.ok(subscription, "Subscription is null/undefined.");
+      });
   }
 };