Ack All

Overview

ack_all_queue_messages/3 acknowledges every pending message on a channel in a single broker-side call, without receiving them first. Reach for it when you want to drain a queue rather than process it — clearing a backlog of stale work after a bad deploy, resetting a channel between test runs, or discarding messages that are no longer relevant — where pulling and acking each message individually would be slow and wasteful.

Because it settles the whole channel at once, it is far cheaper than a receive-then-ack loop: the broker confirms all in-flight messages atomically and reports how many were affected via affected_messages on the result, using the wait_timeout option to bound how long it waits for in-flight transactions to settle before counting.

Gotchas: this is a blunt, irreversible instrument — it acknowledges all currently-pending messages, not a selected subset, so anything unprocessed is discarded, not redelivered. A busy channel may need a larger wait_timeout to catch messages still landing. For routine, per-message cleanup use ordinary acks, an expiration policy, or a dead-letter policy instead — save ack-all for deliberate, wholesale purges.

Prerequisites

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

Code

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

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

IO.puts("Sent 5 messages")

case KubeMQ.Client.ack_all_queue_messages(client, channel, wait_timeout: 5_000) do
  {:ok, result} ->
    IO.puts("Acknowledged #{result.affected_messages} messages")
    IO.puts("Error: #{result.is_error}")

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

case KubeMQ.Client.receive_queue_messages(client, channel,
       max_messages: 10,
       wait_timeout: 2_000
     ) do
  {:ok, result} ->
    IO.puts("Messages remaining: #{result.messages_received}")

  {:error, _} ->
    IO.puts("Queue is empty (as expected)")
end

KubeMQ.Client.close(client)

How It Works

• ack_all_queue_messages/3 acknowledges all pending messages without receiving them
• Returns %QueueAckAllResult{} with affected_messages count
• Useful for clearing a queue of stale messages

Related

• Send & Receive
• Ack/Reject