Examples

Example 1

import { MessagingPlugin } from '@setu-ts/messaging-plugin';
import { CAPABILITIES } from '@setu-ts/common';

// GCP Pub/Sub
app.register(MessagingPlugin({
  broker: 'pubsub',
  projectId: 'my-project',
}));

// Azure Service Bus
app.register(MessagingPlugin({
  broker: 'service-bus',
  connectionString: 'Endpoint=sb://...',
}));

// Custom broker
app.register(MessagingPlugin({
  broker: 'custom',
  instance: myBroker,
}));

Classes

c
ChainGateTimeoutError(timeoutMs: number)

Thrown when a delivery held on the ingress behaviour-chain gate (PipelinedBroker) waits longer than the configured chainReadyTimeoutMs (default 10 000 ms; 0 waits forever) for the behaviour chain to open.

c
CloudBrokerUnavailableError(
backend: string,
specifier: string
)

Thrown by a cloud broker's IMessageBroker.connect when the runtime platform is Cloudflare Workers and the SDK cannot function (gRPC, AMQP, or long-poll — not fetch). The throw fails app.start() at the earliest possible point.

c
GcpPubSubBroker(
runtime: IRuntimeServices,
serializer: ISerializer,
options?: PubSubOptions
)

GCP Pub/Sub message broker.

c
InMemoryBroker(
runtime: IRuntimeServices,
serializer: ISerializer,
options?: InMemoryBrokerOptions
)

In-memory message broker implementation.

c
IntegrationEventRejectedError(details: { reason: IntegrationEventRejectionReason; topic: string; expectedType: string; expectedVersion: number; detail: string; cause?: unknown; })

Thrown by onIntegrationEvent's wrapper when a delivered message is refused before the application handler runs. One class rather than four keeps the consumer's instanceof branch a single import, while reason discriminates.

c
JetStreamStreamError(
stream: string,
cause?: unknown
)

Thrown by NatsBroker.connect when the JetStream stream could not be ensured: the stream is absent and NatsOptions.streamSubjects was not supplied, or the platform refused the stream read/create. The platform's own error is carried as cause when there is one.

c
JetStreamUnavailableError(cause: unknown)

Thrown by NatsBroker.connect when the NATS server rejects the JetStream manager probe — the server does not have JetStream enabled, which the nats broker requires. The platform's own error (typically the raw 503 / NO_RESPONDERS reply for the $JS.API subjects) is carried as cause.

c
JsonSerializer

JSON serializer implementation for message payloads.

c
KafkaBroker(
runtime: IRuntimeServices,
serializer: ISerializer,
options?: KafkaOptions
)

Kafka message broker implementation.

c
NatsBroker(
runtime: IRuntimeServices,
serializer: ISerializer,
options?: NatsOptions
)

NATS JetStream message broker implementation.

c
RabbitMqBroker(
runtime: IRuntimeServices,
serializer: ISerializer,
options?: RabbitMqOptions
)

RabbitMQ message broker implementation using AMQP 0-9-1 topic exchange.

c
RedisStreamsBroker(
runtime: IRuntimeServices,
serializer: ISerializer,
options?: RedisStreamsOptions
)

Redis Streams message broker implementation.

c
RemoteHandlerError(remoteMessage: string)

Thrown by IMessageBroker.request when the remote responder threw while handling the request. The responder's error message is propagated back to the caller in remoteMessage.

c
ReplyInboxUnavailableError(topic: string)

Thrown by GcpPubSubBroker and ServiceBusBroker when the per-instance RPC reply subscription cannot be created (missing Manage right or the reply topic does not exist).

c
RequestTimeoutError(message?: string)

Thrown by IMessageBroker.request when no correlated reply arrives within the configured timeoutMs window. The pending request is abandoned and its correlation entry cleaned up; a reply that arrives afterwards is dropped.

c
ServiceBusBroker(
runtime: IRuntimeServices,
serializer: ISerializer,
options?: ServiceBusOptions
)

Azure Service Bus message broker.

c
MessagingNotSupportedError(message?: string)

Signals that a broker's transport cannot support brokered request-reply.

Functions

f
causedBy(envelope: IntegrationEventEnvelope): { correlationId: string; causationId: string; }

