Bug: Subscribers registered in runtime do not clean up after close

[saschnet]

I am trying to use the new dynamic subscription from https://github.com/airtai/faststream/releases/tag/0.5.0 (see point number 9).

While adding the dynamic subscription works well, I have issues closing the dynamic subscription. See the following example, where I start a subscription and immediately stop it. Hence, the message that is published afterwards, must not be picked up by the handler.

Running the code below raises the exception in the handler, as it is executed.

I was also able to replicate this behavior using a real NATS server, hence, this is not a bug in the TestNatsBroker implementation.

How to reproduce

import asyncio
from faststream.nats import NatsBroker, TestNatsBroker
from faststream.nats.subscriber.asyncapi import AsyncAPISubscriber


async def start_stop(broker: NatsBroker) -> None:
    def handler_method() -> None:
        raise ValueError("This must never be executed")

    subscriber: AsyncAPISubscriber = broker.subscriber("dynamic")
    subscriber(handler_method)
    broker.setup_subscriber(subscriber)
    await subscriber.start()
    await subscriber.close()


async def main() -> None:
    broker = NatsBroker()
    async with TestNatsBroker(broker) as br:
        await start_stop(br)
        await broker.publish(subject="dynamic", message=None)

asyncio.run(main())

Environment
FastStream 0.5.34 with CPython 3.12.1 on Windows
Same on Linux.

Thanks for your work on this awesome project! :)

[Lancetnik]

Thank you for the report! I know about Issue and planning to fix it in 0.6.*
Unfortunately, such changes requires refactoring deep inside the project core