redningsselskapet / nestjs-plugins

Plugins for NestJS framework

nestjs-plugins

• nestjs-nats-jetstream-transport

nestjs-plugins's Issues

nestjs-nats-jetstream-transport: creates a new connection for every publish

Creates a new connection for every publish. Connection should be reused.

Publish message is sent to all in queue group

When publishing a message, every consumer in a queue group receives the message. Only one should receive and create response.

NatsStreamConfig is not exposed, prevents stream configuration

When creating a NatsJetsStreamServer, the streamConfig element does not expose NatsStreamConfig. In general the NatsStreamConfig type cannot be imported.

Found in Version: 1.3.8

Steps to repeat

1. Setup project using the supplied example from the Readme.md
2. import { NatsStreamConfig } from '@nestjs-plugins/nestjs-nats-jetstream-transport'
3. run yarn build

yarn build
yarn run v1.22.19
warning ../../package.json: No license field
$ rimraf dist
$ nest build
src/main.ts:7:10 - error TS2305: Module '"@nestjs-plugins/nestjs-nats-jetstream-transport"' has no exported member 'NatsStreamConfig'.

7 import { NatsStreamConfig } from '@nestjs-plugins/nestjs-nats-jetstream-transport'
           ~~~~~~~~~~~~~~~~

Found 1 error(s).

error Command failed with exit code 1.

Workaround:

Copy the code out of the module. Eg: Include the following:

mport { StreamConfig, RetentionPolicy } from 'nats';

type $StreamConfig = Pick<
  StreamConfig,
  | 'storage'
  | 'retention'
  | 'discard'
  | 'max_msgs'
  | 'max_msgs_per_subject'
  | 'max_msg_size'
  | 'max_age'
  | 'duplicate_window'
  | 'num_replicas'
>;

export interface NatsStreamConfig extends Partial<$StreamConfig> {
  name: string;
  subjects: string[];
}

How to log connection errors and reconnects?

I think I need to listen to connection events as described here. I looked at the code, but could not figure it out. Any suggestions?

Message deduplication

Is your feature request related to a problem? Please describe.
Nats allows deduplication of messages. This is done by setting the message header id to a specific unique value per message.
I haven't found an API that allows me to pass the message header, so I can set the message id or the client header if I want.

Describe the solution you'd like
It would be nice to be able to pass header object through the emit function.
some thing like that:

await this.jetSteamClient.emit<{ uuid: string, sentAt: string }>('registration.password.reset.email.sent', {
                uuid,
                sentAt: now,
            }, {
              msgID: uuid,
              "other-custom-heaser": value,
           });

Additional context
A use case described by NATS: https://nats.io/blog/nats-jetstream-deduplication-for-lfh/

Can't mix NATS JetStream with classic NATS server

Describe the bug
When mixing Jetstream with nats-server we got this error:

Error: no stream matches subject
    at JetStreamClientImpl.<anonymous> (C:\Users\mjarrai\Documents\ecomrepo\i.sale.authentication.service\node_modules\nats\nats-base-client\jsbaseclient_api.ts:97:13)
    at Generator.next (<anonymous>)
    at fulfilled (C:\Users\mjarrai\Documents\ecomrepo\i.sale.authentication.service\node_modules\nats\lib\nats-base-client\jsbaseclient_api.js:19:58)
    at processTicksAndRejections (node:internal/process/task_queues:95:5)

To Reproduce
In the same microservice mix nats-server with nats-jetstream

