lhisp-eventbus-consumer

PT-BR | EN

Português

Biblioteca TypeScript para consumir eventos via RabbitMQ (AMQP) usando amqplib.

Ela fornece:

• LhispEventbus: utilitário leve para conectar, criar channels, assertExchange e publishMessage.
• LhispEventbusConsumer: classe abstrata que configura exchange/queue/bind/consume e delega o processamento para o seu método consume().

Instalação

npm i lhisp-eventbus-consumer

Importação

import {
  LhispEventbus,
  LhispEventbusConsumer,
  type LhispEventbusConsumerConstructorParams,
  type LhispBaseEvent,
  type EventBusMessage,
  type Logger,
} from "lhisp-eventbus-consumer";

Conceitos

• Exchange de entrada: onde sua aplicação consome eventos.
• Queue: fila que será vinculada (bind) à exchange.
• Routing key / pattern: padrão usado no bindQueue.
  • Por padrão, queuePattern = "".
• Ack/Nack:
  • Por padrão, noAck = true (RabbitMQ considera a mensagem entregue sem necessidade de ack).
  • Se noAck = false, a classe dá ack quando o processamento termina sem erro; em erro/timeout, dá nack.
• Prefetch (paralelismo):
  • Se maxParallelProcesses > 0, é aplicado channel.prefetch(maxParallelProcesses).
• Serialização por stream (opcional):
  • Se serialProcessing = true, o consumer processa mensagens em série por stream id.
  • O stream id é definido sobrescrevendo getSerialProcessingStreamId(acao, evento, msg).
  • Sem stream id (retorno undefined), o comportamento continua paralelo.
• Reconexão automática:
  • Quando a conexão ou um canal cai (queda do broker, ECONNRESET, restart do RabbitMQ), o consumer reabre conexão, canais, exchange, fila, bind e consume sozinho.
  • As tentativas usam backoff exponencial de reconnectDelayMs até reconnectMaxDelayMs.
  • Mensagens não confirmadas (noAck = false) voltam para a fila e são reentregues pelo broker.
  • Use stop() no shutdown da aplicação para encerrar sem disparar reconexão.

Parâmetros do consumer

A classe LhispEventbusConsumer recebe um objeto com:

• eventBusUrl (obrigatório): URL do AMQP, por exemplo amqp://user:pass@host:5672.
• exchangeName (obrigatório): exchange de entrada.
• queueName (obrigatório): nome da fila.
• exchangeType (opcional, default fanout).
• exchangeOptions (opcional, default { durable: true }).
• queuePattern (opcional, default "").
• queueOptions (opcional, default { durable: true, exclusive: false, autoDelete: false }).
• noAck (opcional, default true).
• maxParallelProcesses (opcional, default 0).
• serialProcessing (opcional, default false): habilita processamento serial por stream id customizado.
• consumeTimeout (opcional, default 60000 ms): timeout do processamento por mensagem.
• reconnect (opcional, default true): reconexão automática quando a conexão/canal cai.
• reconnectDelayMs (opcional, default 1000 ms): atraso da primeira tentativa, dobrado a cada falha.
• reconnectMaxDelayMs (opcional, default 30000 ms): teto do atraso entre tentativas.
• Callback/Status exchange (opcionais):
  • callbackModulo (default ""): se definido, publica um evento de callback quando ocorrer erro no consumo.
  • callbackExchangeName (default DaemonStatus).
  • callbackExchangeType (default fanout).
• Sync exchange (opcionais):
  • syncExchangeName (default SyncEvents).
  • syncExchangeType (default topic).
• logger (opcional): precisa implementar a interface Logger.

Exemplo: criando um consumer

Crie uma classe que estende LhispEventbusConsumer e implemente:

• consume(acao, evento, msg, logger): processamento do evento.
• handleChannelError(error): handler de erro do channel.

import { LhispEventbusConsumer, type EventBusMessage, type LhispBaseEvent, type Logger } from "lhisp-eventbus-consumer";

interface UserCreatedEvent extends LhispBaseEvent {
  payload: {
    id: string;
    email: string;
  };
}

