Delay Policy

Overview

QueueMessagePolicy.new(delay_seconds: ...) is the object that defers when a queued message becomes visible to consumers — attach it to the message before sending, and the broker excludes the message from delivery until the countdown expires. It starts the moment the broker accepts the message, not when the client sends it, and is evaluated once, at send time.

Gotchas: the delay is a floor, not a guarantee — the message becomes eligible when the timer expires, but actual delivery still waits for a consumer to poll, so don't rely on it for precise scheduling. It's one-shot: there's no recurrence or cron-like behavior, so long or repeating delays need application logic on top. And it's independent of redelivery — a delayed message that's later nacked or times out after delivery follows normal visibility-timeout/retry rules, not the original send-time delay.

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.delay-policy'

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

  policy = KubeMQ::Queues::QueueMessagePolicy.new(delay_seconds: 3)
  msg = KubeMQ::Queues::QueueMessage.new(
    channel: channel,
    metadata: 'delayed-3s',
    body: 'process-later',
    policy: policy
  )
  result = client.send_queue_message(msg)
  puts "Sent message with delay=3s: id=#{result.id}"

  receiver = client.create_downstream_receiver
  request = KubeMQ::Queues::QueuePollRequest.new(channel: channel, max_items: 1, wait_timeout: 1)
  response = receiver.poll(request)
  puts "Immediate poll: #{response.messages.size} messages (should be 0)"

  puts 'Waiting for delay to expire...'
  sleep 4

  response2 = receiver.poll(request)
  puts "After delay: #{response2.messages.size} message(s)"
  response2.messages.each do |m|
    puts "  #{m.metadata}: #{m.body}"
    m.ack
  end
rescue KubeMQ::Error => e
  puts "KubeMQ error: #{e.message}"
ensure
  receiver&.close
  client&.close
  puts 'Done'
end

How It Works

• delay_seconds in QueueMessagePolicy defers message visibility for the specified duration.
• The message is enqueued immediately but not delivered until the delay expires.
• Review timeouts, channel names, and client IDs before running against shared environments.

Related

• Pattern overview
• Ruby SDK Reference
• Delayed Messages
• Expiration Policy