• In main.ts

 ...
  app.connectMicroservice<NatsOptions>({
    transport: Transport.NATS,
    options: {
      servers: [configService.get('nats.url')],
      name: `${configService.get('nats.name')}-subscriber`,
      queue: 'lb_queue1',
      reconnect: configService.get('nats.reconnect'),
      maxReconnectAttempts: configService.get('nats.maxReconnectAttempts'),
    },
  });

  app.connectMicroservice<CustomStrategy>({

    strategy: new NatsJetStreamServer({
      connectionOptions: {
        servers: [configService.get('nats.url')],
        name: `${configService.get('nats.name')}-durable-subscriber`,
        // name: `myservice-listener`,
        debug: true,
      },

      consumerOptions: {
        durable: 'myservice-durable', // Consumer name
        deliverTo: 'myservice-messages', // Delivery target -> push
        deliverGroup: 'myservice-group', // Delivery Queue Group
        manualAck: true,
      },

      streamConfig: {
        name: 'mystream',
        subjects: ['order.*'],
      },

    }),
  });

  await app.startAllMicroservices();
  await app.listen(servicePort, serviceIp);

• In controller

    ...
    @EventPattern('BULK_VALIDATION_MAIL_REQUESTED') // <- event from nats-server
    async processValidationMailRequestedBulk(@Payload() arr: { email: string, [key: string]: string }[]) {
        await this.registrationService.updateBulk(arr, State.VALIDATION_MAIL_REQUESTED);
    }

    @EventPattern('order.updated') // <- event from nats-jetstream
    public async orderUpdatedHandler(
      @Payload() data: string,
      @Ctx() context: NatsJetStreamContext,
    ) {
        console.log('received: ' + context.message.subject, data); 
        context.message.ack();
    }

After enabling connection debugging, I found that events from the nats server are being picked up by the jetstream event handler! In my case, the library tries to treat the "BULK_VALIDATION_MAIL_REQUESTED" event as a jetStream event.

Please check the log:

< PUB $JS.API.STREAM.NAMES _INBOX.69LFNR8K3IS2RS0AH4JYVL.69LFNR8K3IS2RS0AH4JZE2 44␍␊{"subject":"BULK_VALIDATION_MAIL_REQUESTED"}␍␊
> MSG _INBOX.69LFNR8K3IS2RS0AH4JYVL.69LFNR8K3IS2RS0AH4JZE2 1 106␍␊{"type":"io.nats.jetstream.api.v1.stream_names_response","total":0,"offset":0,"limit":1024,"streams":null}␍␊
Error: no stream matches subject
    at JetStreamClientImpl.<anonymous> (C:\Users\mjarrai\Documents\ecomrepo\i.sale.authentication.service\node_modules\nats\nats-base-client\jsbaseclient_api.ts:97:13)
    at Generator.next (<anonymous>)
    at fulfilled (C:\Users\mjarrai\Documents\ecomrepo\i.sale.authentication.service\node_modules\nats\lib\nats-base-client\jsbaseclient_api.js:19:58)
    at processTicksAndRejections (node:internal/process/task_queues:95:5) 

Expected behavior
Find a way to differentiate between nats-server and nats-jetstream event handlers.

Sending message without consumer - 503 Service not available not handled

