Overview

Events Store is KubeMQ's persistent pub/sub pattern → see Events Store for how the broker persists and replays event streams. Unlike plain Events, which are fire-and-forget, messages published with events_store= are persisted on the broker and a subscriber can replay the stored stream from any position.

In kubemq-faststream, you opt into Events Store by using the events_store= keyword on @broker.subscriber(...) and broker.publish(...), and you control where replay begins with the start_position argument (and, where required, start_value).

from kubemq_faststream import KubeMQBroker, StartPosition

broker = KubeMQBroker("kubemq://localhost:50000")


@broker.subscriber(
    events_store="audit-log",
    start_position=StartPosition.START_FROM_FIRST,
)
async def on_audit(msg: dict) -> None:
    print(f"Audit event: {msg}")

StartPosition is exported from the kubemq_faststream package alongside KubeMQBroker. Publishing to an events-store channel uses the same broker.publish(...) call you use for plain events, with events_store= in place of events=:

await broker.publish({"action": "login", "user": "alice"}, events_store="audit-log")

Start Positions

StartPosition is a string enum with six values. Three of them (START_AT_SEQUENCE, START_AT_TIME, START_AT_TIME_DELTA) need a companion start_value to say where to start; the other three are positional and ignore start_value.

Prop

Type

If you omit start_position, the subscriber behaves as START_FROM_NEW and only sees messages published after it connects.

START_FROM_NEW vs FIRST and LAST

The three positional values cover the common cases: ignore history, replay all of it, or jump to the latest.

START_FROM_NEW skips everything already in the store and only delivers messages published after the subscription is active — it is the persistent-channel equivalent of plain Events behavior:

start_from_new.py

import asyncio
import os

from faststream import FastStream

from kubemq_faststream import KubeMQBroker, StartPosition

KUBEMQ_ADDRESS = os.environ.get("KUBEMQ_ADDRESS", "kubemq://localhost:50000")

broker = KubeMQBroker(KUBEMQ_ADDRESS)
app = FastStream(broker)

CHANNEL = "example.events_store.start_from_new"


@broker.subscriber(
    events_store=CHANNEL,
    start_position=StartPosition.START_FROM_NEW,
    group="new-group",
)
async def handle_new_event(msg: dict) -> None:
    """Only receives events published after subscription starts."""
    print(f"[NEW] Received: {msg}")


@app.after_startup
async def run_demo() -> None:
    for i in range(1, 4):
        await broker.publish({"seq": i}, events_store=CHANNEL)
        print(f"Pre-subscription event {i} published (will NOT be received)")

    # Small delay to simulate a gap between historical and new events.
    await asyncio.sleep(1)

    for i in range(4, 6):
        await broker.publish({"seq": i}, events_store=CHANNEL)
        print(f"Post-subscription event {i} published")

    await asyncio.sleep(3)
    await app.stop()


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

START_FROM_FIRST replays the entire stored stream, and START_FROM_LAST delivers only the most recent stored message. The example below registers both on the same channel — each subscriber uses its own group so they replay independently — then publishes a few events and lets them replay:

basic_positions.py

import asyncio

from faststream import FastStream

from kubemq_faststream import KubeMQBroker, StartPosition

broker = KubeMQBroker("kubemq://localhost:50000")
app = FastStream(broker)

CHANNEL = "example.events_store.positions"


@broker.subscriber(
    events_store=CHANNEL,
    start_position=StartPosition.START_FROM_FIRST,
    group="first-group",
)
async def from_first(msg: dict) -> None:
    """Replay every stored event from the beginning."""
    print(f"[FIRST] Received: {msg}")


@broker.subscriber(
    events_store=CHANNEL,
    start_position=StartPosition.START_FROM_LAST,
    group="last-group",
)
async def from_last(msg: dict) -> None:
    """Receive only the most recent stored event."""
    print(f"[LAST] Received: {msg}")


@app.after_startup
async def run_demo() -> None:
    for i in range(1, 4):
        await broker.publish(
            {"order_id": i, "item": f"widget-{i}"},
            events_store=CHANNEL,
        )
        print(f"Published event {i}")

    await asyncio.sleep(3)
    await app.stop()


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

START_FROM_FIRST receives all three events; START_FROM_LAST receives only the latest.

Resume from a Sequence Number

Every message in an events-store channel has a monotonically increasing sequence number. START_AT_SEQUENCE with a start_value resumes delivery from that exact position — the canonical way to pick up where a consumer left off after a restart. Track the last sequence you processed, persist it, and pass it back as start_value on reconnect.

start_at_sequence.py

import asyncio

from faststream import FastStream

from kubemq_faststream import KubeMQBroker, StartPosition

broker = KubeMQBroker("kubemq://localhost:50000")
app = FastStream(broker)

CHANNEL = "example.events_store.at_sequence"
RESUME_AT = 3


