Pub/Sub

See Pub/Sub for the full guide — delivery semantics, subscription lifecycle, and cross-SDK topics.

Subscribing

@queue.subscriber()

@queue.subscriber(
    topic: str,
    name: str | None = None,
    queue: str = "default",
    durable: bool = True,
    **task_kwargs: Any,
) -> Callable[[Callable], TaskWrapper]

Register fn as an independent subscriber of topic. The function becomes a normal task — task_kwargs are forwarded to @queue.task() — and the subscription is written to storage when run_worker() starts, or via declare_subscriptions() in a producer-only process.

queue.declare_subscriptions()

queue.declare_subscriptions() -> None

Write pending durable subscriptions to storage. Called automatically at run_worker() startup — call it explicitly in a producer-only process (one that imports subscriber modules but never runs a worker) so publish() sees the subscriptions. Ephemeral subscriptions are skipped; they need an owning worker.

Publishing

queue.publish()

queue.publish(
    topic: str,
    *args: Any,
    idempotency_key: str | None = None,
    metadata: dict[str, Any] | None = None,
    notes: dict[str, Any] | None = None,
    priority: int | None = None,
    delay_seconds: float | None = None,
    max_retries: int | None = None,
    timeout: int | None = None,
    expires: float | None = None,
    result_ttl: int | None = None,
    **kwargs: Any,
) -> list[JobResult]

Publish a message to topic. Every active subscription receives an independent job carrying the same (args, kwargs) payload (at-least-once per subscriber). Returns one JobResult per delivery — empty when the topic has no active subscribers, a valid pub/sub no-op.

Unrecognized keyword arguments are forwarded as part of the task payload's kwargs, same as enqueue().

Log subscriptions

See Log topics for the cursor, at-least-once, and retention semantics.

queue.subscribe_log()

queue.subscribe_log(topic: str, name: str) -> None

Register a durable log subscription — a named cursor with no handler. Writes to storage immediately, so register it before the publishes it should see. Unlike @queue.subscriber(), there's no separate declare_subscriptions() step.

@queue.log_consumer()

@queue.log_consumer(
    topic: str,
    name: str | None = None,
    *,
    poll_interval: float = 1.0,
    batch_size: int = 100,
    on_error: str = "retry",
)

Decorator: register a managed consumer of a log topic. It creates the durable log subscription (like subscribe_log()) and, when run_worker() starts, spawns a daemon thread that pulls messages, invokes handler(*args, **kwargs) per message, and advances the cursor — the read/ack loop you'd otherwise hand-write. The handler may be sync or async.

• name defaults to the handler's name.
• poll_interval — seconds to wait after an empty poll before re-reading.
• batch_size — max messages pulled per poll.
• on_error — "retry" (default) leaves a failed message un-acked so the batch re-reads; "skip" acks past it and continues.

@queue.log_consumer("orders", "audit-log")
def audit(order_id: int) -> None:
    audit_sink.write(order_id)

queue.declare_topic()

queue.declare_topic(name: str, *, retention: float | None = None) -> None

Declare a log topic so its publishes are retained even with no subscriber (removing the late-join boundary). retention (seconds) bounds a sub-less backlog — each stored message expires that long after publish; None keeps messages until a subscriber consumes them. Idempotent: re-declaring updates retention.

queue.list_declared_topics()

queue.list_declared_topics() -> list[dict[str, Any]]

List declared topics. Each dict has name, mode, retention_ms (None if unbounded), and created_at (Unix ms).

queue.read_topic()

queue.read_topic(topic: str, name: str, limit: int = 100) -> list[TopicMessage]

Pull up to limit messages after name's cursor, oldest first and exclusive of it. Each message's payload is decoded with the queue serializer. Returns an empty list once the subscription is caught up. At-least-once — process the batch, then advance the cursor with ack_topic().

queue.ack_topic()

queue.ack_topic(topic: str, name: str, cursor: str) -> bool

Advance a log subscription's cursor to cursor (a message id). A high-water mark — acking an id acks everything up to and including it. Monotonic: acking an id older than what's already acked is a no-op and returns False.

queue.lease_topic()

queue.lease_topic(
    topic: str, name: str, limit: int = 100, visibility: float = 30.0
) -> list[TopicMessage]

Per-message alternative to the cursor read. Leases up to limit available messages for visibility seconds and tracks each individually: :meth:ack_message when done, :meth:nack_message to redeliver now, or let the lease expire to redeliver — so one poison message no longer blocks its siblings. Oldest first; in-flight (leased, un-expired) messages are skipped. Don't mix with read_topic/ack_topic on the same subscription.

queue.ack_message() / queue.nack_message()

queue.ack_message(topic: str, name: str, message_id: str) -> bool
queue.nack_message(topic: str, name: str, message_id: str) -> bool

ack_message ends a leased delivery (never redelivered); nack_message makes it available for redelivery immediately. Both return False when there was no un-acked delivery for message_id.

queue.topic_log_stats()

queue.topic_log_stats() -> list[dict[str, Any]]

Lag snapshot for every log subscription. Each dict contains topic, subscription, cursor (None if nothing has been acked yet), lag (un-acked message count), and oldest_unacked_age_ms (None when caught up).

TopicMessage

from flexiq import TopicMessage

One message pulled from a log topic, returned by read_topic() and lease_topic().

Managing subscriptions

queue.unsubscribe()

queue.unsubscribe(topic: str, name: str) -> bool

Remove a subscription. Returns False if none matched.

queue.pause_subscription() / queue.resume_subscription()

queue.pause_subscription(topic: str, name: str) -> bool
queue.resume_subscription(topic: str, name: str) -> bool

Stop or resume deliveries without unregistering. Returns False if the subscription is unknown.

queue.list_subscriptions()

queue.list_subscriptions(topic: str | None = None) -> list[dict[str, Any]]

List subscriptions — all of them, or one topic's active ones. Each dict contains topic, name, task_name, queue, active, and durable.

queue.list_topics()

queue.list_topics() -> list[str]

Distinct topics that currently have at least one subscription.