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 only runs intermittently and would rather ask "is there anything for me?" than keep a subscription alive.

A single call to receiver.poll with a QueuePollRequest 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. Each returned message is acknowledged individually via ack, giving you a chance to skip one you can't process.

Gotchas: a short wait_timeout (1 second here) makes the poll feel near-instant but also means more frequent empty responses on a quiet queue — check response.error? before assuming that's a problem; the timeout bounds latency, not throughput, so a small max_items on a busy queue means many round trips; and un-acked messages return to the queue only after the visibility timeout, not immediately.

Prerequisites

• KubeMQ server running on localhost:50000
• Ruby SDK installed (gem install kubemq)

Code

require 'kubemq'

address = ENV.fetch('KUBEMQ_ADDRESS', 'localhost:50000')
channel = 'ruby-queues.poll-mode'

begin
  client = KubeMQ::QueuesClient.new(address: address, client_id: 'qstream-poll-example')
  puts "Connected to #{address}"

  msg = KubeMQ::Queues::QueueMessage.new(channel: channel, metadata: 'poll-item', body: 'data')
  client.send_queue_message(msg)
  puts 'Sent 1 message'

  receiver = client.create_downstream_receiver

  request = KubeMQ::Queues::QueuePollRequest.new(
    channel: channel,
    max_items: 10,
    wait_timeout: 1
  )
  response = receiver.poll(request)

  if response.error?
    puts "Poll error: #{response.error}"
  else
    puts "Non-blocking poll returned #{response.messages.size} messages:"
    response.messages.each do |m|
      puts "  #{m.metadata}: #{m.body}"
      m.ack
    end
  end
rescue KubeMQ::Error => e
  puts "KubeMQ error: #{e.message}"
ensure
  receiver&.close
  client&.close
  puts 'Done'
end

How It Works

• Setting a short wait_timeout (e.g. 1 second) creates a near-non-blocking poll.
• If no messages are available within the timeout, the response contains an empty messages array.
• Review timeouts, channel names, and client IDs before running against shared environments.

Related

• Pattern overview
• Ruby SDK Reference
• Stream Receive
• Send & Receive