# Stream Receive (/sdks/nodejs/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 receive 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.

`receiveQueueMessages()` fetches a batch under manual settlement — messages come back locked, not auto-removed — so nothing leaves the queue until you explicitly settle it. Each returned message is settled on its own: calling `msg.ack()` on success permanently removes it, while `msg.nack()` on failure returns it to the queue for redelivery once the visibility timeout expires.

**Gotchas:** a crash between receiving and settling redelivers the whole batch, so processing must be idempotent; forgetting to settle just delays redelivery until the timeout, it doesn't drop the message; and routing every exception to `nack()`, as shown here, retries poison messages forever unless paired with a dead-letter policy or a retry-count check.

## Prerequisites [#prerequisites]

* KubeMQ server running on `localhost:50000`
* Node.js SDK installed (`npm install kubemq-js`)

## Code [#code]

```typescript title="stream-receive.ts"
/**
 * Example: Stream Downstream (Receive with Ack/Reject/Requeue)
 *
 * Demonstrates receiving messages from a queue with fine-grained control
 * over each message: acknowledge on success, reject on permanent failure,
 * or requeue for later retry.
 *
 * Prerequisites:
 *   - KubeMQ server running on localhost:50000
 *   - Messages in the "js-queues-stream.stream-receive" channel (run stream-send.ts first)
 *
 * Run: npx tsx examples/queues-stream/stream-receive.ts
 */
import { KubeMQClient } from 'kubemq-js';

async function main(): Promise<void> {
  const client = await KubeMQClient.create({
    address: 'localhost:50000',
    clientId: 'js-queues-stream-stream-receive-client',
  });

  try {
    const messages = await client.receiveQueueMessages({
      channel: 'js-queues-stream.stream-receive',
      waitTimeoutSeconds: 10,
      maxMessages: 5,
    });

    console.log(`Received ${messages.length} messages`);

    for (const msg of messages) {
      const body = new TextDecoder().decode(msg.body);
      console.log(`Processing: ${body}`);

      try {
        // Simulate processing.
        await processMessage(body);
        await msg.ack();
        console.log('  ✓ Acknowledged');
      } catch {
        // On failure, reject the message.
        await msg.nack();
        console.log('  ✗ Rejected');
      }
    }
  } finally {
    await client.close();
  }
}

async function processMessage(_body: string): Promise<void> {
  // Simulate occasional failures.
  if (Math.random() < 0.2) {
    throw new Error('Processing failed');
  }
  await new Promise((resolve) => setTimeout(resolve, 100));
}

main().catch(console.error);

```

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

* `receiveQueueMessages()` with `maxMessages: 5` and `waitTimeoutSeconds: 10` fetches up to 5 locked messages, waiting up to 10 seconds if the queue is initially empty.
* Each message is processed individually: `await msg.ack()` on success removes it permanently; `await msg.nack()` on failure returns it to the queue for redelivery.
* The `try/catch` around `processMessage()` routes success to `ack` and any exception to `nack`, demonstrating at-least-once processing with explicit failure handling.
* `processMessage()` simulates a 20% failure rate using `Math.random()`; in production replace this with real business logic and domain-specific error handling.

## Related [#related]

* [Pattern overview](/learn/queues/getting-started)
* [Node.js SDK Reference](/sdks/nodejs/reference)
* [Send & Receive](/sdks/nodejs/tutorials/send-receive)
* [Ack All](/sdks/nodejs/how-to/queues/ack-all)