Stream Send

Overview

Publishing events one at a time means each call pays its own round-trip: write the request, wait on the connection, then move to the next event. That's fine for occasional notifications, but it caps throughput when you need to push hundreds or thousands of events per second — log forwarding, sensor telemetry, change-data-capture feeds — where per-call overhead dominates.

client.createEventStream() opens one persistent bidirectional gRPC stream up front and returns a stream object. Each subsequent await stream.send(...) writes a frame directly onto that already-open connection instead of negotiating a new call, so a sender loop isn't blocked waiting on a broker round-trip for every event.

Gotchas: because sends don't wait on a per-message round-trip, write failures surface asynchronously through stream.onError() — register that handler before you send anything, or failures go unnoticed. Events are still fire-and-forget pub/sub underneath: no subscriber means a streamed event is dropped just like a regular one. Always call stream.close() when you're done; a stream left open holds a gRPC connection on the broker.

Prerequisites

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

Code

import { KubeMQClient, createEventMessage } from 'kubemq-js';

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

  try {
    const stream = client.createEventStream();
    stream.onError((err) => {
      console.error('Stream error:', err.message);
    });

    for (let i = 1; i <= 5; i++) {
      await stream.send(createEventMessage({ channel: 'js-events.stream-send', body: `event #${i}` }));
      console.log('Sent event', i);
    }

    await new Promise((r) => setTimeout(r, 500));
    stream.close();
    console.log('Stream closed');
  } finally {
    await client.close();
  }
}

main().catch(console.error);

How It Works

• createEventStream() opens a persistent bidirectional gRPC stream; subsequent send() calls reuse it without per-message connection overhead.
• stream.onError() registers an error handler before any sends — if the stream breaks mid-batch, the handler fires rather than silently dropping events.
• Each await stream.send(...) waits for the write to be flushed to the gRPC stream, preserving send order across the loop.
• stream.close() sends a half-close signal to the server; client.close() then tears down the underlying gRPC channel after all in-flight callbacks complete.

Related

• Pattern overview
• Node.js SDK Reference
• Basic Pub/Sub
• Cancel Subscription