Stream Send

Overview

Stream send covers publishing a batch of persistent events over one long-lived connection instead of opening a new request for each message. A single-shot publish call is fine for one-off writes, but if you're bulk-loading history, replicating a firehose of records, or backfilling an Events Store channel, paying gRPC connection overhead once instead of per-message turns network latency into your throughput ceiling instead of an app-level bottleneck.

create_events_store_sender opens a persistent gRPC stream and hands back a sender you reuse for every message. Each sender.publish(msg) call still confirms storage synchronously, returning a result with the assigned id and a sent flag, before you explicitly close the sender when you're done. Gotchas: because each publish awaits its own confirmation, this pattern is latency-bound per call — true concurrent throughput needs multiple in-flight sends, not just a shared connection; closing the sender while a publish call is still pending can cut off its confirmation; and for occasional publishing, opening and tearing down a stream sender is pure overhead — send events store messages directly instead.

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

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

  sender = client.create_events_store_sender
  puts 'Created events store stream sender'

  10.times do |i|
    msg = KubeMQ::PubSub::EventStoreMessage.new(
      channel: channel,
      metadata: "stream es event #{i}",
      body: "payload-#{i}"
    )
    result = sender.publish(msg)
    puts "Sent event #{i}: sent=#{result.sent}, id=#{result.id}"
  end

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

How It Works

• create_events_store_sender opens a persistent gRPC stream with synchronous per-message confirmation.
• Each publish call returns a result confirming the event was persisted.
• Close the sender explicitly when finished.
• Review timeouts, channel names, and client IDs before running against shared environments.

Related

• Pattern overview
• Ruby SDK Reference
• Persistent Pub/Sub