NatsError: 503
at Function.errorForCode (/Users/bernt/Coding/rs-dev/nestjs-plugins/packages/nats-publisher/node_modules/nats/nats-base-client/error.ts

EventHandler cleanup uses undefine this.nc

Describe the bug
Seeing this in my jests:

TypeError: Cannot read properties of undefined (reading 'isDraining')
  at /app/node_modules/@nestjs-plugins/nestjs-nats-jetstream-transport/src/server.ts:90:21
  at processTicksAndRejections (node:internal/process/task_queues:95:5)

It seems the cleanup code:

nestjs-plugins/packages/nestjs-nats-jetstream-transport/src/server.ts

Lines 88 to 91 in 71cc12a

	done.then(() => {
	// if the connection is closed or draining, we don't need to unsubscribe as the subscription will be unsubscribed automatically
	if (this.nc.isDraining() || this.nc.isClosed()) {
	return;

is running after the server has already reset nc after completing a drain/close:

nestjs-plugins/packages/nestjs-nats-jetstream-transport/src/server.ts

Lines 48 to 51 in 71cc12a

	async close() {
	await this.nc.drain();
	this.nc = undefined;
	}

Seems like removing this would fix it:

this.nc = undefined;

To Reproduce
Sporadic failure after jests run.

Expected behavior
No error.

Cannot see effect of manualAck

Describe the bug
Cannot see effect of manualAck.

To Reproduce
As specified in the docs i am using the below configuration:

strategy: NatsJetStreamServer.

consumerOptions: {
        ackPolicy: 'All',
        deliverGroup: 'hh-identity-group',
        /*durable: process.env.IDENTITY_QUEUE_GROUP_NAME,*/
        deliverTo: `${process.env.IDENTITY_QUEUE_GROUP_NAME}`,
        **manualAck: true,**
        deliverPolicy: 'New',
        ackWait: 10,
      },

In the code of the consumer (@EventPattern)

await context.message?.ack()

If I comment out the above code, in the consumer, still messages are getting processed and getting auto acked.
The consumer code is NOT returning anything.

Seems there is no effect of manualAck.

Expected behavior

Should not acknowledge the messages when manulAck is set to true and message is not acknowledged.

Server:

• OS: Ubuntu Linux
• "@nestjs/core": "^9.1.4", "@nestjs/microservices": "^9.1.4",
• "@nestjs-plugins/nestjs-nats-jetstream-transport": "^1.3.11",

Reason for some events not being received?

Environment

OS: Ubuntu 22.04
Node: 16.13v

Problem

I would send multiple events, but some events don't get received through handler function. I'm expecting to see every event and message to get logged, but that's not the case.

Code

app.controller.ts

import { NatsJetStreamContext } from '@nestjs-plugins/nestjs-nats-jetstream-transport';
import { Controller, Get } from '@nestjs/common';
import { Ctx, EventPattern, Payload } from '@nestjs/microservices';
import { AppService } from './app.service';

@Controller()
export class AppController {
  constructor(private readonly appService: AppService) {}

  @Get('create')
  createOrder(): string {
    return this.appService.createdOrder();
  }

  @EventPattern('order.created')
  async orderCreatedHandler(
    @Payload() data: { id: number; name: string },
    @Ctx() context: NatsJetStreamContext,
  ) {
    console.log('-----------------------------');
    console.log(context);
    context.message.ack();
    console.log(context);
    console.log(context.message.seq);
    console.log('recieved: ' + context.message.subject, data);
  }
}

app.service.ts

import { NatsJetStreamClient } from '@nestjs-plugins/nestjs-nats-jetstream-transport';
import { Injectable } from '@nestjs/common';

interface OrderCreatedEvent {
  id: number;
  product: string;
  quantity: number;
}

const ORDER_CREATED = 'order.created';

@Injectable()
export class AppService {
  constructor(private client: NatsJetStreamClient) {}

  createdOrder(): string {
    this.client
      .emit<OrderCreatedEvent>(ORDER_CREATED, {
        id: 1,
        product: 'Socks',
        quantity: 1,
      })
      .subscribe((pubAck) => {
        console.log('pubAck', pubAck);
      });

    return 'order created';
  }
}

main.ts

import { NatsJetStreamServer } from '@nestjs-plugins/nestjs-nats-jetstream-transport';
import { NestFactory } from '@nestjs/core';
import { CustomStrategy } from '@nestjs/microservices';
import { AppModule } from './app.module';

async function bootstrap() {
  const options: CustomStrategy = {
    strategy: new NatsJetStreamServer({
      connectionOptions: {
        servers: 'localhost:4222',
        name: 'myservice-listener',
      },
      consumerOptions: {
        deliverGroup: 'myservice-group',
        durable: 'myservice-durable',
        deliverTo: 'myservice-messages',
        manualAck: true,
      },
      streamConfig: {
        name: 'mystream',
        subjects: ['order.*'],
      },
    }),
  };
  const app = await NestFactory.create(AppModule);

  const microservice = app.connectMicroservice(options);
  microservice.listen();
  await app.listen(3000);
}
bootstrap();

Output

Discussion

Any advice or thoughts on why this happens?

How i can check ack?

Hi! A lot of thanks for youre work!
I have issuse:

this code

this.natsClient
  .emit<DataResponse>('entity.created', {
  success: true,
  dataResponse,
  message: 'Ok',
  })
  .subscribe((pubAck: PubAck) => {
  console.log('pubAck', pubAck)
})

return everetime message like this
pubAck { stream: 'mystream', seq: 22, duplicate: false }

even though i didn't ack this and the same if i did
I dont understan how i can check ack

my config

		connectionOptions: {
				servers: 'nats://0.0.0.0:4222',
				name: 'listener',
				timeout: 2000,
			},
			consumerOptions: {
				deliverGroup: 'group',
				deliverTo: 'group',
				manualAck: true,
				ackPolicy: 'None',
				deliverPolicy: 'Last',
			},
			streamConfig: {
				name: 'mystream',
				subjects: ['entity.*'],
			},

I stuck in this

streams.update(config: StreamConfig) api changed ... update your code.

Changes in the Nats API:

Trace: >> streams.update(config: StreamConfig) api changed to streams.update(name: string, config: StreamUpdateConfig) - this shim will be removed - update your code.
at StreamAPIImpl. (/Users/raarts/work/company/group/product/node_modules/nats/nats-base-client/jsmstream_api.ts:73:15)
at Generator.next ()
at /Users/raarts/work/company/group/product/node_modules/nats/lib/nats-base-client/jsmstream_api.js:22:71
at new Promise ()
at __awaiter (/Users/raarts/work/company/group/product/node_modules/nats/lib/nats-base-client/jsmstream_api.js:18:12)
at StreamAPIImpl.update (/Users/raarts/work/company/group/product/node_modules/nats/lib/nats-base-client/jsmstream_api.js:54:16)
at NatsJetStreamServer.setupStream (/Users/raarts/work/company/group/product/node_modules/@nestjs-plugins/nestjs-nats-jetstream-transport/src/server.ts:121:49)
at processTicksAndRejections (node:internal/process/task_queues:96:5)
at NatsJetStreamServer.listen (/Users/raarts/work/company/group/product/node_modules/@nestjs-plugins/nestjs-nats-jetstream-transport/src/server.ts:36:7)

nats-io/nats.js#488

Cannot read properties of undefined (reading 'replaceAll')

TypeError: Cannot read properties of undefined (reading 'replaceAll')
at serverConsumerOptionsBuilder (/node_modules/@nestjs-plugins/nestjs-nats-jetstream-transport/src/utils/server-consumer-options-builder.ts:53:29)

I think there is some issue with this

Also let me know if you think I have made an error from my side.

request-response hanging on response

Using the NatsJetStreamClientProxy requests are sent correctly but the response is never received by the client.

To Reproduce
Steps to reproduce the behavior:
Setup the example as per the homepage https://www.npmjs.com/package/@nestjs-plugins/nestjs-nats-jetstream-transport.
However, the first issue is that the NatsJetStreamClient is not provided as part of the register in the module so swapped the NatsJetStreamClient for the NatsJetStreamClientProxy which is provided.

Expected behavior
Calling /sum on the service should result in the return value of 6, however, the client hangs.
The console logs from the REST controller and the microservice controller are printed (as per below), but the client never returns.

% nest start [Nest] 2601 - 04/04/2023, 14:44:01 LOG [NestFactory] Starting Nest application... [Nest] 2601 - 04/04/2023, 14:44:01 LOG [InstanceLoader] NatsJetStreamTransport dependencies initialized +6ms [Nest] 2601 - 04/04/2023, 14:44:01 LOG [InstanceLoader] AppModule dependencies initialized +1ms [Nest] 2601 - 04/04/2023, 14:44:01 LOG [RoutesResolver] AppController {/}: +10ms [Nest] 2601 - 04/04/2023, 14:44:01 LOG [RouterExplorer] Mapped {/, GET} route +1ms [Nest] 2601 - 04/04/2023, 14:44:01 LOG [RouterExplorer] Mapped {/create, GET} route +0ms [Nest] 2601 - 04/04/2023, 14:44:01 LOG [RouterExplorer] Mapped {/update, GET} route +0ms [Nest] 2601 - 04/04/2023, 14:44:01 LOG [RouterExplorer] Mapped {/delete, GET} route +0ms [Nest] 2601 - 04/04/2023, 14:44:01 LOG [RouterExplorer] Mapped {/sum, GET} route +0ms [Nest] 2601 - 04/04/2023, 14:44:01 LOG [NestApplication] Nest application successfully started +1ms [Nest] 2601 - 04/04/2023, 14:44:06 LOG [Server] Stream mystream updated [Nest] 2601 - 04/04/2023, 14:44:06 LOG [Server] Subscribed to order.updated events [Nest] 2601 - 04/04/2023, 14:44:06 LOG [Server] Subscribed to order.created events [Nest] 2601 - 04/04/2023, 14:44:06 LOG [Server] Subscribed to order.deleted events [Nest] 2601 - 04/04/2023, 14:44:06 LOG [Server] Subscribed to {"cmd":"sum"} messages [Nest] 2601 - 04/04/2023, 14:44:06 LOG [NestMicroservice] Nest microservice successfully started +0ms sum controller message conroller [ 1, 2, 3 ]

nats 2.9.15
node 18.13.0

"@nestjs-plugins/nestjs-nats-jetstream-transport": "^1.3.14",
"@nestjs/common": "^9.0.0",
"@nestjs/core": "^9.0.0",
"@nestjs/microservices": "^9.3.12",
"@nestjs/platform-express": "^9.0.0",
"nats": "^2.13.1",
"reflect-metadata": "^0.1.13",
"rxjs": "^7.2.0"

Support for NATS Websockets?

Is your feature request related to a problem? Please describe.
The current implementation allows us to easily connect to Nats Jetstream, but currently unable to connect Nats Jetstream with a web socket connection (e.g. ws://localhost:8888).

Describe the solution you'd like
It would be nice to be able to connect to Nats Jetstream with web socket support.

async function bootstrap() {
  const options: CustomStrategy = {
    strategy: new NatsJetStreamServer({
      connectionOptions: {
        servers: 'ws://localhost:8888',
        name: 'artifact-listener',
      },
      consumerOptions: {
        deliverGroup: 'artifact-group',
        durable: 'artifact-durable',
        deliverTo: 'artifact-messages',
        manualAck: true,
      },
      streamConfig: {
        name: 'BUILD',
        subjects: ['task.*', 'artifact.*'],
      },
    }),
  };

  const app = await NestFactory.create(AppModule);

  const microservice = app.connectMicroservice(options);
  microservice.listen();
  await app.listen(4000);
}
bootstrap();

NatsError: CONNECTION_DRAINING at application shutdown

Hello :)

I got uncatchable NatsError: CONNECTION_DRAINING exception when shutting down app with app.close()
I need it at my tests to cleanup.

To Reproduce
Steps to reproduce the behavior:
Add to packages/nats-listener/src/main.ts at bootstrap end after await microService.listen();:

  setTimeout(async() => {
    logger.log("closing app");
    try {
      await app.close();
    } catch (err) {
      logger.error(err);
    }
    logger.log("closed");
  }, 10*1000);

Expected behavior
App closes, logging "closed";

Screenshots

Looks like app tries to delete consumer via $JS.API.CONSUMER.DELETE .... after drain is called, can you plz take a look?

Thanks

Events are consumed by wrong consumer.

It seems like all the different events are load balance across all consumers. This is not correct behavior. Consumer should only receive events it subscribes.

Connects fails with Error: Maximum call stack size exceeded

Calling connect from NatsJetStreamClient couses error Maximum call stack size exceeded

RangeError: Maximum call stack size exceeded
at NatsJetStreamClient.connect (/Users/bernt/Coding/rs-dev/mylab/aistracker/packages/kystverket-aisdata-import/node_modules/@nestjs-plugins/nestjs-nats-jetstream-transport/src/nats-jetstream-client.ts:16:10)
at NatsJetStreamClient.connect (/Users/bernt/Coding/rs-dev/mylab/aistracker/packages/kystverket-aisdata-import/node_modules/@nestjs-plugins/nestjs-nats-jetstream-transport/src/nats-jetstream-client.ts:16:10)
at NatsJetStreamClient.connect (/Users/bernt/Coding/rs-dev/mylab/aistracker/packages/kystverket-aisdata-import/node_modules/@nestjs-plugins/nestjs-nats-jetstream-transport/src/nats-jetstream-client.ts:16:10)
at NatsJetStreamClient.connect (/Users/bernt/Coding/rs-dev/mylab/aistracker/packages/kystverket-aisdata-import/node_modules/@nestjs-plugins/nestjs-nats-jetstream-transport/src/nats-jetstream-client.ts:16:10)
at NatsJetStreamClient.connect (/Users/bernt/Coding/rs-dev/mylab/aistracker/packages/kystverket-aisdata-import/node_modules/@nestjs-plugins/nestjs-nats-jetstream-transport/src/nats-jetstream-client.ts:16:10)
at NatsJetStreamClient.connect (/Users/bernt/Coding/rs-dev/mylab/aistracker/packages/kystverket-aisdata-import/node_modules/@nestjs-plugins/nestjs-nats-jetstream-transport/src/nats-jetstream-client.ts:16:10)
at NatsJetStreamClient.connect (/Users/bernt/Coding/rs-dev/mylab/aistracker/packages/kystverket-aisdata-import/node_modules/@nestjs-plugins/nestjs-nats-jetstream-transport/src/nats-jetstream-client.ts:16:10)
at NatsJetStreamClient.connect (/Users/bernt/Coding/rs-dev/mylab/aistracker/packages/kystverket-aisdata-import/node_modules/@nestjs-plugins/nestjs-nats-jetstream-transport/src/nats-jetstream-client.ts:16:10)
at NatsJetStreamClient.connect (/Users/bernt/Coding/rs-dev/mylab/aistracker/packages/kystverket-aisdata-import/node_modules/@nestjs-plugins/nestjs-nats-jetstream-transport/src/nats-jetstream-client.ts:16:10)
at NatsJetStreamClient.connect (/Users/bernt/Coding/rs-dev/mylab/aistracker/packages/kystverket-aisdata-import/node_modules/@nestjs-plugins/nestjs-nats-jetstream-transport/src/nats-jetstream-client.ts:16:10)

Unable to create pull-based consumer

Hi! first of all, thanks for creating this great library.

I tried to create pull-based consumer using it but it seems that it only works with the push-based one. I'd like to confirm whether you're aware of this and going to support this.

Thanks!

Allow overriding global `consumerOptions` on specific Message/Event patterns

Hey there! Thank you for this amazing plugin!

Is your feature request related to a problem? Please describe.
We want to create multiple NATS Jetstream consumer callbacks (with the @EventPattern() decorator) with different replay and ack strategies.
It would be super useful to be able to override the global consumerOptions on specific Message/Event patterns.

Describe the solution you'd like
The decorators MessagePattern and EventPattern support the extra?: Record<string, any> property (see https://github.com/nestjs/nest/blob/master/packages/microservices/server/server.ts#L51) that can be used for this kind a needs.

IMO, a simple solution could be to use the extra property if available in the handler callback object and merge its content with the base consumerOptions in the server config (see https://github.com/Redningsselskapet/nestjs-plugins/blob/master/packages/nestjs-nats-jetstream-transport/src/server.ts#L65).

It could be something like this:

const options = !eventHandler.extra ? consumerOptions : Object.assign({}, consumerOptions, eventHandler.extra);
const subscription = await js.subscribe(subject, options);

What do you think?

Describe alternatives you've considered
After looking at the plugin code, I cannot make this work without updating the plugin code.

I can submit a PR if you like, just tell me if you have any recommandation about the solution I gave.

Property 'replaceAll' does not exist on type 'string'.

Got this error after merging #29

TypeError: subject.replaceAll is not a function

main.ts

async function bootstrap() {
  const options: CustomStrategy = {
    strategy: new NatsJetStreamServer({
      connectionOptions: {
        servers: 'localhost:4222',
        name: 'some-name',
      },
      consumerOptions: {
        deliverGroup: 'myservice-group',
        durable: 'mydurable-service',
        deliverTo: 'myservice-messages',
        manualAck: true,
      },
    }),
  };
  const app = await NestFactory.createMicroservice<MicroserviceOptions>(
    AppModule,
    options,
  );

  await app.listen();
}

tsconfig.json

{
  "compilerOptions": {
    "module": "commonjs",
    "declaration": true,
    "removeComments": true,
    "emitDecoratorMetadata": true,
    "experimentalDecorators": true,
    "allowSyntheticDefaultImports": true,
    "target": "es2021",
    "sourceMap": true,
    "outDir": "./dist",
    "baseUrl": "./",
    "incremental": true,
    "skipLibCheck": true,
    "strictNullChecks": false,
    "noImplicitAny": false,
    "strictBindCallApply": false,
    "forceConsistentCasingInFileNames": false,
    "noFallthroughCasesInSwitch": false,
  
  }
}

Multiple connections to separate NATS instances?

So in my case I want to connect my app to 2 different NATS JetStream servers and I register 2 different connections. However I don't see how to utilize them in the service and I can't seem to find any documentation on the matter

import { Module } from '@nestjs/common';
import { AppController } from './app.controller';
import { AppService } from './app.service';
import { NatsJetStreamTransport } from '@nestjs-plugins/nestjs-nats-jetstream-transport';

@Module({
  imports: [
    NatsJetStreamTransport.register({
      connectionOptions: {
        servers: 'localhost:4222',
        name: 'myservice-publisher',
      },
    }),
    NatsJetStreamTransport.register({
      connectionOptions: {
        servers: 'localhost:6222',
        name: 'myservice-publisher-2',
      },
    }),
  ],
  controllers: [AppController],
  providers: [AppService],
})
export class AppModule {}

I would imagine something like this?

constructor(
    private client: NatsJetStreamClientProxy,
    private client_2: NatsJetStreamClientProxy
) {}

with some sort of Inject to specify the connection?

Consumer option ackWait ignored

Describe the bug
Consumer option ackWait ignored.

Option ackWait is not set by the serverConsumerOptionsBuilder

nestjs-plugins/packages/nestjs-nats-jetstream-transport/src/utils/server-consumer-options-builder.ts

Line 34 in f3933fa

const opts = consumerOpts();

subject.replaceAll is not a function

Hello. I found a bug, it’s not clear how to fix it. I went into the source code and corrected it, everything worked. The problem is in the line with replaceAll. It gave an error that subject.replaceAll is not a function

solution

Wrong typing for emit().subscribe()

According to the typings the function passed to subscribe() receives a string parameter but in fact it's an object like this one:

pubAck = {
  stream: 'mystream',
  seq: 1,
  duplicate: false
};

Consumer event handlers omitted when using inheritAppConfig and ClassSerializerInterceptor

Describe the bug
No event handler is trigger when using inheritAppConfig and ClassSerializerInterceptor

To Reproduce
Steps to reproduce the behavior:

1. Start the microservice application with inheritAppConfig: true.
2. Set Serializer as Global Interceptor with app.useGlobalInterceptors(new ClassSerializerInterceptor(app.get(Reflector)))
3. Emit an event order.updated
4. See that function with decorator @EventHandler('order.updated') is omitted
5. Comment out Serializer as Global Interceptor with // app.useGlobalInterceptors(new ClassSerializerInterceptor(app.get(Reflector)))
6. Emit an event order.updated
7. See that function with decorator @EventHandler('order.updated') is triggered

Expected behavior
All functions with decorator @EventHandler() are triggered