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.

KubeMQ.Client.queue_upstream/1 opens a persistent, bidirectional gRPC stream and returns a handle; you reuse that handle to push any number of message batches without paying a new connection cost per batch. KubeMQ.QueueUpstreamHandle.send/2 writes a batch onto the stream and returns {:ok, results} with a per-message message_id and is_error flag, or {:error, err} if the stream itself failed.

Gotchas: an {:error, err} from send/2 typically means the stream is broken, not just one bad message — a fresh call to queue_upstream/1 is needed to get a working handle again. Even {:ok, results} can contain entries with is_error: true, so check each result rather than trusting the outer :ok alone. Always call KubeMQ.QueueUpstreamHandle.close/1 when done sending — an open handle holds a live stream and server-side resources for as long as the process keeps it.

Prerequisites

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

Code

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

{:ok, handle} = KubeMQ.Client.queue_upstream(client)
IO.puts("Upstream stream opened")

messages =
  for i <- 1..5 do
    KubeMQ.QueueMessage.new(channel: channel, body: "Stream msg #{i}")
  end

case KubeMQ.QueueUpstreamHandle.send(handle, messages) do
  {:ok, results} ->
    IO.puts("Sent #{length(results)} messages via stream")

    Enum.each(results, fn r ->
      IO.puts("  ID: #{r.message_id}, error: #{r.is_error}")
    end)

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

KubeMQ.QueueUpstreamHandle.close(handle)
KubeMQ.Client.close(client)
IO.puts("Done.")

How It Works

• queue_upstream/1 opens a bidirectional gRPC stream for efficient sending
• QueueUpstreamHandle.send/2 sends a batch of messages through the stream
• The handle should be closed when sending is complete to release resources

Related

• Batch Send
• Stream Receive