Stream Send

Overview

Sending one queue message per call works fine for occasional traffic, but each call carries its own round trip. At high volume — event ingestion, sensor telemetry, log shipping — that per-call overhead caps your throughput well below what the connection can support.

send_queue_message_stream lazily opens a persistent upstream gRPC sender the first time you call it, then reuses that stream for every subsequent call — so repeated sends avoid the cost of establishing a new stream each time. Each call returns a result carrying the broker-assigned id, confirming the message was accepted without tearing down and reopening the connection between sends.

Gotchas: if the underlying stream breaks mid-run, the SDK raises StreamBrokenError and resets its internal state — your code needs to catch that and retry the call to re-establish a fresh stream, rather than assume one failure dooms every later send. Pick a unique channel and client ID before running against a shared server; the example's names are easy to collide with across parallel runs. Always close the client in an ensure block so the stream and connection are released even if a send raises.

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

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

  10.times do |i|
    msg = KubeMQ::Queues::QueueMessage.new(channel: channel, metadata: "item-#{i}", body: "data-#{i}")
    result = client.send_queue_message_stream(msg)
    puts "Sent item-#{i}: id=#{result.id}"
  end

  puts 'All messages sent via stream'
rescue KubeMQ::Error => e
  puts "KubeMQ error: #{e.message}"
ensure
  client&.close
  puts 'Done'
end

How It Works

• send_queue_message_stream lazily creates an upstream sender on first call for persistent gRPC streaming.
• If the stream breaks, it raises StreamBrokenError and resets — retry the call to establish a new stream.
• Review timeouts, channel names, and client IDs before running against shared environments.

Related

• Pattern overview
• Ruby SDK Reference
• Stream Receive
• Batch Send