Stream Send

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.

client.sendQueueMessage() reuses the SDK's internal gRPC channel across calls rather than reconnecting per message, so repeated sends stay cheap. tags on each message let downstream consumers filter or route by sensor ID or batch without parsing the body. For higher throughput than a sequential await loop, the SDK also exposes sendQueueMessagesBatch() for one-shot batches and createQueueUpstream() for a persistent upload stream that pipelines many batches without waiting on each result.

Gotchas: an await-in-a-loop sends one at a time, bounded by round-trip latency — for real high-volume ingestion prefer sendQueueMessagesBatch() or createQueueUpstream(). Always await client.close() in a finally block; an unclosed client leaks the connection. Check each result for errors — a broker-side rejection doesn't throw, so a message can silently fail to enqueue if you skip inspecting it.

Prerequisites

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

Code

/**
 * Example: Stream Upstream (Send via Stream)
 *
 * Demonstrates sending messages through a persistent gRPC stream for
 * high-throughput queue ingestion. The stream avoids per-message
 * connection overhead.
 *
 * Prerequisites:
 *   - KubeMQ server running on localhost:50000
 *
 * Run: npx tsx examples/queues-stream/stream-send.ts
 */
import { KubeMQClient, createQueueMessage } from 'kubemq-js';

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

  try {
    // Send multiple messages — the SDK reuses the underlying gRPC stream.
    for (let i = 1; i <= 10; i++) {
      await client.sendQueueMessage(
        createQueueMessage({
          channel: 'js-queues-stream.stream-send',
          body: `Sensor reading #${i}: temp=${(20 + Math.random() * 10).toFixed(1)}°C`,
          tags: { sensor: 'temp-01', batch: 'stream-demo' },
        }),
      );
    }

    console.log('Sent 10 messages via stream');
  } finally {
    await client.close();
  }
}

main().catch(console.error);

How It Works

• Each sendQueueMessage() call in the loop uses the SDK's internal connection pool; the SDK reuses the underlying gRPC channel rather than opening a new connection per message.
• 10 simulated sensor readings are sent sequentially with await; for even higher throughput consider batching with sendQueueMessagesBatch() or using createQueueUpstream() for a true persistent upload stream.
• tags on each message allow downstream consumers to filter or route by sensor ID or batch without parsing the body.
• Run stream-receive.ts on the same channel to consume these messages with per-message ack/nack control.

Related

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