Delayed Messages

Overview

A delivery delay holds a queue message out of consumers' reach for a fixed window after it's sent — the message is accepted and persisted immediately, but invisible to pollers until the delay expires. It's the building block for scheduled work — a reminder to fire in an hour, a retry with back-off, a task queued for off-peak processing — without standing up a separate scheduler or cron service.

Set it with delay_seconds on KubeMQ.QueuePolicy before sending; the broker does the waiting. The send result's delayed_to field returns the timestamp when the message becomes visible, so you can log or monitor exactly when delivery will happen. Until then, any poll via receive_queue_messages against that channel simply returns nothing for that message — it isn't hidden in a separate place, it's the same queue, just not yet eligible for delivery.

Gotchas: the delay is set once at send time and can't be extended or shortened afterward — if you need a different wait, send a new message. A long delay still counts as an in-flight, persisted message, so it survives a broker restart, but it also occupies queue storage for the whole waiting period. Don't confuse this with a visibility timeout after delivery — that's a separate mechanism for redelivery on failed acknowledgment, not initial availability.

Prerequisites

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

Code

channel = "elixir-queues.delayed-messages"
{:ok, client} = KubeMQ.Client.start_link(address: "localhost:50000", client_id: "elixir-queue-delayed")

msg = KubeMQ.QueueMessage.new(
  channel: channel,
  body: "Delayed delivery!",
  policy: KubeMQ.QueuePolicy.new(delay_seconds: 5)
)

{:ok, result} = KubeMQ.Client.send_queue_message(client, msg)
IO.puts("Message sent with 5s delay. Delayed to: #{result.delayed_to}")

IO.puts("Trying to receive immediately...")

case KubeMQ.Client.receive_queue_messages(client, channel,
       max_messages: 1,
       wait_timeout: 2_000
     ) do
  {:ok, result} ->
    IO.puts("Immediate receive: #{result.messages_received} messages (expected 0)")

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

IO.puts("Waiting for delay to expire...")
Process.sleep(6_000)

case KubeMQ.Client.receive_queue_messages(client, channel,
       max_messages: 1,
       wait_timeout: 5_000
     ) do
  {:ok, result} ->
    IO.puts("After delay: #{result.messages_received} message(s)")
    Enum.each(result.messages, fn m -> IO.puts("  Body: #{m.body}") end)

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

KubeMQ.Client.close(client)

How It Works

• QueuePolicy.new(delay_seconds: 5) defers delivery for 5 seconds
• The message is invisible to consumers until the delay expires
• result.delayed_to shows the timestamp when the message becomes available

Related

• Expiration Policy