Delayed Messages

Overview

A delivery delay holds a queue message out of consumers' reach for a fixed window after it's sent — the message is accepted and persisted immediately, but invisible to pollers until the delay expires. It's the building block for scheduled work — a reminder to fire in an hour, a retry with back-off, a task queued for off-peak processing — without standing up a separate scheduler or cron service.

Set it with delay_in_seconds on the QueueMessage before sending; the broker does the waiting. The send result reports the message ID, and a poll against the channel before the delay elapses simply returns zero messages — it isn't hidden in a separate place, it's the same queue, just not yet eligible for delivery. Once the delay window passes, the next poll (via receive_queue_messages) retrieves it normally.

Gotchas: the delay is set once at send time and can't be extended or shortened afterward — if you need a different wait, send a new message. A long delay still counts as an in-flight, persisted message, so it survives a broker restart, but it also occupies queue storage for the whole waiting period. Don't confuse this with a visibility timeout after delivery — that's a separate mechanism for redelivery on failed acknowledgment, not initial availability.

Prerequisites

• KubeMQ server running on localhost:50000
• Python SDK installed (pip install kubemq)

Code

"""Example: Delayed messages — send a message that becomes available after a delay."""

from __future__ import annotations

import asyncio

from kubemq import AsyncQueuesClient
from kubemq import QueueMessage


async def main() -> None:
    async with AsyncQueuesClient(
        address="localhost:50000",
        client_id="python-queues-delayed-messages-client",
    ) as client:
        # Send a message with a 5-second delay
        result = await client.send_queue_message(
            QueueMessage(
                channel="python-queues.delayed-messages",
                body=b"message with delay",
                delay_in_seconds=5,
            )
        )
        print(f"Sent delayed message: {result}")

        # Try to receive immediately — should get nothing
        response = await client.receive_queue_messages(
            channel="python-queues.delayed-messages",
            max_messages=1,
            wait_timeout_seconds=1,
            auto_ack=True,
        )
        print(f"Immediate poll: {len(response.messages)} messages (expected 0)")

        # Wait for the delay to expire
        print("Waiting 6 seconds for delay to expire...")
        await asyncio.sleep(6)

        # Now the message should be available
        response = await client.receive_queue_messages(
            channel="python-queues.delayed-messages",
            max_messages=1,
            wait_timeout_seconds=5,
            auto_ack=True,
        )
        print(f"After delay: {len(response.messages)} messages (expected 1)")
        for msg in response.messages:
            print(f"  Received: {msg.body.decode('utf-8')}")


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

How It Works

• QueueMessage.delay_in_seconds=5 instructs the broker to hold the message for 5 seconds before making it available to consumers.
• The immediate poll with wait_timeout_seconds=1 confirms the message is not yet available (0 messages returned).
• After asyncio.sleep(6) the delay window has elapsed; the second poll successfully retrieves the message.
• Delayed delivery is useful for scheduled tasks, retry back-off, and deferred processing without a separate scheduler.

Related

• Delay Policy — field-level reference for delay_in_seconds
• Pattern overview
• Python SDK Reference
• Send & Receive
• Ack All