@broker.subscriber(
    events_store=CHANNEL,
    start_position=StartPosition.START_AT_SEQUENCE,
    start_value=RESUME_AT,
)
async def handle_from_seq(msg: dict) -> None:
    """Process events starting from sequence 3."""
    print(f"[SEQ>={RESUME_AT}] Received: {msg}")


@app.after_startup
async def run_demo() -> None:
    for i in range(1, 6):
        await broker.publish(
            {"seq_demo": i, "data": f"payload-{i}"},
            events_store=CHANNEL,
        )
        print(f"Published event {i}")

    await asyncio.sleep(3)
    await app.stop()


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

After publishing 5 events, the subscriber starting at sequence 3 receives only events 3, 4, and 5.

Time-Based Replay

Two start positions replay by time instead of by sequence:

START_AT_TIME — start_value is an absolute Unix timestamp (seconds since epoch). Delivery begins at the first message stored at or after that moment.
START_AT_TIME_DELTA — start_value is a number of seconds to look back. The broker translates "N seconds ago" into a starting point relative to now.

start_at_time.py

import asyncio
import time

from faststream import FastStream

from kubemq_faststream import KubeMQBroker, StartPosition

broker = KubeMQBroker("kubemq://localhost:50000")
app = FastStream(broker)

CHANNEL = "example.events_store.at_time"


@broker.subscriber(
    events_store=CHANNEL,
    start_position=StartPosition.START_AT_TIME_DELTA,
    start_value=60,
    group="delta-group",
)
async def from_last_60s(msg: dict) -> None:
    """Receive events published in the last 60 seconds."""
    print(f"[DELTA 60s] Received: {msg}")


@broker.subscriber(
    events_store=CHANNEL,
    start_position=StartPosition.START_AT_TIME,
    start_value=int(time.time()) - 30,
    group="abs-time-group",
)
async def from_absolute_time(msg: dict) -> None:
    """Receive events published since 30 seconds ago (absolute timestamp)."""
    print(f"[ABS TIME] Received: {msg}")


@app.after_startup
async def run_demo() -> None:
    for i in range(1, 4):
        await broker.publish(
            {"event_id": i, "ts": time.time()},
            events_store=CHANNEL,
        )
        print(f"Published event {i}")

    await asyncio.sleep(3)
    await app.stop()


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

START_AT_TIME_DELTA with start_value=60 replays the last minute; START_AT_TIME with start_value=int(time.time()) - 30 replays from an absolute timestamp 30 seconds in the past.

Consumer-Group Replay

Start positions combine with consumer groups the same way they do for plain Events. When multiple subscribers join the same group on an events-store channel, messages are load-balanced across the group members rather than fanned out to all of them — while each member still honors the configured start_position. This lets you scale a replay across several workers that share the stored backlog.

consumer_group_replay.py

import asyncio

from faststream import FastStream

from kubemq_faststream import KubeMQBroker, StartPosition

broker = KubeMQBroker("kubemq://localhost:50000")
app = FastStream(broker)

CHANNEL = "example.events_store.group_replay"
GROUP = "workers"


@broker.subscriber(
    events_store=CHANNEL,
    group=GROUP,
    start_position=StartPosition.START_FROM_FIRST,
)
async def worker_a(msg: dict) -> None:
    """Worker A in the consumer group."""
    print(f"[Worker-A] Received: {msg}")


@broker.subscriber(
    events_store=CHANNEL,
    group=GROUP,
    start_position=StartPosition.START_FROM_FIRST,
)
async def worker_b(msg: dict) -> None:
    """Worker B in the consumer group."""
    print(f"[Worker-B] Received: {msg}")


@app.after_startup
async def run_demo() -> None:
    for i in range(1, 7):
        await broker.publish(
            {"task_id": i, "action": f"process-{i}"},
            events_store=CHANNEL,
        )
        print(f"Published task {i}")

    await asyncio.sleep(3)
    await app.stop()


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

The six published tasks are distributed across Worker-A and Worker-B instead of each worker receiving all six.

Like plain Events, Events Store is fire-and-forget delivery: ack / nack operations are no-ops for this pattern. Acknowledgement and requeue semantics apply only to Queues. Replay position is controlled entirely by start_position, not by settlement.

Events (Pub/Sub)

The non-persistent counterpart — fire-and-forget broadcast with optional group load balancing.

Events Store concept

How KubeMQ persists and replays event streams at the broker level.

Events Store (Persistent Pub/Sub)

Overview

Start Positions

START_FROM_NEW vs FIRST and LAST

Resume from a Sequence Number

Time-Based Replay

Consumer-Group Replay

Events (Pub/Sub)

Events Store concept

On this page

Events Store (Persistent Pub/Sub)

Overview

Start Positions

START_FROM_NEW vs FIRST and LAST

Resume from a Sequence Number

Time-Based Replay

Consumer-Group Replay

Related

Events (Pub/Sub)

Events Store concept

On this page