Adding persistent subscription support

2016-03-10 22:57:39 -08:00
parent 29f6bf210b
commit f1c3c42d46
25 changed files with 1939 additions and 117 deletions
--- a/src/eventStorePersistentSubscriptionBase.js
+++ b/src/eventStorePersistentSubscriptionBase.js
@ -0,0 +1,167 @@
+var util = require('util');
+var ensure = require('./common/utils/ensure');
+var PersistentSubscriptionNakEventAction = require('./persistentSubscriptionNakEventAction');
+var SubscriptionDropReason = require('./subscriptionDropReason');
+
+function DropSubscriptionEvent() {}
+
+function EventStorePersistentSubscriptionBase(
+    subscriptionId, streamId,
+    eventAppeared, subscriptionDropped,
+    userCredentials, log, verboseLogging, settings, bufferSize, autoAck
+) {
+  bufferSize = bufferSize === undefined ? 10 : bufferSize;
+  autoAck = autoAck === undefined ? true : autoAck;
+
+  this._subscriptionId = subscriptionId;
+  this._streamId = streamId;
+  this._eventAppeared = eventAppeared;
+  this._subscriptionDropped = subscriptionDropped;
+  this._userCredentials = userCredentials;
+  this._log = log;
+  this._verbose = verboseLogging;
+  this._settings = settings;
+  this._bufferSize = bufferSize;
+  this._autoAck = autoAck;
+
+  this._subscription = null;
+  this._dropData = null;
+  this._queue = [];
+  this._isProcessing = false;
+  this._isDropped = false;
+}
+
+EventStorePersistentSubscriptionBase.prototype.start = function() {
+  this._stopped = false;
+
+  var self = this;
+  this._startSubscription(this._subscriptionId, this._streamId, this._bufferSize, this._userCredentials,
+                          this._onEventAppeared.bind(this), this._onSubscriptionDropped.bind(this), this._settings)
+      .then(function(subscription) {
+        self._subscription = subscription;
+      });
+};
+
+EventStorePersistentSubscriptionBase.prototype._startSubscription = function() {
+  throw new Error("EventStorePersistentSubscriptionBase._startSubscription abstract method called." +
+                  this.constructor.name);
+};
+
+/**
+ * @param {ResolvedEvent[]|ResolvedEvent} events
+ */
+EventStorePersistentSubscriptionBase.prototype.acknowledge = function(events) {
+  ensure.notNull(events, "events");
+
+  if (this._subscription === null) throw new Error("Invalid operation. Subscription is not active yet.");
+  if (!Array.isArray(events))
+    events = [events];
+  var ids = events.map(function(x) { return x.originalEvent.eventId; });
+  this._subscription.notifyEventsProcessed(ids);
+};
+
+/**
+ * @param {ResolvedEvent[]|ResolvedEvent} events
+ * @param {number} action One of PersistentSubscriptionNakEventAction
+ * @param {string} reason
+ */
+EventStorePersistentSubscriptionBase.prototype.fail = function(events, action, reason) {
+  ensure.notNull(events, "events");
+  PersistentSubscriptionNakEventAction.isValid(action);
+  ensure.notNull(reason, "reason");
+
+  if (this._subscription === null) throw new Error("Invalid operation. Subscription is not active yet.");
+  if (!Array.isArray(events))
+    events = [events];
+  var ids = events.map(function(x) { return x.originalEvent.eventId; });
+  this._subscription.notifyEventsFailed(ids, action, reason);
+};
+
+EventStorePersistentSubscriptionBase.prototype.stop = function() {
+  if (this._verbose) this._log.debug("Persistent Subscription to %s: requesting stop...", this._streamId);
+  this._enqueueSubscriptionDropNotification(SubscriptionDropReason.UserInitiated, null);
+  //TODO figure out timeout with Promise still running
+  //if (!_stopped.Wait(timeout))
+    //throw new TimeoutException(string.Format("Could not stop {0} in time.", GetType().Name));
+};
+
+EventStorePersistentSubscriptionBase.prototype._enqueueSubscriptionDropNotification = function(reason, error) {
+  // if drop data was already set -- no need to enqueue drop again, somebody did that already
+  if (!this._dropData) {
+    this._dropData = {reason: reason, error: error};
+    this._enqueue(new DropSubscriptionEvent());
+  }
+};
+
+EventStorePersistentSubscriptionBase.prototype._onSubscriptionDropped = function(subscription, reason, exception) {
+  this._enqueueSubscriptionDropNotification(reason, exception);
+};
+
+EventStorePersistentSubscriptionBase.prototype._onEventAppeared = function(subscription, resolvedEvent) {
+  this._enqueue(resolvedEvent);
+};
+
+EventStorePersistentSubscriptionBase.prototype._enqueue = function(resolvedEvent) {
+  this._queue.push(resolvedEvent);
+  if (!this._isProcessing) {
+    this._isProcessing = true;
+    setImmediate(this._processQueue.bind(this));
+  }
+};
+
+EventStorePersistentSubscriptionBase.prototype._processQueue = function() {
+  //do
+  //{
+    var e = this._queue.shift();
+    while (e)
+    {
+      if (e instanceof DropSubscriptionEvent) // drop subscription artificial ResolvedEvent
+      {
+        if (this._dropData === null) throw new Error("Drop reason not specified.");
+        this._dropSubscription(this._dropData.reason, this._dropData.error);
+        return;
+      }
+      if (this._dropData !== null)
+      {
+        this._dropSubscription(this._dropData.reason, this._dropData.error);
+        return;
+      }
+      try
+      {
+        this._eventAppeared(this, e);
+        if(this._autoAck)
+          this._subscription.notifyEventsProcessed([e.originalEvent.eventId]);
+        if (this._verbose)
+          this._log.debug("Persistent Subscription to %s: processed event (%s, %d, %s @ %d).",
+              this._streamId, e.originalEvent.eventStreamId, e.originalEvent.eventNumber, e.originalEvent.eventType,
+              e.originalEventNumber);
+      }
+      catch (err)
+      {
+        //TODO GFY should we autonak here?
+        this._dropSubscription(SubscriptionDropReason.EventHandlerException, err);
+        return;
+      }
+      e = this._queue.shift();
+    }
+    this._isProcessing = false;
+  //} while (_queue.Count > 0 && Interlocked.CompareExchange(ref _isProcessing, 1, 0) == 0);
+};
+
+EventStorePersistentSubscriptionBase.prototype._dropSubscription = function(reason, error) {
+  if (!this._isDropped)
+  {
+    this._isDropped = true;
+    if (this._verbose)
+      this._log.debug("Persistent Subscription to %s: dropping subscription, reason: %s %s.",
+                      this._streamId, reason, error);
+
+    if (this._subscription !== null)
+      this._subscription.unsubscribe();
+    if (this._subscriptionDropped !== null)
+      this._subscriptionDropped(this, reason, error);
+    this._stopped = true;
+  }
+};
+
+module.exports = EventStorePersistentSubscriptionBase;