Delay Policy

Overview

A delay policy defers when a queued message becomes visible to consumers — you send it now, but nothing can receive it until a countdown you set expires. That's the mechanism behind retry-after-backoff, rate-limited notifications, "remind me in an hour" workflows, and staggering a burst of work so it doesn't hit downstream consumers all at once, all without standing up a separate scheduler.

It works entirely at send time: KubeMQ.QueuePolicy.new(delay_seconds: 3) attaches a delay to the message before it's passed to the upstream handle's send. The broker starts the countdown the moment it accepts the message and simply excludes it from delivery until the timer elapses — after that it behaves like any other queued message, available to whichever consumer calls poll_queue next.

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
• Elixir SDK installed ({:kubemq, "~> 1.0"} in mix.exs)

Code

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

{:ok, handle} = KubeMQ.Client.queue_upstream(client)

msg = KubeMQ.QueueMessage.new(
  channel: channel,
  body: "Delayed via stream",
  policy: KubeMQ.QueuePolicy.new(delay_seconds: 3)
)

case KubeMQ.QueueUpstreamHandle.send(handle, [msg]) do
  {:ok, results} ->
    r = hd(results)
    IO.puts("Sent with delay. Delayed to: #{r.delayed_to}")

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

IO.puts("Polling immediately...")

case KubeMQ.Client.poll_queue(client,
       channel: channel,
       max_items: 1,
       wait_timeout: 2_000
     ) do
  {:ok, poll} ->
    IO.puts("Immediate poll: #{length(poll.messages)} messages")

  {:error, _} ->
    IO.puts("No messages yet")
end

IO.puts("Waiting 4 seconds...")
Process.sleep(4_000)

case KubeMQ.Client.poll_queue(client,
       channel: channel,
       max_items: 1,
       wait_timeout: 5_000
     ) do
  {:ok, poll} ->
    IO.puts("After delay: #{length(poll.messages)} message(s)")
    Enum.each(poll.messages, fn m -> IO.puts("  #{m.body}") end)
    {:ok, _} = KubeMQ.PollResponse.ack_all(poll)

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

KubeMQ.QueueUpstreamHandle.close(handle)
KubeMQ.Client.close(client)

How It Works

• QueuePolicy.new(delay_seconds: 3) defers delivery for 3 seconds
• The message is sent via the upstream stream for efficient sending
• An immediate poll returns nothing; after the delay, the message is available

Related

• Expiration Policy