Class Broker

A wrapper around rascal broker to be used by Mosaic services.

Index

Methods

addMessageHandler bounce connect isConnected nuke off on publish publishCommand publishEvent purge shutdown subscribe unsubscribeAll create

Methods

addMessageHandler

addMessageHandler<TContent>(
handler: MessageHandler<TContent>,
middleware?: OnMessageMiddleware<unknown>[],
): Promise<SubscriberSessionAsPromised>
Attaches a MessageHandler to a subscriptions "On Message" event.
Type Parameters
- TContent
Parameters
- handler: MessageHandler<TContent>
  The message handler class to handle a specific message type
- middleware: OnMessageMiddleware<unknown>[] = []
  Apply middleware on receiving a message
Returns Promise<SubscriberSessionAsPromised>

bounce

bounce(): Promise<void>
Disconnects and re-initializes the broker.

Returns Promise<void>

connect

connect(key: string): Promise<Connection>
Allows to programmatically establish a connection to a vhost.
Parameters
- key: string
  vhost name
Returns Promise<Connection>

isConnected

isConnected(): boolean
Returns a boolean indicating if the service is connected to the Broker.

Returns boolean

nuke

nuke(): Promise<void>
Total teardown of vhost, subscriptions and queues. Can be used in unit tests in an "After All" function

Returns Promise<void>

off

off(event: "broker_error", listener: (data: BrokerErrorEvent) => void): void
Unsubscribes from the 'broker_error' event.
Parameters
- event: "broker_error"
  The event name ('broker_error').
- listener: (data: BrokerErrorEvent) => void
  The callback function to remove.
Returns void
off(
event: "broker_reconnected",
listener: (data: BrokerReconnectedEvent) => void,
): void
Unsubscribes from the 'broker_reconnected' event.
Parameters
- event: "broker_reconnected"
  The event name ('broker_reconnected').
- listener: (data: BrokerReconnectedEvent) => void
  The callback function to remove.
Returns void

on

on(event: "broker_error", listener: (data: BrokerErrorEvent) => void): void
Subscribes to the 'broker_error' event, which is emitted when the broker encounters an error.
Parameters
- event: "broker_error"
  The event name ('broker_error').
- listener: (data: BrokerErrorEvent) => void
  A callback function that receives a BrokerErrorEvent object containing:
  
  error: The error that occurred.
  
  connection: Details about the connection where the error occurred.
Returns void
on(
event: "broker_reconnected",
listener: (data: BrokerReconnectedEvent) => void,
): void
Subscribes to the 'broker_reconnected' event, which is emitted when the broker successfully reconnects.
Parameters
- event: "broker_reconnected"
  The event name ('broker_reconnected').
- listener: (data: BrokerReconnectedEvent) => void
  A callback function that receives a BrokerReconnectedEvent object containing:
  
  connection: Details about the re-established connection.
  
  disconnectedDurationInMilliseconds: The duration (in milliseconds) for which the connection was lost.
Returns void

publish

publish<T>(
    aggregateId: string,
    type: { aggregateType: string; messageType: string },
    payload: T,
    envelopeOverrides?: MessageEnvelopeOverrides,
    options?: PublicationConfig,
): Promise<PublicationSession>
Type Parameters
- T
Parameters
- aggregateId: string
  The (database) ID of the aggregate type or use the const UNKNOWN_AGGREGATE_ID or MULTIPLE_AGGREGATE_IDS values.
- type: { aggregateType: string; messageType: string }
  The aggregate type and message type e.g. channel and ChannelPublishedEvent
- payload: T
  message payload
- OptionalenvelopeOverrides: MessageEnvelopeOverrides
  explicitly defined message envelope values, e.g. auth token or message context
- Optionaloptions: PublicationConfig
  explicitly defined message options, e.g. if additional header values should be passed
Returns Promise<PublicationSession>
Deprecated
please use publishCommand or publishEvent instead.

Publish a message with a defined routing key.

publishCommand

publishCommand<T>(
    aggregateId: string,
    type: { aggregateType: string; messageType: string },
    payload: T,
    envelopeOverrides?: MessageEnvelopeOverrides,
    commandConfig?: PublicationConfig,
): Promise<PublicationSession>
Publish a command message with a defined routing key.
Type Parameters
- T
Parameters
- aggregateId: string
  The (database) ID of the aggregate type or use the const UNKNOWN_AGGREGATE_ID or MULTIPLE_AGGREGATE_IDS values.
- type: { aggregateType: string; messageType: string }
  The aggregate type and message type e.g. image and EnsureImageExists
- payload: T
  message payload
- OptionalenvelopeOverrides: MessageEnvelopeOverrides
  explicitly defined message envelope values, e.g. auth token or message context
- OptionalcommandConfig: PublicationConfig
  explicitly defined command configuration, e.g. if additional header values should be passed
Returns Promise<PublicationSession>

publishEvent

