fix: pubsub UX minor fixes and refactor

src/index.ts

@@ -47,6 +47,7 @@ import * as CacheSortedSetRemoveElement from './messages/responses/cache-sorted-
 import * as CacheSortedSetRemoveElements from './messages/responses/cache-sorted-set-remove-elements';
 import * as TopicPublish from './messages/responses/topic-publish';
 import * as TopicSubscribe from './messages/responses/topic-subscribe';
+import {TopicItem} from './messages/responses/topic-item';
 
 import {CacheInfo} from './messages/cache-info';
 import {CollectionTtl} from './utils/collection-ttl';
@@ -147,6 +148,7 @@ export {
   CacheSortedSetRemoveElement,
   CacheSortedSetRemoveElements,
   TopicClient,
+  TopicItem,
   TopicPublish,
   TopicSubscribe,
   MomentoErrorCode,

src/internal/pubsub-client.ts

@@ -9,6 +9,7 @@ import {cacheServiceErrorMapper} from '../errors/cache-service-error-mapper';
 import {ChannelCredentials, Interceptor, ServiceError} from '@grpc/grpc-js';
 import {Status} from '@grpc/grpc-js/build/src/constants';
 import {
+  TopicItem,
   TopicPublish,
   TopicSubscribe,
   Configuration,
@@ -33,7 +34,7 @@ import {SubscriptionState} from './subscription-state';
 interface SendSubscribeOptions {
   cacheName: string;
   topicName: string;
-  onItem: (data: TopicSubscribe.Item) => void;
+  onItem: (item: TopicItem) => void;
   onError: (
     error: TopicSubscribe.Error,
     subscription: TopicSubscribe.Subscription
@@ -297,9 +298,9 @@ export class PubsubClient {
         options.subscriptionState.lastTopicSequenceNumber =
           resp.item.topic_sequence_number;
         if (resp.item.value.text) {
-          options.onItem(new TopicSubscribe.Item(resp.item.value.text));
+          options.onItem(new TopicItem(resp.item.value.text));
         } else if (resp.item.value.binary) {
-          options.onItem(new TopicSubscribe.Item(resp.item.value.binary));
+          options.onItem(new TopicItem(resp.item.value.binary));
         } else {
           this.logger.error(
             'Received subscription item with unknown type; topic: %s',

src/messages/responses/topic-item.ts

@@ -0,0 +1,43 @@
+import {truncateString} from '../../internal/utils/display';
+
+/**
+ * Represents the data received from a topic subscription.
+ *
+ * @remarks A subscription is created by calling {@link TopicClient.subscribe}.
+ * The value is guaranteed to be either a {@link string} or a {@link Uint8Array}.
+ * Call the appropriate accessor if you know the type of the value.
+ */
+export class TopicItem {
+  private readonly _value: string | Uint8Array;
+  constructor(_value: string | Uint8Array) {
+    this._value = _value;
+  }
+  /**
+   * Returns the data read from the stream.
+   * @returns string | Uint8Array
+   */
+  public value(): string | Uint8Array {
+    return this._value;
+  }
+
+  /**
+   * Returns the data read from the stream as a {@link string}.
+   * @returns string
+   */
+  public valueString(): string {
+    return this.value().toString();
+  }
+
+  /**
+   * Returns the data read from the stream as a {@link Uint8Array}.
+   * @returns Uint8Array
+   */
+  public valueUint8Array(): Uint8Array {
+    return this.value() as Uint8Array;
+  }
+
+  public toString(): string {
+    const display = truncateString(this.value().toString());
+    return `${this.constructor.name}: ${display}`;
+  }
+}

src/messages/responses/topic-subscribe.ts

@@ -1,7 +1,6 @@
 // older versions of node don't have the global util variables https://github.com/nodejs/node/issues/20365
 import {SdkError} from '../../errors/errors';
 import {ResponseBase, ResponseError} from './response-base';
-import {truncateString} from '../../internal/utils/display';
 import {SubscriptionState} from '../../internal/subscription-state';
 
 /**
@@ -26,42 +25,6 @@ import {SubscriptionState} from '../../internal/subscription-state';
  */
 export abstract class Response extends ResponseBase {}
 
-export class Item extends Response {
-  private readonly _value: string | Uint8Array;
-  constructor(_value: string | Uint8Array) {
-    super();
-    this._value = _value;
-  }
-  /**
-   * Returns the data read from the stream.
-   * @returns string | Uint8Array
-   */
-  public value(): string | Uint8Array {
-    return this._value;
-  }
-
-  /**
-   * Returns the data read from the stream as a string.
-   * @returns string
-   */
-  public valueString(): string {
-    return this.value().toString();
-  }
-
-  /**
-   * Returns the data read from the stream as a Uint8Array.
-   * @returns Uint8Array
-   */
-  public valueUint8Array(): Uint8Array {
-    return this.value() as Uint8Array;
-  }
-
-  public override toString(): string {
-    const display = truncateString(this.value().toString());
-    return `${super.toString()}: ${display}`;
-  }
-}
-
 /**
  * Encapsulates a topic subscription.
  *

src/topic-client.ts

@@ -55,7 +55,7 @@ export class TopicClient {
   public async subscribe(
     cacheName: string,
     topicName: string,
-    options: SubscribeCallOptions = {}
+    options: SubscribeCallOptions
   ): Promise<TopicSubscribe.Response> {
     return await this.client.subscribe(cacheName, topicName, options);
   }

src/utils/topic-call-options.ts

@@ -1,4 +1,4 @@
-import {TopicSubscribe} from '..';
+import {TopicItem, TopicSubscribe} from '..';
 
 /**
  * Options for the subscribe call.
@@ -7,9 +7,9 @@ export interface SubscribeCallOptions {
   /**
    * The callback to invoke when data is received from the topic subscription.
    *
-   * @param data The data received from the topic subscription.
+   * @param item The data received from the topic subscription.
    */
-  onItem?: (data: TopicSubscribe.Item) => void;
+  onItem?: (item: TopicItem) => void;
 
   /**
    * The callback to invoke when an error is received from the topic subscription.

test/integration/topic-client.test.ts

@@ -4,6 +4,7 @@ import {
   DeleteCache,
   TopicClient,
   MomentoErrorCode,
+  TopicItem,
   TopicPublish,
   TopicSubscribe,
   InvalidArgumentError,
@@ -117,7 +118,11 @@ describe('#subscribe', () => {
   });
 
   it('should error when subscribing to a cache that does not exist and unsubscribe from the error handler', async () => {
-    const response = await topicClient.subscribe(v4(), 'topic');
+    const response = await topicClient.subscribe(
+      v4(),
+      'topic',
+      trivialHandlers
+    );
     expect((response as IResponseError).errorCode()).toEqual(
       MomentoErrorCode.NOT_FOUND_ERROR
     );
@@ -139,7 +144,7 @@ describe('subscribe and publish', () => {
       IntegrationTestCacheName,
       topicName,
       {
-        onItem: (item: TopicSubscribe.Item) => {
+        onItem: (item: TopicItem) => {
           receivedValues.push(item.value());
         },
         onError: (error: TopicSubscribe.Error) => {
@@ -185,7 +190,7 @@ describe('subscribe and publish', () => {
       IntegrationTestCacheName,
       topicName,
       {
-        onItem: (item: TopicSubscribe.Item) => {
+        onItem: (item: TopicItem) => {
           receivedValues.push(item.value());
         },
         onError: (error: TopicSubscribe.Error) => {
@@ -226,7 +231,12 @@ describe('subscribe and publish', () => {
 
     const subscribeResponse = await topicClient.subscribe(
       IntegrationTestCacheName,
-      topicName
+      topicName,
+      {
+        onItem: () => {
+          return;
+        },
+      }
     );
     expect(subscribeResponse).toBeInstanceOf(TopicSubscribe.Subscription);
     (subscribeResponse as TopicSubscribe.Subscription).unsubscribe();
@@ -244,7 +254,7 @@ describe('subscribe and publish', () => {
         randomCacheName,
         v4(),
         {
-          onItem: (item: TopicSubscribe.Item) => {
+          onItem: (item: TopicItem) => {
             expect(1).fail(
               `Should not receive an item but got one: ${item.toString()}`
             );