Send & Receive

Overview

Queue send/receive is the foundational operation for guaranteed-delivery, point-to-point messaging: you reach for it whenever work needs to survive past the moment it's created and be handled by exactly one consumer, not broadcast to every subscriber. Unlike pub/sub, a queued message sits durably on the broker until something pulls it, so the producer and consumer never need to be online at the same time — a slow or offline worker adds latency, it doesn't drop the message.

This tutorial builds the smallest possible version of that round trip: KubeMQ.Client.send_queue_message/2 enqueues a message on a channel, and KubeMQ.Client.receive_queue_messages/3 pulls it back within a bounded wait_timeout. This simple API acknowledges messages automatically on receive, so a successful {:ok, result} means the message is already gone from the queue.

Gotchas: because acknowledgment happens automatically on receive, a message is considered handled the moment it's delivered — if your process crashes right after receiving but before finishing the work, there's no chance to retry it. Calling receive_queue_messages/3 against an empty queue isn't an error; it just returns {:ok, result} with messages_received at zero once wait_timeout elapses. And max_messages caps how many messages one call can return, so a single call won't necessarily drain a queue with more messages waiting.

Prerequisites

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

Code

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

msg = KubeMQ.QueueMessage.new(channel: channel, body: "Hello Queue!", metadata: "order-data")

case KubeMQ.Client.send_queue_message(client, msg) do
  {:ok, result} ->
    IO.puts("Message sent! ID: #{result.message_id}")

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

case KubeMQ.Client.receive_queue_messages(client, channel,
       max_messages: 1,
       wait_timeout: 5_000
     ) do
  {:ok, result} ->
    IO.puts("Received #{result.messages_received} message(s)")

    Enum.each(result.messages, fn msg ->
      IO.puts("  Body: #{msg.body}")
      IO.puts("  Metadata: #{msg.metadata}")
    end)

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

KubeMQ.Client.close(client)

How It Works

• send_queue_message/2 sends a single message and returns {:ok, %QueueSendResult{}}
• receive_queue_messages/3 pulls messages from the queue with max_messages and wait_timeout
• Messages are automatically acknowledged on receive with the simple API

Related

• Queues Pattern Overview
• Batch Send
• Poll Mode