Derives the causal metadata a consumed envelope contributes to the event its handling publishes next — the whole of the chain-root rule, extracted so no handler copies it by hand.

f
loadPubSubModule(): Promise<PubSubSdkModule>

Lazily load the GCP Pub/Sub SDK.

f
loadServiceBusModule(): Promise<ServiceBusSdkModule>

Lazily load the Azure Service Bus SDK.

f
onIntegrationEvent<T>(
definition: IntegrationEventDefinition<T>,
handler: IntegrationEventHandler<T>,
options?: SubscribeOptions
): SubscriptionDefinition

Produces a SubscriptionDefinition for an integration-event contract — the declarative form, plugging straight into MessagingPlugin({ subscriptions }), or spread by hand into an imperative broker.subscribe after start().

f
publishIntegrationEvent<T>(
runtime: IRuntimeServices,
broker: IMessageBroker,
definition: IntegrationEventDefinition<T>,
payload: T,
metadata?: IntegrationEventMetadata
): Promise<void>

Publishes one integration event: builds the envelope from the caller's already-typed payload and hands it to broker.publish on the definition's topic.

Interfaces

I
EventsMessagingBridgeOptions

Options for the EventsMessagingBridge factory.

I
IMessageBroker

Message broker for cross-service integration events.

I
INatsHeaders

Public members used from NATS MsgHdrs.

I
InMemoryBrokerOptions

Options for the in-memory broker — the adapter behind the memory arm of MessagingPluginOptions, constructed by it and by applications that build InMemoryBroker directly.

  • onDispatchError: (
    error: unknown,
    metadata: MessageMetadata
    ) => void | Promise<void>

    Called once per REJECTED subscription handler, with the error and the message metadata of the failed dispatch. publish resolves on dispatch hand-off — never on handler completion — so this reporter is the terminus of the broker's failure path: the in-memory broker has no ack model and no redelivery to fall back on (unlike RabbitMQ, where a rejection reaching the broker's failure path can nack and redeliver). Absent, the rejection is still observed and settled — never an unhandled rejection — then dropped. A reporter that itself throws or rejects is swallowed by the broker: it is the last-resort sink, so its own failure can neither reject publish nor abort the sibling fan-out nor surface as an unhandled rejection. MessagingPlugin always supplies one backed by the application's logger, so the absent case is reachable only by constructing the broker directly.

I
IntegrationEventDefinition

A named, versioned cross-service event contract.

  • parse: (value: unknown) => T

    Narrows the delivered payload for the application handler. Runs on the consumer side only — never on publish.

  • topic: string

    The transport topic the event is published to and consumed from.

  • type: string

    The event's semantic name, carried in the envelope's type field.

  • version: number

    The contract version. A bump is a breaking payload change.

I
IntegrationEventEnvelope

The wire shape of a published integration event.

I
IntegrationEventMetadata

Optional causal metadata for publishIntegrationEvent.

I
IPubSubSubscription

Handle for an open Pub/Sub subscription.

I
IPubSubTransport

Domain port for GCP Pub/Sub operations. The broker depends on this, not the SDK directly.

I
ISerializer

Serializer contract for converting messages to/from string payloads.

I
IServiceBusProcessErrorArgs

Structural type matching the real SDK's ProcessErrorArgs callback argument (npm:@azure/service-bus@^7).

I
IServiceBusSubscribeOptions

Structural receive-options matching the real SDK's SubscribeOptions (npm:@azure/service-bus@^7). The property is autoCompleteMessages, not autoComplete.

I
IServiceBusSubscription

Handle for an open Service Bus subscription receiver.

I
ISubscription

An active subscription.

I
KafkaOptions

Kafka-specific options (internal use).

I
MemoryMessagingOptions

Default (memory) arm. The discriminant is optional so that MessagingPlugin() and MessagingPlugin({}) remain valid.

I
MessageMetadata

Transport metadata accompanying a delivered message.

I
MessagingCommonOptions

