feat: support topics sequence page

momentohq · Nov 6, 2024 · dbcfe9e · dbcfe9e
1 parent 44b5334
commit dbcfe9e
Show file tree

Hide file tree

Showing 8 changed files with 65 additions and 19 deletions.
diff --git a/packages/client-sdk-nodejs/package-lock.json b/packages/client-sdk-nodejs/package-lock.json
diff --git a/packages/client-sdk-nodejs/package.json b/packages/client-sdk-nodejs/package.json
@@ -58,7 +58,7 @@
     "uuid": "8.3.2"
   },
   "dependencies": {
-    "@gomomento/generated-types": "0.113.0",
+    "@gomomento/generated-types": "0.119.2",
     "@gomomento/sdk-core": "file:../core",
     "@grpc/grpc-js": "1.10.9",
     "@types/google-protobuf": "3.15.10",

diff --git a/packages/client-sdk-nodejs/src/internal/pubsub-client.ts b/packages/client-sdk-nodejs/src/internal/pubsub-client.ts
@@ -169,11 +169,13 @@ export class PubsubClient extends AbstractPubsubClient<ServiceError> {
       topic: options.topicName,
       resume_at_topic_sequence_number:
         options.subscriptionState.resumeAtTopicSequenceNumber,
+      sequence_page: options.subscriptionState.lastTopicSequencePage,
     });
 
     this.getLogger().trace(
-      'Subscribing to topic with resume_at_topic_sequence_number: %s',
-      options.subscriptionState.resumeAtTopicSequenceNumber
+      'Subscribing to topic with resume_at_topic_sequence_number %s and sequence_page %s',
+      options.subscriptionState.resumeAtTopicSequenceNumber,
+      options.subscriptionState.lastTopicSequencePage ?? 'undefined'
     );
     const call = this.client.Subscribe(request, {
       interceptors: this.streamingInterceptors,
@@ -210,11 +212,14 @@ export class PubsubClient extends AbstractPubsubClient<ServiceError> {
 
       if (resp.item) {
         const sequenceNumber = resp.item.topic_sequence_number;
+        const sequencePage = resp.item.sequence_page;
         options.subscriptionState.lastTopicSequenceNumber = sequenceNumber;
+        options.subscriptionState.lastTopicSequencePage = sequencePage;
         this.getLogger().trace(
-          'Received an item on subscription stream; topic: %s; sequence number: %s',
+          'Received an item on subscription stream; topic: %s; sequence number: %s; sequence page: %s',
           truncateString(options.topicName),
-          sequenceNumber
+          sequenceNumber,
+          sequencePage
         );
         if (resp.item.value.text) {
           options.onItem(
@@ -254,7 +259,8 @@ export class PubsubClient extends AbstractPubsubClient<ServiceError> {
         options.onDiscontinuity(
           new TopicDiscontinuity(
             resp.discontinuity.last_topic_sequence,
-            resp.discontinuity.new_topic_sequence
+            resp.discontinuity.new_topic_sequence,
+            resp.discontinuity.new_sequence_page
           )
         );
       } else {

diff --git a/packages/client-sdk-web/package-lock.json b/packages/client-sdk-web/package-lock.json
diff --git a/packages/client-sdk-web/package.json b/packages/client-sdk-web/package.json
@@ -63,7 +63,7 @@
     "xhr2": "0.2.1"
   },
   "dependencies": {
-    "@gomomento/generated-types-webtext": "0.113.0",
+    "@gomomento/generated-types-webtext": "0.119.2",
     "@gomomento/sdk-core": "file:../core",
     "@types/google-protobuf": "3.15.6",
     "google-protobuf": "3.21.2",

diff --git a/packages/client-sdk-web/src/internal/pubsub-client.ts b/packages/client-sdk-web/src/internal/pubsub-client.ts
@@ -134,6 +134,15 @@ export class PubsubClient<
     request.setResumeAtTopicSequenceNumber(
       options.subscriptionState.resumeAtTopicSequenceNumber
     );
+    if (options.subscriptionState.lastTopicSequencePage !== undefined) {
+      request.setSequencePage(options.subscriptionState.lastTopicSequencePage);
+    }
+
+    this.getLogger().trace(
+      'Subscribing to topic with resume_at_topic_sequence_number %s and sequence_page %s',
+      options.subscriptionState.resumeAtTopicSequenceNumber,
+      options.subscriptionState.lastTopicSequencePage ?? 'undefined'
+    );
 
     const call = this.client.subscribe(request, {
       ...this.clientMetadataProvider.createClientMetadata(),
@@ -179,10 +188,18 @@ export class PubsubClient<
 
       if (item) {
         const sequenceNumber = item.getTopicSequenceNumber();
+        const sequencePage = item.getSequencePage();
         options.subscriptionState.lastTopicSequenceNumber = sequenceNumber;
+        options.subscriptionState.lastTopicSequencePage = sequencePage;
         const publisherId = item.getPublisherId();
         const itemText = item.getValue()?.getText();
         const itemBinary = item.getValue()?.getBinary();
+        this.getLogger().trace(
+          'Received an item on subscription stream; topic: %s; sequence number: %s; sequence page: %s',
+          truncateString(options.topicName),
+          sequenceNumber,
+          sequencePage
+        );
         if (itemText) {
           options.onItem(
             new TopicItem(itemText, sequenceNumber, {tokenId: publisherId})
@@ -211,13 +228,16 @@ export class PubsubClient<
         options.onHeartbeat(new TopicHeartbeat());
       } else if (discontinuity) {
         this.getLogger().trace(
-          'Received discontinuity from subscription stream; topic: %s',
-          truncateString(options.topicName)
+          'Received a discontinuity; topic: %s; new sequence number: %s; new sequence page: %s',
+          truncateString(options.topicName),
+          discontinuity.getNewTopicSequence(),
+          discontinuity.getNewSequencePage()
         );
         options.onDiscontinuity(
           new TopicDiscontinuity(
             discontinuity.getLastTopicSequence(),
-            discontinuity.getNewTopicSequence()
+            discontinuity.getNewTopicSequence(),
+            discontinuity.getNewSequencePage()
           )
         );
       } else {

diff --git a/packages/core/src/internal/subscription-state.ts b/packages/core/src/internal/subscription-state.ts
@@ -4,6 +4,7 @@
 export class SubscriptionState {
   private _unsubscribeFn: () => void;
   public lastTopicSequenceNumber?: number;
+  public lastTopicSequencePage?: number;
   private _isSubscribed: boolean;
   constructor() {
     this._unsubscribeFn = () => {
@@ -43,6 +44,7 @@ export class SubscriptionState {
     return JSON.stringify(
       {
         lastTopicSequenceNumber: this.lastTopicSequenceNumber,
+        lastTopicSequencePage: this.lastTopicSequencePage,
         isSubscribed: this._isSubscribed,
       },
       null,

diff --git a/packages/core/src/messages/responses/topic-discontinuity.ts b/packages/core/src/messages/responses/topic-discontinuity.ts
@@ -6,10 +6,16 @@
 export class TopicDiscontinuity {
   private readonly _lastSequenceNumber: number;
   private readonly _newSequenceNumber: number;
+  private readonly _newSequencePage: number;
 
-  constructor(_lastSequenceNumber: number, _newSequenceNumber: number) {
+  constructor(
+    _lastSequenceNumber: number,
+    _newSequenceNumber: number,
+    _newSequencePage: number
+  ) {
     this._lastSequenceNumber = _lastSequenceNumber;
     this._newSequenceNumber = _newSequenceNumber;
+    this._newSequencePage = _newSequencePage;
   }
 
   /**
@@ -28,6 +34,14 @@ export class TopicDiscontinuity {
     return this._newSequenceNumber;
   }
 
+  /**
+   * Returns the new sequence page after the discontinuity.
+   * @returns number
+   */
+  public newSequencePage(): number {
+    return this._newSequencePage;
+  }
+
   public toString(): string {
     const displayValue = `Last Sequence Number: ${this._lastSequenceNumber}; New Sequence Number: ${this._newSequenceNumber}`;
     return `${this.constructor.name}: ${displayValue}`;