nestjs-plugins
redningsselskapet / nestjs-plugins Goto Github PK
View Code? Open in Web Editor NEWPlugins for NestJS framework
License: ISC License
Plugins for NestJS framework
License: ISC License
nestjs-plugins
Creates a new connection for every publish. Connection should be reused.
When publishing a message, every consumer in a queue group receives the message. Only one should receive and create response.
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
import { NatsStreamConfig } from '@nestjs-plugins/nestjs-nats-jetstream-transport'
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.
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[];
}
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?
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/
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
...
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);
...
@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.
NatsError: 503
at Function.errorForCode (/Users/bernt/Coding/rs-dev/nestjs-plugins/packages/nats-publisher/node_modules/nats/nats-base-client/error.ts
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
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
Seems like removing this would fix it:
this.nc = undefined;
To Reproduce
Sporadic failure after jests run.
Expected behavior
No error.
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 22.04
Node: 16.13v
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.
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();
Any advice or thoughts on why this happens?
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
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)
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.
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"
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();
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";
Looks like app tries to delete consumer via $JS.API.CONSUMER.DELETE ....
after drain is called, can you plz take a look?
Thanks
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.
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)
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!
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.
Got this error after merging #29
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,
}
}
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?
Describe the bug
Consumer option ackWait ignored.
Option ackWait is not set by the serverConsumerOptionsBuilder
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
};
Describe the bug
No event handler is trigger when using inheritAppConfig and ClassSerializerInterceptor
To Reproduce
Steps to reproduce the behavior:
inheritAppConfig: true
.app.useGlobalInterceptors(new ClassSerializerInterceptor(app.get(Reflector)))
order.updated
@EventHandler('order.updated')
is omitted// app.useGlobalInterceptors(new ClassSerializerInterceptor(app.get(Reflector)))
order.updated
@EventHandler('order.updated')
is triggeredExpected behavior
All functions with decorator @EventHandler()
are triggered
A declarative, efficient, and flexible JavaScript library for building user interfaces.
🖖 Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
An Open Source Machine Learning Framework for Everyone
The Web framework for perfectionists with deadlines.
A PHP framework for web artisans
Bring data to life with SVG, Canvas and HTML. 📊📈🎉
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
Some thing interesting about web. New door for the world.
A server is a program made to process requests and deliver data to clients.
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
Some thing interesting about visualization, use data art
Some thing interesting about game, make everyone happy.
We are working to build community through open source technology. NB: members must have two-factor auth.
Open source projects and samples from Microsoft.
Google ❤️ Open Source for everyone.
Alibaba Open Source for everyone
Data-Driven Documents codes.
China tencent open source team.