subscriptions.py

import asyncio

from natsrpy import Message, Nats


async def main() -> None:
    """Main function to run the example."""
    nats = Nats(["nats://localhost:4222"])
    await nats.startup()

    cb_lock = asyncio.Event()

    async def callback(message: Message) -> None:
        print(f"[FROM_CALLBACK] {message.payload!r}")  # noqa: T201
        cb_lock.set()

    async with (
        # When subscribing you can set callback.
        # In that case CallbackSubscription is returned.
        # This type of subscription cannot be iterated.
        nats.subscribe("cb-subj", callback=callback) as cb_sub,
        # When callback is not set, you get a subscription
        # that should be used along with `async for`
        nats.subscribe("iter-subj") as iter_sub,
        # Subscriptions with queue argument create
        # subscription with a queue group to distribute
        # messages along all subscribers.
        nats.subscribe("queue-subj", queue="example-queue") as queue_sub,
    ):
        await nats.publish("cb-subj", "message for callback")
        await nats.publish("iter-subj", "message for iterator")
        await nats.publish("queue-subj", "message for queue sub")

        # We can unsubscribe after a particular amount of messages.
        await iter_sub.unsubscribe(limit=1)
        await cb_sub.unsubscribe(limit=1)
        await queue_sub.unsubscribe(limit=1)

        async for message in iter_sub:
            print(f"[FROM_ITERATOR] {message.payload!r}")  # noqa: T201

        async for message in queue_sub:
            print(f"[FROM_QUEUED] {message.payload!r}")  # noqa: T201

        # Making sure that the message in callback is received.
        await cb_lock.wait()

    # Don't forget to call shutdown.
    await nats.shutdown()


if __name__ == "__main__":
    asyncio.run(main())