Bug: NATS stream publish message with subject path parameters not working

[nason]

Describe the bug
Provide a clear and concise description of the bug.

I am trying to set up a NATS stream subject that includes route parameters. When I publish a message to it, I get nats.js.errors.NoStreamResponseError: nats: no response from stream

How to reproduce

Minimal example:

from faststream import FastStream, Path, Logger
from faststream.nats import JStream, NatsBroker, NatsMessage
from pydantic import BaseModel

broker = NatsBroker(verbose=True)
app = FastStream(broker=broker)

stream = JStream(name="testing")

class Event(BaseModel):
    id: str


class Result(BaseModel):
    status: str


# These work fine

@broker.subscriber("plain.{param}")
async def example_non_stream(event: Event, msg: NatsMessage, logger: Logger, param: str = Path()) -> Result:
    logger.info(f"Event: {event}, param: {param}")
    await msg.ack()
    return {"status": "pending"}


@broker.subscriber("example.stream-no-param", stream=stream, deliver_policy="new")
async def example(event: Event, msg: NatsMessage, logger: Logger) -> Result:
    logger.info(f"Event: {event}")
    await msg.ack()
    return {"status": "pending"}


# This does not

@broker.subscriber("example.stream-with-param.{param}", stream=stream, deliver_policy="new")
async def example_with_path(event: Event, msg: NatsMessage, logger: Logger, param: str = Path()) -> Result:
    logger.info(f"Event: {event}, param: {param}")
    await msg.ack()
    return {"status": "pending"}


@app.after_startup
async def after_startup():
    await broker.connect()

    await broker.publish(subject="plain.test", message={"id": "0"})
    await broker.publish(subject="example.stream-no-param", message={"id": "1"}, stream="testing")
    await broker.publish(subject="example.stream-with-param.test", message={"id": "2"}, stream="testing")

Steps to reproduce the behavior:

• faststream run bug_repro:app --reload

Expected behavior

I expect the example.stream-with-param.test message to publish successfully, and the example_with_path subscriber to log param as "test"

Observed behavior

❯ faststream run app.asyncapi.repro:app --reload
2023-10-26 15:01:53,197 INFO     - Started reloader process [89581] using WatchFiles
2023-10-26 15:01:53,370 INFO     - FastStream app starting...
2023-10-26 15:01:53,376 INFO     -         | plain.*                           |            - `ExampleNonStream` waiting for messages
2023-10-26 15:01:53,378 INFO     - testing | example.stream-no-param           |            - `Example` waiting for messages
2023-10-26 15:01:53,382 INFO     - testing | example.stream-with-param.*       |            - `ExampleWithPath` waiting for messages
2023-10-26 15:01:53,386 INFO     -         | plain.{param}                     | 76c3965e-4 - Received
2023-10-26 15:01:53,386 INFO     - testing | example.stream-no-param           | 6e0091b3-d - Received
2023-10-26 15:01:53,387 INFO     - testing | example.stream-no-param           | 6e0091b3-d - Event: id='1'
2023-10-26 15:01:53,387 INFO     - testing | example.stream-no-param           | 6e0091b3-d - Processed
2023-10-26 15:01:53,387 INFO     -         | plain.{param}                     | 76c3965e-4 - Event: id='0', param: test
2023-10-26 15:01:53,388 INFO     -         | plain.{param}                     | 76c3965e-4 - Processed
Process SpawnProcess-1:
Traceback (most recent call last):
   <snipped for brevity>
nats.js.errors.NoStreamResponseError: nats: no response from stream

Environment

❯ faststream -v        
Running FastStream 0.2.8 with CPython 3.11.5 on Darwin

Additional context
I'm fairly new to NATS and to FastStream, but I think this should work, right? There's a good chance I'm missing something but thought I'd report here just in case it is a bug

If I log await broker.stream.consumers_info("testing") before I start publishing in after_startup, I see:

