RabbitMQ Stream Client for JavaScript/TypeScript

Common errors

• Error: Code 83 (NOT_SUPPORTED_FORMAT) when message is published using amqplib

  cause Messages published via the AMQP 0-9-1 protocol (e.g., using `amqplib`) might not be compatible with the RabbitMQ Stream protocol due to differences in message formatting and properties.
  fix Ensure all messages intended for RabbitMQ Streams are published using `rabbitmq-stream-js-client`. Do not mix publishing messages to streams using `amqplib` or other AMQP 0-9-1 clients, as the stream protocol has its own specific message structure.

• Error: connect ECONNREFUSED 127.0.0.1:5552

  cause The client could not establish a connection to the RabbitMQ Stream server. This typically means the server is not running, is not listening on the specified host/port, or a firewall is blocking the connection.
  fix Verify that your RabbitMQ server is running and the Stream plugin is enabled. Check the `hostname` and `port` (default 5552 for streams) in your connection configuration. Ensure no firewall rules are preventing the connection.

• TypeError: rabbit.connect is not a function

  cause You are likely attempting to use a CommonJS `require` syntax in a TypeScript or ESM project without correctly destructuring the module's named exports.
  fix Use ESM named imports: `import { connect } from 'rabbitmq-stream-js-client';`. If you must use `require`, destructure it: `const { connect } = require('rabbitmq-stream-js-client');`.

Warnings

• breaking The `Connection closed` listener is no longer called by the client. Applications relying on this specific event for connection state management will need to adjust.
  Fix: Review connection management strategies. The client now handles auto-reconnecting internally. Refer to updated README examples for managing connection stability, e.g., how to handle auto-reconnects as shown in the v0.4.0 changelog.
• gotcha When using message deduplication with `publisherRef` or `publishingId`, it is solely the user's responsibility to guarantee the uniqueness and incremental order of these IDs. RabbitMQ Stream does not enforce or manage these across producers, which can lead to message loss or unexpected deduplication if not handled correctly.
  Fix: Implement a robust strategy for generating unique and monotonic `publishingId`s, potentially using a shared sequence generator or UUIDs with external storage for persistence across application restarts.
• gotcha The `ca` parameter for SSL/TLS connections is optional. If not provided, the client will rely on the system's default CAs or accept self-signed certificates depending on Node.js configuration, which might be a security risk in production environments.
  Fix: Always provide the `ca` (Certificate Authority) certificate when connecting through TLS/SSL in production environments to ensure proper certificate chain validation and secure communication.
• breaking Internal refactoring around connection pooling (e.g., connection pool as an instance, managing connections inside the pool) in v0.6.2 might affect applications that previously attempted to interfere with or manage underlying connections directly.
  Fix: Rely on the client's public API for connection management. Avoid direct manipulation of internal connection objects, as the library now explicitly manages connection lifecycles within its pool.

Install

• npm install rabbitmq-stream-js-client npm
• yarn add rabbitmq-stream-js-client yarn
• pnpm add rabbitmq-stream-js-client pnpm

Imports

• connect
  wrong:

  const rabbit = require('rabbitmq-stream-js-client');
  const client = await rabbit.connect(...);

  correct:

  import { connect } from 'rabbitmq-stream-js-client';

  For modern TypeScript and ESM projects, prefer named imports. The 'require' style is shown in older examples but may cause issues with type inference or bundlers.
• StreamClient

  import { StreamClient } from 'rabbitmq-stream-js-client';

  While 'connect' is the primary entry point, importing StreamClient is useful for type annotations or if you need to extend its functionality.
• Publisher

  import { Publisher } from 'rabbitmq-stream-js-client';

  Publisher instances are typically created via `client.declarePublisher()`, but the type can be imported for explicit typing.

Quickstart

Demonstrates connecting to RabbitMQ Stream, declaring a stream, publishing a message, and consuming it.

import { connect } from 'rabbitmq-stream-js-client';
import { Buffer } from 'buffer'; // Node.js Buffer is needed for message content

async function runRabbitStreamExample() {
  const client = await connect({
    hostname: process.env.RABBITMQ_HOST ?? 'localhost',
    port: parseInt(process.env.RABBITMQ_STREAM_PORT ?? '5552', 10),
    username: process.env.RABBITMQ_USERNAME ?? 'rabbit',
    password: process.env.RABBITMQ_PASSWORD ?? 'rabbit',
    vhost: process.env.RABBITMQ_VHOST ?? '/',
  });

  const streamName = 'my-first-stream';

  // Declare a stream (if it doesn't exist, it will be created)
  await client.declareStream({ stream: streamName });

  const publisher = await client.declarePublisher({
    stream: streamName,
    publisherRef: 'my-publisher-ref',
  });

  const messageContent = Buffer.from('Hello, RabbitMQ Stream!');
  await publisher.send(messageContent);
  console.log('Message published:', messageContent.toString());

  const consumer = await client.createConsumer(
    {
      stream: streamName,
      consumerRef: 'my-consumer-ref',
      offset: 'first',
    },
    (message) => {
      console.log('Message consumed:', message.getData().toString());
      consumer.close(); // Close consumer after first message for this example
    }
  );

  // Give some time for the consumer to receive the message
  await new Promise(resolve => setTimeout(resolve, 1000));

  await publisher.close();
  await client.close();
  console.log('Client and publisher closed.');
}

runRabbitStreamExample().catch(console.error);