# Stream Receive (/sdks/elixir/how-to/queues/stream-receive)



## Overview [#overview]

A **downstream receiver** is the transactional way to pull queue messages: rather than a plain request/response fetch, it's backed by a gRPC stream and gives you an explicit settlement step before a batch is removed from the queue. That matters for any consumer that needs "fetch, process, then confirm" instead of "fetch and it's already gone."

`KubeMQ.Client.poll_queue/2` fetches a batch under a transaction — nothing is removed from the queue until you explicitly settle it. Acknowledging the whole batch with `KubeMQ.PollResponse.ack_all/1` permanently removes every message in one round-trip; until that call succeeds, the messages stay invisible to other consumers but are not yet gone.

**Gotchas:** a crash between polling and calling `ack_all/1` redelivers the whole batch once the visibility timeout expires, so processing must be idempotent; `ack_all/1` is all-or-nothing — a single bad message in the batch means you either ack everything (including messages you couldn't process) or nothing; and `wait_timeout` bounds how long the poll blocks on an empty queue, not how long processing may take afterward.

## 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-receive"
{:ok, client} = KubeMQ.Client.start_link(address: "localhost:50000", client_id: "elixir-qs-recv")

for i <- 1..3 do
  {:ok, _} = KubeMQ.Client.send_queue_message(client,
    KubeMQ.QueueMessage.new(channel: channel, body: "Queued item #{i}"))
end

IO.puts("Sent 3 messages")

case KubeMQ.Client.poll_queue(client,
       channel: channel,
       max_items: 10,
       wait_timeout: 5_000
     ) do
  {:ok, poll} ->
    IO.puts("Polled #{length(poll.messages)} messages")

    Enum.each(poll.messages, fn msg ->
      IO.puts("  Channel: #{msg.channel}")
      IO.puts("  Body: #{msg.body}")
    end)

    {:ok, _} = KubeMQ.PollResponse.ack_all(poll)
    IO.puts("All messages acknowledged")

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

KubeMQ.Client.close(client)
```

## How It Works [#how-it-works]

* `poll_queue/2` uses the downstream streaming API to poll for messages with transactional control — distinct from the plain `receive_queue_messages/3` pull, which has no settlement step
* Messages must be explicitly acknowledged via `PollResponse.ack_all/1`
* Until acknowledged, messages remain invisible to other consumers

## Related [#related]

* [Stream Send](/sdks/elixir/how-to/queues/stream-send)
* [Auto Ack](/sdks/elixir/how-to/queues/auto-ack)
* [Poll Mode](/sdks/elixir/how-to/queues/poll-mode)