Stream Receive

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

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

Code

/**
 * 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

• 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

• Pattern overview
• Node.js SDK Reference
• Send & Receive
• Ack All