class UserCreatedConsumer extends LhispEventbusConsumer<UserCreatedEvent> {
  protected handleChannelError(error: any): void {
    this.logger?.error({ message: "Channel Error", error });
  }

  protected async consume(
    acao: string,
    evento: UserCreatedEvent,
    _msg: EventBusMessage,
    logger: Logger,
  ): Promise<void> {
    logger.info({ message: "Consuming event", acao, evento });

    // Seu processamento aqui
    // - use `acao` (routing key)
    // - use `evento.dbname`, `evento.EmpresaId`, `evento.payload`

    // Se você precisar responder eventos síncronos:
    // if (evento.uuid) await this.replySyncEvent(evento.uuid, { ok: true });
  }

  protected getSerialProcessingStreamId(_acao: string, evento: UserCreatedEvent): string | undefined {
    const empresaId = evento?.payload?.EmpresaId;
    const sva = evento?.payload?.sva;
    if (!empresaId || !sva) return undefined;
    return `${empresaId}_${sva}`;
  }
}

async function main() {
  const consumer = new UserCreatedConsumer({
    eventBusUrl: process.env.EVENTBUS_URL ?? "amqp://localhost:5672",
    exchangeName: "UserEvents",
    queueName: "user-created-consumer",
    queuePattern: "user.created",
    noAck: false,
    maxParallelProcesses: 10,
    serialProcessing: true,
    consumeTimeout: 60_000,

    // Opcional: publicar callback em caso de erro
    // callbackModulo: "my-service",
    // callbackExchangeName: "DaemonStatus",
  });

  consumer.banner();
  await consumer.start();
}

main().catch((err) => {
  console.error(err);
  process.exit(1);
});

Exemplo: publicando mensagens

A classe LhispEventbus pode ser usada para publicar mensagens (fire-and-forget) em uma exchange.

import { LhispEventbus } from "lhisp-eventbus-consumer";

async function publishExample() {
  const bus = new LhispEventbus({
    eventBusUrl: process.env.EVENTBUS_URL ?? "amqp://localhost:5672",
  });

  await bus.publishMessage("UserEvents", "user.created", {
    dbname: "tenant_db",
    EmpresaId: "1",
    payload: { id: "123", email: "[email protected]" },
  });
}

Scripts

• Rodar testes: npm test
• Rodar testes em watch: npm run test:watch
• Build: npm run build

English

TypeScript library to consume RabbitMQ (AMQP) events using amqplib.

It provides:

• LhispEventbus: a small helper to connect, create channels, assertExchange, and publishMessage.
• LhispEventbusConsumer: an abstract class that sets up exchange/queue/bind/consume and delegates processing to your consume() method.

Installation

npm i lhisp-eventbus-consumer

Import

import {
  LhispEventbus,
  LhispEventbusConsumer,
  type LhispEventbusConsumerConstructorParams,
  type LhispBaseEvent,
  type EventBusMessage,
  type Logger,
} from "lhisp-eventbus-consumer";

Concepts

• Input exchange: where your application consumes events from.
• Queue: the queue that will be bound to the exchange.
• Routing key / pattern: pattern used in bindQueue.
  • Default is queuePattern = "".
• Ack/Nack:
  • Default is noAck = true (RabbitMQ auto-acknowledges delivery).
  • If noAck = false, the consumer will ack when processing succeeds; on error/timeout it will nack.
• Prefetch (parallelism):
  • If maxParallelProcesses > 0, channel.prefetch(maxParallelProcesses) is applied.
• Stream-based serialization (optional):
  • If serialProcessing = true, the consumer processes messages sequentially per stream id.
  • Stream id is defined by overriding getSerialProcessingStreamId(action, event, msg).
  • If stream id is undefined, messages keep default parallel behavior.
• Automatic reconnection:
  • When the connection or a channel drops (broker outage, ECONNRESET, RabbitMQ restart), the consumer reopens connection, channels, exchange, queue, bind and consume on its own.
  • Retries use exponential backoff from reconnectDelayMs up to reconnectMaxDelayMs.
  • Unacked messages (noAck = false) are requeued and redelivered by the broker.
  • Call stop() on application shutdown to close without triggering a reconnect.

