Jetstream context can not be reused

[benjfield]

Observed behavior

Hi,

I may be doing something wrong but I couldn't find a good way to do this. If I try to reuse a JS context, it fails to send any messages and times out on every send

JS = None

async def _async_publish_to_nats(
  topic: str, 
  payload: Optional[bytes] = None, 
  headers: Optional[dict[str, str]] = None):
    global JS
    if JS is None:
      nc = await nats.connect(settings.NATS_URL)
      JS = nc.jetstream()

    await JS.publish(**kwargs)

Apologies if I am trying to solve this the wrong way, I have been through all the docs, asked on the slack and gone over most of the source code.

Expected behavior

The context should be capable of sending messages later. Ideally, without reconnecting as the connect takes around 30ms.

Server and client version

nats-py v2.10.0
nats-server v2.11.8

Host environment

No response

Steps to reproduce

No response

[caspervonb]

So the issue is likely, that you can't reuse a NATS connection across different event loops. When you do something like multiple asyncio.run() calls (or in a web framework where each request might create a new async context), the connection from the first event loop becomes unusable in the second one. The subscriptions are tied to the original event loop, so responses never make it back → timeout.

[benjfield]

Yeah, thats the case. There's also nothing in the library that generates a thread or an event loop I can see, so you have to do that yourself I think. The tests run with asyncio.run() which creates a loop and then closes it after the async calls have finished.

If other people want to get it running, I think you want to do something like this:

class JSSyncToAsync:
    nats_servers: list[str]
    loop: AbstractEventLoop

    def set_loop_and_connect(self: Self):
        self.loop = asyncio.new_event_loop()
        asyncio.set_event_loop(self.loop)
        self.added_topics = set()

        Thread(target=self.run_client, daemon=True).start()

        asyncio.run_coroutine_threadsafe(self.connect(), self.loop)

    def run_client(self: Self):
        self.loop.run_forever()

    async def connect(self: Self):
        ...

This will create a thread and run the event loop on it forever, which will handle the async calls from the nats library.