NATS subject overlap understanding

[theobouwman]

I have these handlers:

2024-09-03 16:42:35,318 INFO     - momo-worker-default | scheduled.bimonthly.organisation.email.summary.send |            - `Handler` waiting for messages
2024-09-03 16:42:35,332 INFO     - momo-worker-default | organisation.email.summary.send                     |            - `Handler` waiting for messages
2024-09-03 16:42:35,336 INFO     - momo-worker-default | scheduled.daily.organisation.event.ended.reminder   |            - `Handler` waiting for messages
2024-09-03 16:42:35,340 INFO     - momo-worker-default | organisation.event.ended.reminder                   |            - `Handler` waiting for messages
2024-09-03 16:42:35,344 INFO     - momo-worker-default | user.auth.signin                                    |            - `Handler` waiting for messages
2024-09-03 16:42:35,350 INFO     - momo-worker-default | organisation.user.added                             |            - `Handler` waiting for messages

All configured like this:

@nats_router.subscriber(
    "scheduled.bimonthly.organisation.email.summary.send",
    durable="bimonthly-organisation-email-summary-send",
    stream=MOMO_WORKER_DEFAULT,
    max_workers=10,
    deliver_policy=DeliverPolicy.ALL,
    pull_sub=PullSub(batch_size=10),
)
@inject

@nats_router.subscriber(
    "organisation.email.summary.send",
    durable="organisation-email-summary-send",
    stream=MOMO_WORKER_DEFAULT,
    max_workers=10,
    deliver_policy=DeliverPolicy.ALL,
    pull_sub=PullSub(batch_size=10),
)
@inject

@nats_router.subscriber(
    "scheduled.daily.organisation.event.ended.reminder",
    durable="daily-organisation-event-ended-reminder",
    stream=MOMO_WORKER_DEFAULT,
    max_workers=10,
    deliver_policy=DeliverPolicy.ALL,
    pull_sub=PullSub(batch_size=10)
)
@inject

@nats_router.subscriber(
    "organisation.event.ended.reminder",
    durable="organisation-event-ended-reminder",
    stream=MOMO_WORKER_DEFAULT,
    max_workers=10,
    deliver_policy=DeliverPolicy.ALL,
    pull_sub=PullSub(batch_size=10)
)
@inject

And the stream configured like this

MOMO_WORKER_DEFAULT = JStream(
    name="momo-worker-default",
    retention=RetentionPolicy.WORK_QUEUE,
    max_age=60*60*1*3,
    declare=True,
    storage=StorageType.FILE,
    allow_direct=False,
)

The problem is that with this setup a message with subject organisation.event.ended.reminder is never handled. Only if I change the subject to event.ended.reminder on the handler and publish it it will be handled.

To my understanding the organisation.email and organisation.event are other "domains" right? So it is not really clear to me why the setup above is not working as expected...

[sheldygg]

Hi, I have create minimal reproducible example from your code with faststream==0.5.20

Code

from faststream import FastStream
from faststream.nats import NatsBroker, JStream, DeliverPolicy, PullSub, RetentionPolicy, StorageType, NatsMessage

broker = NatsBroker()
app = FastStream(broker)

MOMO_WORKER_DEFAULT = JStream(
    name="momo-worker-default",
    retention=RetentionPolicy.WORK_QUEUE,
    max_age=60 * 60 * 1 * 3,
    declare=True,
    storage=StorageType.FILE,
    allow_direct=False,
)


@broker.subscriber(
    "scheduled.bimonthly.organisation.email.summary.send",
    durable="bimonthly-organisation-email-summary-send",
    stream=MOMO_WORKER_DEFAULT,
    max_workers=10,
    deliver_policy=DeliverPolicy.ALL,
    pull_sub=PullSub(batch_size=10),
)
async def handle_1(msg: NatsMessage):
    print(msg)


@broker.subscriber(
    "organisation.email.summary.send",
    durable="organisation-email-summary-send",
    stream=MOMO_WORKER_DEFAULT,
    max_workers=10,
    deliver_policy=DeliverPolicy.ALL,
    pull_sub=PullSub(batch_size=10),
)
async def handle_2(msg: NatsMessage):
    print(msg)


@broker.subscriber(
    "scheduled.daily.organisation.event.ended.reminder",
    durable="daily-organisation-event-ended-reminder",
    stream=MOMO_WORKER_DEFAULT,
    max_workers=10,
    deliver_policy=DeliverPolicy.ALL,
    pull_sub=PullSub(batch_size=10)
)
async def handle_3(msg: NatsMessage):
    print(msg)


@broker.subscriber(
    "organisation.event.ended.reminder",
    durable="organisation-event-ended-reminder",
    stream=MOMO_WORKER_DEFAULT,
    max_workers=10,
    deliver_policy=DeliverPolicy.ALL,
    pull_sub=PullSub(batch_size=10)
)
async def handle_4(msg: NatsMessage):
    print(msg)

When I publish message with natscli nats publish organisation.event.ended.reminder 123
I receive it with faststream
NatsMessage(raw_message=Msg(_client=<nats client v2.9.0>, subject='organisation.event.ended.reminder', reply='$JS.ACK.momo-worker-default.organisation-event-ended-reminder.1.3.3.1725395942678804542.0', data=b'123', headers=None, _metadata=None, _ackd=False, _sid=5), body=b'123', headers={}, batch_headers=[], path={}, content_type='', reply_to='', message_id='21a2cd92-1a6c-43ca-ae28-3423b5a4daa3', correlation_id='2a80ccac-f948-4b4d-8997-ef3d59ad62e4', processed=False, committed=None, _decoded_body=123)

[theobouwman]

I don't exactly see what you have changed to get it working?

[sheldygg]

I just use NatsBroker instead of NatsRouter and add app to run with cli.
I don't change your stream and subjects.

[Lancetnik]

Probably, problem is in router logic? We had some problems with them already