Poll Mode

Overview

Poll mode is a pull-based way to consume queue messages: the consumer decides exactly when to ask for work and how much, instead of holding an open stream the broker pushes into. That control matters for batch jobs, cron-triggered workers, and any consumer that would rather ask "is there anything for me?" than keep a subscription alive — call poll_queue in a loop for continuous consumption without a persistent stream.

A single call to KubeMQ.Client.poll_queue sends a channel, max_items, and wait_timeout; the broker holds the request open as a long poll and returns once enough messages are available or the timeout elapses, so the call never spins on an empty queue. KubeMQ.PollResponse.ack_all/1 settles an entire batch at once after processing it.

Gotchas: nothing is acknowledged until you explicitly call ack_all/1 — a crash between receiving and acking leaves messages redelivered on the next poll; the timeout bounds latency, not throughput, so a small max_items on a busy queue means many round trips; and an {:ok, poll} with an empty messages list just means nothing arrived in that window, not that the queue is drained for good.

Prerequisites

• KubeMQ server running on localhost:50000
• Elixir SDK installed ({:kubemq, "~> 1.0"} in mix.exs)

Code

channel = "elixir-queues.poll-mode"
{:ok, client} = KubeMQ.Client.start_link(address: "localhost:50000", client_id: "elixir-qs-poll")

Task.start(fn ->
  Process.sleep(1_000)

  for i <- 1..5 do
    {:ok, _} = KubeMQ.Client.send_queue_message(client,
      KubeMQ.QueueMessage.new(channel: channel, body: "Poll msg #{i}"))
    Process.sleep(500)
  end

  IO.puts("All messages sent")
end)

IO.puts("Starting poll loop (3 iterations)...")

for iteration <- 1..3 do
  IO.puts("\n--- Poll iteration #{iteration} ---")

  case KubeMQ.Client.poll_queue(client,
         channel: channel,
         max_items: 5,
         wait_timeout: 3_000
       ) do
    {:ok, poll} when length(poll.messages) > 0 ->
      IO.puts("Got #{length(poll.messages)} messages")

      Enum.each(poll.messages, fn msg ->
        IO.puts("  Processing: #{msg.body}")
      end)

      {:ok, _} = KubeMQ.PollResponse.ack_all(poll)
      IO.puts("  Acked all")

    {:ok, _poll} ->
      IO.puts("No messages available")

    {:error, err} ->
      IO.puts("Poll error: #{err.message}")
  end
end

KubeMQ.Client.close(client)
IO.puts("\nPoll loop complete.")

How It Works

• A background Task sends messages while the main process polls
• Each poll iteration waits up to wait_timeout milliseconds for messages
• Guard clause when length(poll.messages) > 0 distinguishes between empty and non-empty polls
• Messages are acknowledged after processing each batch

Related

• Stream Receive
• Auto Ack