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



## Overview [#overview]

A **downstream receiver** is the persistent-connection way to pull queue messages: instead of opening and tearing down a request for every batch, you open one gRPC stream and reuse it across many poll cycles. That matters for any consumer that runs continuously — a worker loop, a background processor — where reconnecting per batch would add latency and churn on both the client and the broker.

The receiver is created once with `create_downstream_receiver`, then each call to `poll` fetches a batch under a transaction; the returned `QueuePollResponse` supports `ack`/`nack`/`requeue` per message, so nothing is removed from the queue until you explicitly settle it. Calling `ack` on a message permanently removes it, while leaving it unsettled returns it for redelivery once the visibility timeout expires.

**Gotchas:** an unclosed receiver holds server-side state — always `close` it (as the `ensure` block does here); a crash between polling and acknowledging redelivers the batch, so processing must be idempotent; and `requeue` versus `nack` differ in whether the message goes back immediately or through the broker's redelivery/dead-letter path — pick deliberately.

## Prerequisites [#prerequisites]

* KubeMQ server running on `localhost:50000`
* Ruby SDK installed (`gem install kubemq`)

## Code [#code]

```ruby title="main.rb"
require 'kubemq'

address = ENV.fetch('KUBEMQ_ADDRESS', 'localhost:50000')
channel = 'ruby-queues.stream-receive'

begin
  client = KubeMQ::QueuesClient.new(address: address, client_id: 'qstream-recv-example')
  puts "Connected to #{address}"

  3.times do |i|
    msg = KubeMQ::Queues::QueueMessage.new(channel: channel, metadata: "item-#{i}", body: "data-#{i}")
    client.send_queue_message(msg)
  end
  puts 'Pre-sent 3 messages'

  receiver = client.create_downstream_receiver
  request = KubeMQ::Queues::QueuePollRequest.new(channel: channel, max_items: 5, wait_timeout: 5)
  response = receiver.poll(request)

  if response.error?
    puts "Poll error: #{response.error}"
  else
    puts "Polled #{response.messages.size} messages:"
    response.messages.each do |m|
      puts "  #{m.metadata}: #{m.body}"
      m.ack
      puts '  -> acked'
    end
  end
rescue KubeMQ::Error => e
  puts "KubeMQ error: #{e.message}"
ensure
  receiver&.close
  client&.close
  puts 'Done'
end
```

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

* `create_downstream_receiver` opens a persistent gRPC stream for receiving messages.
* `poll` returns a `QueuePollResponse` with transactional `ack`/`nack`/`requeue` support per message.
* Close the receiver when finished.
* Review timeouts, channel names, and client IDs before running against shared environments.

## Related [#related]

* [Pattern overview](/learn/queues/getting-started)
* [Ruby SDK Reference](/sdks/ruby/reference)
* [Stream Send](/sdks/ruby/how-to/queues/stream-send)
* [Poll Mode](/sdks/ruby/how-to/queues/poll-mode)