Stream Send

Overview

Publishing events one at a time means each call pays its own round-trip: write the request, wait on the connection, then move to the next event. That's fine for occasional notifications, but it caps throughput when you need to push hundreds or thousands of events per second — log forwarding, sensor telemetry, change-data-capture feeds — where per-call overhead dominates.

KubeMQ.Client.send_event_stream/1 opens one bidirectional gRPC stream up front and returns a handle. Each subsequent KubeMQ.EventStreamHandle.send(handle, event) writes onto that already-open stream instead of negotiating a new call, so a sender loop isn't blocked waiting on a broker round-trip for every event.

Gotchas: the stream handle is backed by a linked process, so if it exits mid-batch, calls against it raise rather than fail quietly — this sample defensively falls back to send_event/2 per event when that happens, which loses the throughput benefit but keeps the batch moving. Events are still fire-and-forget pub/sub underneath: no subscriber means a streamed event is dropped just like a regular one. Process.flag(:trap_exit, true) is required so the stream process's exit is delivered as a message instead of crashing the caller.

Prerequisites

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

Code

Process.flag(:trap_exit, true)

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

{:ok, _sub} =
  KubeMQ.Client.subscribe_to_events(client, channel,
    on_event: fn event ->
      IO.puts("Received: #{event.body}")
      send(parent, :received)
    end
  )

Process.sleep(500)

{handle, stream_ok?} =
  try do
    case KubeMQ.Client.send_event_stream(client) do
      {:ok, h} ->
        IO.puts("Event stream opened")
        {h, true}
      _ ->
        {nil, false}
    end
  catch
    :exit, _ -> {nil, false}
  end

sent =
  Enum.reduce(1..5, 0, fn i, acc ->
    event = KubeMQ.Event.new(channel: channel, body: "Stream event #{i}")

    result =
      if stream_ok? do
        try do
          KubeMQ.EventStreamHandle.send(handle, event)
        catch
          :exit, _ -> {:error, :stream_dead}
        end
      else
        {:error, :no_stream}
      end

    case result do
      :ok ->
        IO.puts("Streamed event #{i}")
        Process.sleep(100)
        acc + 1
      _ ->
        try do
          KubeMQ.Client.send_event(client, event)
          IO.puts("Sent event #{i} (regular)")
        catch
          :exit, _ -> IO.puts("Sent event #{i} (fallback)")
        end
        Process.sleep(100)
        acc + 1
    end
  end)

IO.puts("Sent #{sent}/5 events")

received =
  Enum.reduce(1..sent, 0, fn _, acc ->
    receive do
      :received -> acc + 1
    after
      5_000 -> acc
    end
  end)

IO.puts("Received #{received}/#{sent} events")
IO.puts("Done.")
System.halt(0)

How It Works

• send_event_stream/1 opens a bidirectional gRPC stream for batching multiple events
• Stream sending avoids per-message connection overhead for high-throughput scenarios
• The code falls back to send_event/2 if the stream handle is unavailable
• Process.flag(:trap_exit, true) ensures clean shutdown of linked stream processes

Related

• Basic Pub/Sub
• Events Store Stream Send