Peek Messages

Overview

Peeking lets you look at what's sitting in a queue without touching it — the messages stay exactly where they are, still waiting for whichever consumer eventually receives them. It's the tool you reach for when you need visibility into queue state — checking backlog depth, inspecting payloads while debugging a stuck pipeline, or building an operational dashboard — without risking a collision with real consumers competing for the same work.

peek_queue_messages is a variant of the same receive call your consumers use, just in read-only mode: it takes the same channel, max_messages, and wait_timeout_seconds and returns an AsyncQueuesPollResponse, but it never marks the messages as delivered, locks them, or starts a visibility timeout — so there's no acknowledgment step, and calling it again returns the same messages.

Gotchas: peeked messages aren't reserved for you — a consumer can receive_queue_messages and remove them the instant after you peek, so treat the count as a point-in-time estimate, not a guarantee. Peek also won't surface messages already locked inside another consumer's in-flight receive transaction, and it's not a substitute for receiving when you actually intend to process what you see.

Prerequisites

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

Code

"""Example: Peek messages — inspect queue messages without consuming them."""

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-peek-messages-client",
    ) as client:
        # Send a message to peek at
        await client.send_queue_message(
            QueueMessage(
                channel="python-queues.peek-messages",
                body=b"peek-me",
                metadata="test",
            )
        )

        # Peek at messages (they remain in the queue)
        waiting_result = await client.peek_queue_messages("python-queues.peek-messages", 5, 10)
        if waiting_result.is_error:
            print(f"Error: {waiting_result.error}")
            return

        print(f"Messages waiting: {len(waiting_result.messages)}")
        for msg in waiting_result.messages:
            print(f"  Peek: id={msg.id}, body={msg.body.decode('utf-8')}")

        # Messages are still available after peeking
        verify_result = await client.peek_queue_messages("python-queues.peek-messages", 5, 10)
        print(f"Messages still waiting after peek: {len(verify_result.messages)}")


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

How It Works

• peek_queue_messages reads messages from the queue non-destructively; the messages remain available and their visibility is unchanged.
• The method takes (channel, max_messages, wait_timeout_seconds) and returns an AsyncQueuesPollResponse with messages and is_error.
• Calling peek_queue_messages a second time returns the same messages, confirming they were not consumed by the first call.
• Peek is useful for monitoring queue depth and inspecting message content without affecting downstream consumers.

Related

• Pattern overview
• Python SDK Reference
• Send & Receive
• Ack All