[
  ConsumerInfo(name='GR6YBg7c', stream_name='testing', config=ConsumerConfig(name=None, durable_name=None, description=None, deliver_policy='new', opt_start_seq=None, opt_start_time=None, ack_policy='explicit', ack_wait=30.0, max_deliver=-1, filter_subject='example.stream-no-param', replay_policy='instant', rate_limit_bps=None, sample_freq=None, max_waiting=None, max_ack_pending=1000, flow_control=None, idle_heartbeat=None, headers_only=None, deliver_subject='_INBOX.g5Zton5u9z3yyNWuaObWkF', deliver_group=None, inactive_threshold=5.0, num_replicas=0, mem_storage=None), delivered=SequenceInfo(consumer_seq=0, stream_seq=12), ack_floor=SequenceInfo(consumer_seq=0, stream_seq=0), num_ack_pending=0, num_redelivered=0, num_waiting=0, num_pending=0, cluster=ClusterInfo(leader='nats-0', name='nats-local', replicas=None), push_bound=True),
  ConsumerInfo(name='caWWKktc', stream_name='testing', config=ConsumerConfig(name=None, durable_name=None, description=None, deliver_policy='new', opt_start_seq=None, opt_start_time=None, ack_policy='explicit', ack_wait=30.0, max_deliver=-1, filter_subject='example.stream-with-param.*', replay_policy='instant', rate_limit_bps=None, sample_freq=None, max_waiting=None, max_ack_pending=1000, flow_control=None, idle_heartbeat=None, headers_only=None, deliver_subject='_INBOX.g5Zton5u9z3yyNWuaObWoP', deliver_group=None, inactive_threshold=5.0, num_replicas=0, mem_storage=None), delivered=SequenceInfo(consumer_seq=0, stream_seq=12), ack_floor=SequenceInfo(consumer_seq=0, stream_seq=0), num_ack_pending=0, num_redelivered=0, num_waiting=0, num_pending=0, cluster=ClusterInfo(leader='nats-0', name='nats-local', replicas=None), push_bound=True)
]

[Lancetnik]

@nason thanks for your bugreport! Wait a bit: 0.2.9 release will fix it
Also, your FastStream code looks great, looks like you got all of its' features
What do you think about the framework?

[nason]

Wow that was fast, thank you for the quick fix!

What do you think about the framework?

Really great work, A+. I've been working with event driven architectures for years and I can already tell FastStream (and also AsyncAPI) are going to make building and developing these types of systems so much better

[davorrunje]

The new release 0.2.9 is out.

[nason]

Confirmed 0.2.9 fixed the issue I was seeing. Thank you!

faststream run app.asyncapi.repro:app --reload
2023-10-26 17:07:55,173 INFO     - Started reloader process [21459] using WatchFiles
2023-10-26 17:07:55,359 INFO     - FastStream app starting...
2023-10-26 17:07:55,365 INFO     -         | plain.*                           |            - `ExampleNonStream` waiting for messages
2023-10-26 17:07:55,367 INFO     - testing | example.stream-no-param           |            - `Example` waiting for messages
2023-10-26 17:07:55,370 INFO     - testing | example.stream-with-param.*       |            - `ExampleWithPath` waiting for messages
2023-10-26 17:07:55,374 INFO     -         | plain.{param}                     | 0f86a66d-c - Received
2023-10-26 17:07:55,374 INFO     - testing | example.stream-no-param           | 29939ad2-4 - Received
2023-10-26 17:07:55,375 INFO     - testing | example.stream-no-param           | 29939ad2-4 - Event: id='1'
2023-10-26 17:07:55,376 INFO     - testing | example.stream-no-param           | 29939ad2-4 - Processed
2023-10-26 17:07:55,376 INFO     -         | plain.{param}                     | 0f86a66d-c - Event: id='0', param: test
2023-10-26 17:07:55,376 INFO     -         | plain.{param}                     | 0f86a66d-c - Processed
2023-10-26 17:07:55,377 INFO     - testing | example.stream-with-param.{param} | 4973d1f2-0 - Received
2023-10-26 17:07:55,377 INFO     - FastStream app started successfully! To exit, press CTRL+C
2023-10-26 17:07:55,377 INFO     - testing | example.stream-with-param.{param} | 4973d1f2-0 - Event: id='2', param: test
2023-10-26 17:07:55,377 INFO     - testing | example.stream-with-param.{param} | 4973d1f2-0 - Processed