add subscribe since API (#146)

centrifugal · Aug 28, 2021 · 9f5570d · 9f5570d
1 parent 799076b
commit 9f5570d
Show file tree

Hide file tree

Showing 3 changed files with 26 additions and 5 deletions.
diff --git a/dist/centrifuge.d.ts b/dist/centrifuge.d.ts
@@ -50,8 +50,8 @@ declare class Centrifuge extends EventEmitter {
     stopBatching(): void;
     startSubscribeBatching(): void;
     stopSubscribeBatching(): void;
-    subscribe(channel: string, events?: (...args: any[]) => void): Centrifuge.Subscription;
-    subscribe(channel: string, events?: Centrifuge.SubscriptionEvents): Centrifuge.Subscription;
+    subscribe(channel: string, events?: (...args: any[], opts?: Centrifuge.SubscribeOptions) => void): Centrifuge.Subscription;
+    subscribe(channel: string, events?: Centrifuge.SubscriptionEvents, opts?: Centrifuge.SubscribeOptions): Centrifuge.Subscription;
 }
 
 declare namespace Centrifuge {
@@ -93,7 +93,7 @@ declare namespace Centrifuge {
     export class Subscription extends EventEmitter {
         channel: string;
         ready(callback: (ctx: SubscribeSuccessContext) => void, errback: (ctx: SubscribeErrorContext) => void): void;
-        subscribe(): void;
+        subscribe(opts?: SubscribeOptions): void;
         unsubscribe(): void;
         publish(data: any): Promise<PublishResult>;
         presence(): Promise<PresenceResult>;
@@ -207,6 +207,10 @@ declare namespace Centrifuge {
         reverse?: boolean;
     }
 
+    export interface SubscribeOptions {
+        since?: StreamPosition;
+    }
+
     export interface StreamPosition {
         offset: number;
         epoch: string;

diff --git a/src/centrifuge.js b/src/centrifuge.js
@@ -1887,7 +1887,13 @@ export class Centrifuge extends EventEmitter {
     }
   };
 
-  subscribe(channel, events) {
+  _setSubscribeSince(sub, since) {
+    this._lastOffset[sub.channel] = since.offset;
+    this._lastEpoch[sub.channel] = since.epoch;
+    sub._setNeedRecover(true);
+  }
+
+  subscribe(channel, events, opts) {
     const currentSub = this._getSub(channel);
     if (currentSub !== null) {
       currentSub._setEvents(events);
@@ -1898,6 +1904,9 @@ export class Centrifuge extends EventEmitter {
     }
     const sub = new Subscription(this, channel, events);
     this._subs[channel] = sub;
+    if (opts && opts.since) {
+      this._setSubscribeSince(sub, opts.since);
+    }
     sub.subscribe();
     return sub;
   };

diff --git a/src/subscription.js b/src/subscription.js
@@ -54,6 +54,11 @@ export default class Subscription extends EventEmitter {
     }).then(function () {}, function () {});
   };
 
+  _setNeedRecover(enabled) {
+    this._recoverable = enabled;
+    this._recover = enabled;
+  }
+
   _needRecover() {
     return this._recoverable === true && this._recover === true;
   };
@@ -203,11 +208,14 @@ export default class Subscription extends EventEmitter {
     }
   };
 
-  subscribe() {
+  subscribe(opts) {
     if (this._status === _STATE_SUCCESS) {
       return;
     }
     this._noResubscribe = false;
+    if (opts && opts.since) {
+      this._centrifuge._setSubscribeSince(this, opts.since);
+    }
     this._centrifuge._subscribe(this);
   };