# Stream Send (/sdks/elixir/how-to/queues/stream-send)



## Overview [#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 [#prerequisites]

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

## Code [#code]

```elixir title="main.exs"
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 [#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 [#related]

* [Batch Send](/sdks/elixir/how-to/queues/batch-send)
* [Stream Receive](/sdks/elixir/how-to/queues/stream-receive)