Shared options present on every MessagingPluginOptions arm.

  • behaviors: readonly (IIngressBehavior | RegistryFactory<IIngressBehavior>)[]

    Ingress behaviours wrapped around every subscription handler — the messaging arm of the transport-neutral behaviour chain shared with the websocket, queue, and scheduler plugins (IIngressBehavior in @setu-ts/common).

  • chainReadyTimeoutMs: number

    Bounds a dispatch held on the behaviour-chain gate, which exists only when a RegistryFactory behaviour is declared. A held dispatch that waits longer than this rejects with ChainGateTimeoutError, whose message names the likely cause (a plugin publishing during its own register()); the gate itself is left in place, so later dispatches refuse the same way rather than delivering through a partial chain.

  • name: string

    Instance name for multi-instance support.

  • serializer: ISerializer

    Serializer for message payloads.

  • subscriptions: readonly SubscriptionEntry[]

    Subscriptions registered declaratively, as an alternative to calling broker.subscribe(topic, handler, options) imperatively after start(). Each entry — instance or RegistryFactory — produces one subscribe() call, so a subscription can be declared where the plugin is composed instead of after the application has started.

  • tracing: boolean

    Whether to create producer and consumer spans when telemetry is available.

I
NatsMessagingOptions

NATS arm.

I
NatsOptions

NATS-specific options (internal use).

I
PubSubOptions

Options for GCP Pub/Sub broker.

I
RabbitMqOptions

RabbitMQ-specific options (internal use).

I
RedisStreamsOptions

Redis-specific options (internal use).

I
RequestOptions

Options accepted by IMessageBroker.request.

  • timeoutMs: number

    Reply wait budget in milliseconds. When no correlated reply arrives within this window, request rejects. Defaults to 5000 when omitted.

I
ServiceBusOptions

Options for Azure Service Bus broker.

I
ServiceBusRetryOptions

Azure Service Bus SDK retry budget for the data client (M90b / X28-6).

  • maxRetries: number

    Maximum number of retry attempts before an operation fails. The SDK default is 3; 0 disables retries entirely.

  • maxRetryDelayInMs: number

    Ceiling the exponential backoff grows to, in milliseconds. The SDK default is 90000.

  • mode: "fixed" | "exponential"

    Backoff curve. Translated to the SDK's numeric RetryMode before ServiceBusClient is constructed (the SDK compares the value with === against its enum). The SDK default when omitted is 'fixed'.

  • retryDelayInMs: number

    Delay before the first retry, in milliseconds. The SDK default is 30000.

  • timeoutInMs: number

    Whole-operation timeout, in milliseconds. The SDK default is 60000.

I
SubscribeOptions

Options accepted when subscribing to a topic.

  • queue: string

    Consumer group / queue name for load-balanced delivery.

I
SubscriptionDefinition

The declarative form of one IMessageBroker.subscribe() call — the entry an application writes instead of calling subscribe() imperatively after start().

Type Aliases

T
IntegrationEventRejectionReason = "malformed" | "type-mismatch" | "version-mismatch" | "parse"

Why an integration-event delivery was refused before the application handler ran. The four values are the four distinct producer-side faults an operator triaging a dead-letter needs to tell apart.

T
PubSubMessagingOptions =
PubSubMessagingOptionsInjected
| PubSubMessagingOptionsProduction

GCP Pub/Sub options — exclusive union of injected and production arms.

T
RequestHandler<TReq = unknown, TRes = unknown> = (
message: TReq,
metadata: MessageMetadata
) => TRes | Promise<TRes>

Responder for a request topic. Its resolved value is sent back to the caller as the reply, correlated to the originating request.

T
ServiceBusMessagingOptions =
ServiceBusMessagingOptionsInjected
| ServiceBusMessagingOptionsProduction

Azure Service Bus options — exclusive union of injected and production arms.

T
SubscriptionEntry =
SubscriptionDefinition
| RegistryFactory<SubscriptionDefinition>

One entry of MessagingCommonOptions.subscriptions: a subscription definition, or a RegistryFactory producing one when the handler needs a resolved capability.

Usage

import * as Message_broker_plugin_for_cross_service_integration_events___Provides_an__IMessageBroker__implementation_with_support_for_in_memory__Redis_Streams__RabbitMQ__NATS__JetStream___Kafka__GCP_Pub_Sub__Azure_Service_Bus__and_custom_injected_backends__plus_an_optional_bridge_from_the_in_process_event_bus_to_external_messaging__ from "messaging-plugin/src/index.ts";