NATS – JetStream

				
					import asyncio
from nats.aio.client import Client as NATS
from nats.aio.errors import ErrConnectionClosed, ErrTimeout, ErrNoServers
from functools import partial

async def message_handler(msg, name):
    """
    Callback function to handle received messages.
    """
    data = msg.data.decode()
    print(f"[{name}] Received a message on '{msg.subject}': {data}")
    await msg.ack()  # Acknowledge the message

async def subscribe_to_subject(js, subject, queue_group, name):
    """
    Subscribes to the specified subject with a given queue group and handles incoming messages.
    """
    handler = partial(message_handler, name=name)
    try:
        # Subscribe to the subject with the specified queue group
        await js.subscribe(subject, queue=queue_group, cb=handler)
        print(f"Subscriber is listening on '{subject}' with queue group '{queue_group}'.")
    except Exception as e:
        print(f"Error subscribing to subject: {e}")

async def main(name):
    """
    Main coroutine that sets up the NATS client, connects to the server,
    and subscribes to a subject to receive messages.
    """
    nc = NATS()

    try:
        # Connect to the NATS server
        await nc.connect(servers=["nats://nats.nats:4222"])
        print("Subscriber connected to NATS server.")

        # Create JetStream context
        js = nc.jetstream()

        # Define the subject to subscribe to and the queue group
        subject = "my.subject.*"
        queue_group = "workers"

        # Subscribe to the subject
        await subscribe_to_subject(js, subject, queue_group, name)

        # Keep the subscriber running to listen for incoming messages
        print(f"Subscriber '{name}' is running. Press Ctrl+C to exit.")
        while True:
            await asyncio.sleep(1)

    except ErrNoServers as e:
        print(f"Could not connect to NATS server: {e}")
    except asyncio.CancelledError:
        pass  # Handle task cancellation gracefully
    except Exception as e:
        print(f"An error occurred: {e}")
    finally:
        # Close the NATS connection
        await nc.close()
        print("Subscriber disconnected from NATS server.")

if __name__ == '__main__':
    # Define the queue group name
    queue_group = "order_group"

    # Create two consumer tasks in the same queue group
    consumers = [
        main("Consumer1"),
        main("Consumer2"),
    ]

    # Run both consumers concurrently
    loop = asyncio.get_event_loop()
    try:
        loop.run_until_complete(asyncio.gather(*consumers))
    except KeyboardInterrupt:
        print("Consumers are shutting down.")