publishEvent<T>(
    aggregateId: string,
    type: { aggregateType: string; messageType: string },
    payload: T,
    envelopeOverrides?: MessageEnvelopeOverrides,
    eventConfig?: PublicationConfig,
): Promise<PublicationSession>
Publish an event message with a defined routing key.
Type Parameters
- T
Parameters
- aggregateId: string
  The (database) ID of the aggregate type or use the const UNKNOWN_AGGREGATE_ID or MULTIPLE_AGGREGATE_IDS values.
- type: { aggregateType: string; messageType: string }
  The aggregate type and message type e.g. channel and ChannelPublishedEvent
- payload: T
  message payload
- OptionalenvelopeOverrides: MessageEnvelopeOverrides
  explicitly defined message envelope values, e.g. auth token or message context
- OptionaleventConfig: PublicationConfig
  explicitly defined event configuration, e.g. if additional header values should be passed
Returns Promise<PublicationSession>

purge

purge(): Promise<void>
Purges all messages from all queues of all vhosts. Can be used together with unsubscribeAll in unit tests in an "After Each" function

Returns Promise<void>

shutdown

shutdown(): Promise<void>
Shutdowns the broker, which cancels all subscriptions and waits for a sort amount of time for inflight messages to be acknowledged, before closing channels and disconnecting.

Returns Promise<void>

subscribe(
key: string,
overrides?: SubscriptionConfig,
): Promise<SubscriberSessionAsPromised>
Wrapper for a default rascal subscribe method, allowing explicit subscription to a queue using a dedicated routing key. Consider using addMessageHandler to initialize a subscription and directly attach a message handled to be executed on message received event.
Parameters
- key: string
- Optionaloverrides: SubscriptionConfig
Returns Promise<SubscriberSessionAsPromised>

unsubscribeAll

unsubscribeAll(): Promise<void>
Cancels all established subscription sessions. Can be used together with purge in unit tests in an "After Each" function

Returns Promise<void>

`Static`create

create(
    config: BrokerConfig,
    logger: Logger,
    components?: Record<string, unknown>,
    signing?: Required<
        Pick<
            ValueObject<
                {
                    rmqChannelMax: () => number;
                    rmqDevMiddleware: () => boolean;
                    rmqEventSigningKeyVersion: () => number;
                    rmqEventSigningKeyVersionsToRevoke: () => number[];
                    rmqEventSigningPrivateKey: () => string;
                    rmqEventSigningPublicKey: () => string;
                    rmqHost: () => string;
                    rmqMgmtHost: () => string | undefined;
                    rmqMgmtPort: () => number | undefined;
                    rmqMgmtProtocol: () => string | undefined;
                    rmqPassword: () => string;
                    rmqPort: () => number;
                    rmqProtocol: () => string;
                    rmqUser: () => string;
                    rmqVHost: () => string;
                    rmqVHostAssert: () => boolean;
                },
            >,
            "rmqEventSigningPrivateKey"
            | "rmqEventSigningKeyVersion",
        >,
    >,
): Promise<Broker>
Initializes a rascal broker instance using a rascal config and components
Parameters
- config: BrokerConfig
- logger: Logger
- components: Record<string, unknown> = {}
- Optionalsigning: Required<
      Pick<
          ValueObject<
              {
                  rmqChannelMax: () => number;
                  rmqDevMiddleware: () => boolean;
                  rmqEventSigningKeyVersion: () => number;
                  rmqEventSigningKeyVersionsToRevoke: () => number[];
                  rmqEventSigningPrivateKey: () => string;
                  rmqEventSigningPublicKey: () => string;
                  rmqHost: () => string;
                  rmqMgmtHost: () => string | undefined;
                  rmqMgmtPort: () => number | undefined;
                  rmqMgmtProtocol: () => string | undefined;
                  rmqPassword: () => string;
                  rmqPort: () => number;
                  rmqProtocol: () => string;
                  rmqUser: () => string;
                  rmqVHost: () => string;
                  rmqVHostAssert: () => boolean;
              },
          >,
          "rmqEventSigningPrivateKey"
          | "rmqEventSigningKeyVersion",
      >,
  >
Returns Promise<Broker>

Class Broker

Index

Methods

Methods

addMessageHandler

Type Parameters

Parameters

Returns Promise<SubscriberSessionAsPromised>

bounce

Returns Promise<void>

connect

Parameters

Returns Promise<Connection>

isConnected

Returns boolean

nuke

Returns Promise<void>

off

Parameters

Returns void

Parameters

Returns void

on

Parameters

Returns void

Parameters

Returns void

publish

Type Parameters

Parameters

Returns Promise<PublicationSession>

Deprecated

publishCommand

Type Parameters

Parameters

Returns Promise<PublicationSession>

publishEvent

Type Parameters

Parameters

Returns Promise<PublicationSession>

purge

Returns Promise<void>

shutdown

Returns Promise<void>

subscribe

Parameters

Returns Promise<SubscriberSessionAsPromised>

unsubscribeAll

Returns Promise<void>

Staticcreate

Parameters

Returns Promise<Broker>

Settings

On This Page

`Static`create