AndreMaz
commented
1 year ago Hi @dremekie thank you for reporting the issue and for supporting the project :smile:
Read options.nats.streamConfig and create one stream using the options specified there. (e.g. name="streamOne" and subjects=["streamOneTopic.*"]
Checked the code and you're right. ATM it's not possible to override stream name and subject so it creates one stream per handler. To be precise, it creates a stream where name is normalized handler name and subject is equal to an array with handler name. So in your example name=streamOneTopic_abc and subjects=['streamOneTopic.abc']
We can fix this by overriding handler default name and subjects with the ones that are defined in options.nats.streamConfig.subjects
For each subscription, create one consumer against the stream that is filtering by the specified topic (e.g. streamOneTopic. should create a consumer where filter_subject="streamOneTopic.".
The filter_subject can be set in handler configs:
broker.createService({
name: "sub",
streamOne: {
"streamOneTopic.abc": {
group: "other",
nats: {
consumerOptions: {
filter_subject: "streamOneTopic.*"
},
streamConfig: {
name: "streamOne", // Has NO effect at the moment
subjects: ["streamOneTopic.*"] // Has NO effect at the moment
}
},
async handler(payload) {
broker.logger.info(
`[---] Processing streamOneTopic: ${JSON.stringify(payload)}}`
);
}
}
}
});However, handler name with * is prohibited (streamOneTopic.* in your example) more info. Here's the snipped where NATS Jetstream client creates the consumer:
Your example produces the following error:
[2023-01-02T14:29:11.055Z] ERROR channelTest/CHANNELS: An error ocurred while create NATS Stream Error: invalid stream name - stream name cannot contain '*'
Can you please clarify the desired behavior? You want a single stream (e.g., orders.*) to handle several topics (e.g, orders.created, orders.received, etc.) ?
dremekie
icebob
Prerequisites
Please answer the following questions for yourself before submitting an issue.
Current Behavior
The adapter currently ignores any specified subjects in
options.nats.streamConfig.subjects. Instead, it will always use[nameOfStream]as the subject. This occurs here: https://github.com/moleculerjs/moleculer-channels/blob/master/src/adapters/nats.js#L182 . The adapter is trying to create a stream for each subscription it finds.The above will work. We end up with two streams:
streamOneTopic_abc(using subject=streamOneTopic.abc) andstreamOneTopic_xyz(using subject=streamOneTopic.xyz).But when you take the code snippet below (that listens for
streamOneTopic.*, it fails withError: no stream matches subject, even thought in though I specifiedoptions.nats.streamConfig.subjects = ["streamOneTopic.*"].At a high level, the following is how the NATS adapter should operate:
options.nats.streamConfigand create one stream using the options specified there. (e.g.name="streamOne"andsubjects=["streamOneTopic.*"]streamOneTopic.*should create a consumer wherefilter_subject="streamOneTopic.*".The current logic is "almost" there. My team uses NATS (and we sponsor Moleculer) and we're happy to battle test this adapter further.
Steps to Reproduce
Reproduce code snippet
Context
Please provide any relevant information about your setup. This is important in case the issue is not reproducible except for under certain conditions.
Failure Logs