Consumer parameters

LhispEventbusConsumer constructor receives:

• eventBusUrl (required): AMQP URL, e.g. amqp://user:pass@host:5672.
• exchangeName (required): input exchange name.
• queueName (required): queue name.
• exchangeType (optional, default fanout).
• exchangeOptions (optional, default { durable: true }).
• queuePattern (optional, default "").
• queueOptions (optional, default { durable: true, exclusive: false, autoDelete: false }).
• noAck (optional, default true).
• maxParallelProcesses (optional, default 0).
• serialProcessing (optional, default false): enables serial processing by custom stream id.
• consumeTimeout (optional, default 60000 ms): per-message processing timeout.
• reconnect (optional, default true): automatic reconnection when connection/channel drops.
• reconnectDelayMs (optional, default 1000 ms): first retry delay, doubled on each failure.
• reconnectMaxDelayMs (optional, default 30000 ms): cap for the delay between retries.
• Callback/Status exchange (optional):
  • callbackModulo (default ""): when provided, publishes a callback event when consumption fails.
  • callbackExchangeName (default DaemonStatus).
  • callbackExchangeType (default fanout).
• Sync exchange (optional):
  • syncExchangeName (default SyncEvents).
  • syncExchangeType (default topic).
• logger (optional): must implement the Logger interface.

Example: creating a consumer

Create a class extending LhispEventbusConsumer and implement:

• consume(action, event, msg, logger): your message handler.
• handleChannelError(error): channel error handler.

import { LhispEventbusConsumer, type EventBusMessage, type LhispBaseEvent, type Logger } from "lhisp-eventbus-consumer";

interface UserCreatedEvent extends LhispBaseEvent {
  payload: {
    id: string;
    email: string;
  };
}

class UserCreatedConsumer extends LhispEventbusConsumer<UserCreatedEvent> {
  protected handleChannelError(error: any): void {
    this.logger?.error({ message: "Channel Error", error });
  }

  protected async consume(
    action: string,
    event: UserCreatedEvent,
    _msg: EventBusMessage,
    logger: Logger,
  ): Promise<void> {
    logger.info({ message: "Consuming event", action, event });

    // Your business logic here.

    // If you need to reply a sync request:
    // if (event.uuid) await this.replySyncEvent(event.uuid, { ok: true });
  }

  protected getSerialProcessingStreamId(_action: string, event: UserCreatedEvent): string | undefined {
    const empresaId = event?.payload?.EmpresaId;
    const sva = event?.payload?.sva;
    if (!empresaId || !sva) return undefined;
    return `${empresaId}_${sva}`;
  }
}

async function main() {
  const consumer = new UserCreatedConsumer({
    eventBusUrl: process.env.EVENTBUS_URL ?? "amqp://localhost:5672",
    exchangeName: "UserEvents",
    queueName: "user-created-consumer",
    queuePattern: "user.created",
    noAck: false,
    maxParallelProcesses: 10,
    consumeTimeout: 60_000,

    // Optional: publish callback events on error
    // callbackModulo: "my-service",
    // callbackExchangeName: "DaemonStatus",
  });

  consumer.banner();
  await consumer.start();
}

main().catch((err) => {
  console.error(err);
  process.exit(1);
});

Example: publishing messages

Use LhispEventbus for fire-and-forget publishing.

import { LhispEventbus } from "lhisp-eventbus-consumer";

async function publishExample() {
  const bus = new LhispEventbus({
    eventBusUrl: process.env.EVENTBUS_URL ?? "amqp://localhost:5672",
  });

  await bus.publishMessage("UserEvents", "user.created", {
    dbname: "tenant_db",
    EmpresaId: "1",
    payload: { id: "123", email: "[email protected]" },
  });
}

Scripts

• Run tests: npm test
• Run tests (watch): npm run test:watch
• Build: npm run build