full wildcard stream

full-wildcard-stream-example.py

import asyncio

import nats
from nats.errors import TimeoutError


async def main():
    nc = await nats.connect("localhost")

    # Create JetStream context.
    js = nc.jetstream()

    # Persist messages on 'foo's subject.
    await js.add_stream(name="segments", subjects=["segments.>"])

    for i in range(0, 10):
        ack = await js.publish("segments.one.A.1", f"msg: {i}".encode(), headers={
            "Nats-Msg-Id": str(i)
        })
        print("Is duplicate?", ack.duplicate)

    # # Create pull based consumer on 'foo'.
    psub = await js.pull_subscribe("segments.*.*.1", "psub8")

    # Fetch and ack messagess from consumer.
    for i in range(0, 10):
        msgs = await psub.fetch(1)
        for msg in msgs:
            print(msg)

    for i in range(0, 10):
        ack = await js.publish("msg.id", f"msg: {i}".encode(), headers={
            "Nats-Msg-Id": str(i)
        })
        print("Is duplicate?", ack.duplicate)

    await nc.close()

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