Add timeout to jetstream (#494)

Author: Varagos

boilerplate/ts/infra/nest-jetstream/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@bitloops/bl-boilerplate-infra-nest-jetstream",
-  "version": "0.0.6",
+  "version": "0.0.7",
   "description": "TypeScript nats jetstream code for Bitloops Language generated projects using NestJS",
   "engines": {
     "node": ">= 13"

boilerplate/ts/infra/nest-jetstream/src/buses/nats-pubsub-command-bus.ts

@@ -1,7 +1,7 @@
 import { Inject, Injectable, Logger } from '@nestjs/common';
 import { NatsConnection, JSONCodec, headers, Msg, MsgHdrs } from 'nats';
 import { Application, Infra } from '@bitloops/bl-boilerplate-core';
-import { ASYNC_LOCAL_STORAGE, ProvidersConstants } from '../jetstream.constants';
+import { ASYNC_LOCAL_STORAGE, TIMEOUT_MILLIS, ProvidersConstants } from '../jetstream.constants';
 import { ContextPropagation } from './utils/context-propagation';
 
 const jsonCodec = JSONCodec();
@@ -16,6 +16,8 @@ export class NatsPubSubCommandBus implements Infra.CommandBus.IPubSubCommandBus
     @Inject(ProvidersConstants.JETSTREAM_PROVIDER) private readonly nats: any,
     @Inject(ASYNC_LOCAL_STORAGE)
     private readonly asyncLocalStorage: any,
+    @Inject(TIMEOUT_MILLIS)
+    private readonly timeoutMillis: number,
   ) {
     this.nc = this.nats.getConnection();
   }
@@ -42,8 +44,8 @@ export class NatsPubSubCommandBus implements Infra.CommandBus.IPubSubCommandBus
   async request(command: Application.Command): Promise<any> {
     const topic = NatsPubSubCommandBus.getTopicFromCommandInstance(command);
 
-    this.logger.log('Requesting in haha :' + topic);
-    console.log('this.asyncLocalStorage.getStore()', this.asyncLocalStorage.getStore());
+    this.logger.log('Requesting in topic:' + topic);
+    // console.log('this.asyncLocalStorage.getStore()', this.asyncLocalStorage.getStore());
     command.correlationId = this.getCorelationId();
     command.context = this.getContext();
     // command
@@ -52,7 +54,7 @@ export class NatsPubSubCommandBus implements Infra.CommandBus.IPubSubCommandBus
     try {
       const response = await this.nc.request(topic, jsonCodec.encode(command), {
         headers,
-        timeout: 10000,
+        timeout: this.timeoutMillis,
       });
       return jsonCodec.decode(response.data);
     } catch (error: any) {

boilerplate/ts/infra/nest-jetstream/src/jetstream-core.module.ts

@@ -9,7 +9,12 @@ import { JetstreamModuleFeatureConfig } from './interfaces/module-feature-input.
 import { BUSES_TOKENS } from './buses/constants';
 import { NatsStreamingIntegrationEventBus, NatsStreamingCommandBus } from './buses';
 import { SubscriptionsService } from './jetstream.subscriptions.service';
-import { ASYNC_LOCAL_STORAGE, HANDLERS_TOKENS, ProvidersConstants } from './jetstream.constants';
+import {
+  ASYNC_LOCAL_STORAGE,
+  HANDLERS_TOKENS,
+  ProvidersConstants,
+  TIMEOUT_MILLIS,
+} from './jetstream.constants';
 import { NatsPubSubIntegrationEventsBus } from './buses/nats-pubsub-integration-events-bus';
 
 const pubSubCommandBus = {
@@ -48,7 +53,7 @@ const pubSubIntegrationEventBus = {
 @Global()
 @Module({})
 export class JetstreamCoreModule {
-  static forRoot(connectionOptions: ConnectionOptions): DynamicModule {
+  static forRoot(connectionOptions: ConnectionOptions, timeoutMillis?: number): DynamicModule {
     const jetstreamProviders = {
       provide: ProvidersConstants.JETSTREAM_PROVIDER,
       useFactory: (): any => {
@@ -61,6 +66,11 @@ export class JetstreamCoreModule {
       useValue: asyncLocalStorage,
     };
 
+    const timeoutMillisProvider = {
+      provide: TIMEOUT_MILLIS,
+      useValue: timeoutMillis ?? 10000,
+    };
+
     return {
       module: JetstreamCoreModule,
       providers: [
@@ -73,6 +83,7 @@ export class JetstreamCoreModule {
         streamingMessageBus,
         pubSubIntegrationEventBus,
         asyncLocalStorageProvider,
+        timeoutMillisProvider,
       ],
       exports: [
         jetstreamProviders,
@@ -84,6 +95,7 @@ export class JetstreamCoreModule {
         streamingMessageBus,
         pubSubIntegrationEventBus,
         asyncLocalStorageProvider,
+        timeoutMillisProvider,
       ],
     };
   }

boilerplate/ts/infra/nest-jetstream/src/jetstream.constants.ts

@@ -11,6 +11,7 @@ export const HANDLERS_TOKENS = {
 };
 
 export const ASYNC_LOCAL_STORAGE = Symbol('ASYNC_LOCAL_STORAGE');
+export const TIMEOUT_MILLIS = Symbol('TIMEOUT_MILLIS');
 
 export const METADATA_HEADERS = {
   CORRELATION_ID: 'correlationId',

boilerplate/ts/infra/nest-jetstream/src/jetstream.module.ts

@@ -6,10 +6,10 @@ import { JetstreamCoreModule } from './jetstream-core.module';
 
 @Module({})
 export class JetstreamModule {
-  static forRoot(connectionOptions: ConnectionOptions): DynamicModule {
+  static forRoot(connectionOptions: ConnectionOptions, timeoutMillis?: number): DynamicModule {
     return {
       module: JetstreamModule,
-      imports: [JetstreamCoreModule.forRoot(connectionOptions)],
+      imports: [JetstreamCoreModule.forRoot(connectionOptions, timeoutMillis)],
     };
   }