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: send_queue_message enqueues a message on a channel over a unary RPC call, and receive_queue_messages pulls it back within a configurable wait_timeout_seconds.

Gotchas: this example uses the simple unary API, which is fine for low-volume or ad-hoc sends but not built for sustained throughput — reach for the stream API (send_queue_message_stream and poll) when you need to push or pull messages continuously. Calling receive_queue_messages against an empty queue isn't an error; it just waits out the timeout and returns an empty list. And max_messages only caps the batch size per call, so a single call won't necessarily drain a queue with more messages waiting.

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.send-receive'

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

  msg = KubeMQ::Queues::QueueMessage.new(channel: channel, metadata: 'order-1', body: 'order-data')
  client.send_queue_message(msg)
  puts 'Sent 1 message'

  messages = client.receive_queue_messages(
    channel: channel,
    max_messages: 5,
    wait_timeout_seconds: 5
  )
  puts "Received #{messages.size} message(s)"
  messages.each { |m| puts "  #{m.metadata}: #{m.body}" }
rescue KubeMQ::Error => e
  puts "KubeMQ error: #{e.message}"
ensure
  client&.close
  puts 'Done'
end

How It Works

• send_queue_message enqueues a message via a unary RPC call.
• receive_queue_messages pulls messages from the queue with a configurable wait timeout.
• For higher throughput, use the stream API (send_queue_message_stream and poll).
• Review timeouts, channel names, and client IDs before running against shared environments.

Related

• Pattern overview
• Ruby SDK Reference
• Stream Send
• Ack/Reject