Examples

import asyncio

from apache_iggy import IggyClient
from apache_iggy import SendMessage as Message

STREAM_NAME = "sample-stream"
TOPIC_NAME = "sample-topic"
PARTITION_ID = 0


async def main():
    client = IggyClient.from_connection_string(
        "iggy+tcp://iggy:iggy@127.0.0.1:8090"
    )
    await client.connect()

    # Re-running this example is fine: only create what is missing.
    if await client.get_stream(STREAM_NAME) is None:
        await client.create_stream(name=STREAM_NAME)

    if await client.get_topic(STREAM_NAME, TOPIC_NAME) is None:
        await client.create_topic(
            stream=STREAM_NAME,
            name=TOPIC_NAME,
            partitions_count=1,
            replication_factor=1,
        )

    messages = [Message(f"message-{i}") for i in range(10)]

    await client.send_messages(
        stream=STREAM_NAME,
        topic=TOPIC_NAME,
        partitioning=PARTITION_ID,
        messages=messages,
    )
    print(f"Sent {len(messages)} message(s)")


asyncio.run(main())

import asyncio

from apache_iggy import IggyClient, PollingStrategy

STREAM_NAME = "sample-stream"
TOPIC_NAME = "sample-topic"
PARTITION_ID = 0


async def main():
    client = IggyClient.from_connection_string(
        "iggy+tcp://iggy:iggy@127.0.0.1:8090"
    )
    await client.connect()

    # Next() with auto_commit=True continues from this consumer's last committed
    # offset, so each run picks up where the previous one finished.
    polled_messages = await client.poll_messages(
        stream=STREAM_NAME,
        topic=TOPIC_NAME,
        partition_id=PARTITION_ID,
        polling_strategy=PollingStrategy.Next(),
        count=10,
        auto_commit=True,
    )

    for message in polled_messages:
        payload = message.payload().decode("utf-8")
        print(f"Offset: {message.offset()}, Payload: {payload}")


asyncio.run(main())