Class Broker

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

Hierarchy

Broker

Constructors

`Private` constructor

new Broker(broker, logger, eventSigning?): Broker
Parameters
- broker: BrokerAsPromised
- logger: Logger
- Optional eventSigning: Required<Pick<ValueObject<{
      rmqChannelMax: (() => number);
      rmqDevMiddleware: (() => boolean);
      rmqEventSigningKeyVersion: (() => number);
      rmqEventSigningKeyVersionsToRevoke: (() => number[]);
      rmqEventSigningPrivateKey: (() => string);
      rmqEventSigningPublicKey: (() => string);
      rmqHost: (() => string);
      rmqMgmtHost: (() => undefined | string);
      rmqMgmtPort: (() => undefined | number);
      rmqMgmtProtocol: (() => undefined | string);
      rmqPassword: (() => string);
      rmqPort: (() => number);
      rmqProtocol: (() => string);
      rmqUser: (() => string);
      rmqVHost: (() => string);
      rmqVHostAssert: (() => boolean);
  }>, "rmqEventSigningPrivateKey" | "rmqEventSigningKeyVersion">>
Returns Broker

Properties

`Private` broker

broker: BrokerAsPromised

`Private` eventEmitter

eventEmitter: EventEmitter = ...

`Private` `Optional` eventSigning

eventSigning?: Required<Pick<ValueObject<{
    rmqChannelMax: (() => number);
    rmqDevMiddleware: (() => boolean);
    rmqEventSigningKeyVersion: (() => number);
    rmqEventSigningKeyVersionsToRevoke: (() => number[]);
    rmqEventSigningPrivateKey: (() => string);
    rmqEventSigningPublicKey: (() => string);
    rmqHost: (() => string);
    rmqMgmtHost: (() => undefined | string);
    rmqMgmtPort: (() => undefined | number);
    rmqMgmtProtocol: (() => undefined | string);
    rmqPassword: (() => string);
    rmqPort: (() => number);
    rmqProtocol: (() => string);
    rmqUser: (() => string);
    rmqVHost: (() => string);
    rmqVHostAssert: (() => boolean);
}>, "rmqEventSigningPrivateKey" | "rmqEventSigningKeyVersion">>

`Private` isConnectedToBroker

isConnectedToBroker: boolean = true

`Private` `Optional` lastDisconnectedAt

lastDisconnectedAt?: Date

`Private` logger

logger: Logger

`Private` publications

publications: Map<string, default> = ...

`Private` subscriptions

subscriptions: Map<string, default<any>> = ...

`Private` `Static` instance

instance: Broker

Methods

addMessageHandler

addMessageHandler<TContent>(handler, middleware?): 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>

`Private` attachBrokerHandlers

attachBrokerHandlers(): void
Returns void

bounce

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

Returns Promise<void>

connect

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

`Private` getPublication

getPublication(messageType): default
Parameters
- messageType: AggregateMessageType
Returns default

`Private` getSubscription

getSubscription<TContent>(handler, middleware?): Promise<default<TContent>>
Type Parameters
- TContent
Parameters
- handler: MessageHandler<TContent>
- middleware: OnMessageMiddleware<unknown>[] = []
Returns Promise<default<TContent>>

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

on

on(event, listener): 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) => void)
  A callback function that receives a BrokerErrorEvent object containing:
  - error: The error that occurred.
  - connection: Details about the connection where the error occurred.
  - - (data): void
    - Parameters
      
      data: BrokerErrorEvent
      
      Returns void
Returns void
on(event, listener): 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) => 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.
  - - (data): void
    - Parameters
      
      data: BrokerReconnectedEvent
      
      Returns void
Returns void

`Private` onBrokerBlocked

onBrokerBlocked(reason, connection): void
Parameters
- reason: string
- connection: ConnectionDetails
Returns void

`Private` onBrokerBusy

onBrokerBusy(stats): void
Parameters
- stats: BrokerStats
Returns void

`Private` onBrokerError

onBrokerError(error, connection): void
Parameters
- error: Error
- connection: ConnectionDetails
Returns void

`Private` onBrokerReady

onBrokerReady(stats): void
Parameters
- stats: BrokerStats
Returns void

`Private` onBrokerUnblocked

onBrokerUnblocked(reason, connection): void
Parameters
- reason: string
- connection: ConnectionDetails
Returns void

`Private` onBrokerVhostInitialized

onBrokerVhostInitialized(connection): void
Parameters
- connection: ConnectionDetails
Returns void

publish

publish<T>(aggregateId, type, payload, envelopeOverrides?, options?): 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
  - aggregateType: string
  - messageType: string
- payload: T
  
  message payload
- Optional envelopeOverrides: MessageEnvelopeOverrides
  
  explicitly defined message envelope values, e.g. auth token or message context
- Optional options: 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, type, payload, envelopeOverrides?, commandConfig?): 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
  - aggregateType: string
  - messageType: string
- payload: T
  
  message payload
- Optional envelopeOverrides: MessageEnvelopeOverrides
  
  explicitly defined message envelope values, e.g. auth token or message context
- Optional commandConfig: PublicationConfig
  
  explicitly defined command configuration, e.g. if additional header values should be passed
Returns Promise<PublicationSession>

publishEvent

publishEvent<T>(aggregateId, type, payload, envelopeOverrides?, eventConfig?): 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
  - aggregateType: string
  - messageType: string
- payload: T
  
  message payload
- Optional envelopeOverrides: MessageEnvelopeOverrides
  
  explicitly defined message envelope values, e.g. auth token or message context
- Optional eventConfig: 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, overrides?): 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
- Optional overrides: 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, logger, components?, signing?): Promise<Broker>
Initializes a rascal broker instance using a rascal config and components
Parameters
- config: BrokerConfig
- logger: Logger
- components: Record<string, unknown> = {}
- Optional signing: Required<Pick<ValueObject<{
      rmqChannelMax: (() => number);
      rmqDevMiddleware: (() => boolean);
      rmqEventSigningKeyVersion: (() => number);
      rmqEventSigningKeyVersionsToRevoke: (() => number[]);
      rmqEventSigningPrivateKey: (() => string);
      rmqEventSigningPublicKey: (() => string);
      rmqHost: (() => string);
      rmqMgmtHost: (() => undefined | string);
      rmqMgmtPort: (() => undefined | number);
      rmqMgmtProtocol: (() => undefined | string);
      rmqPassword: (() => string);
      rmqPort: (() => number);
      rmqProtocol: (() => string);
      rmqUser: (() => string);
      rmqVHost: (() => string);
      rmqVHostAssert: (() => boolean);
  }>, "rmqEventSigningPrivateKey" | "rmqEventSigningKeyVersion">>
Returns Promise<Broker>

Class Broker

Hierarchy

Index

Constructors

Properties

Methods

Constructors

Private constructor

Parameters

broker: BrokerAsPromised

logger: Logger

Returns Broker

Properties

Private broker

Private eventEmitter

Private Optional eventSigning

Private isConnectedToBroker

Private Optional lastDisconnectedAt

Private logger

Private publications

Private subscriptions

Private Static instance

Methods

addMessageHandler

Type Parameters

TContent

Parameters

handler: MessageHandler<TContent>

middleware: OnMessageMiddleware<unknown>[] = []

Returns Promise<SubscriberSessionAsPromised>

Private attachBrokerHandlers

Returns void

bounce

Returns Promise<void>

connect

Parameters

key: string

Returns Promise<Connection>

Private getPublication

Parameters

messageType: AggregateMessageType

Returns default

Private getSubscription

Type Parameters

TContent

Parameters

handler: MessageHandler<TContent>

middleware: OnMessageMiddleware<unknown>[] = []

Returns Promise<default<TContent>>

isConnected

Returns boolean

nuke

Returns Promise<void>

off

Parameters

event: "broker_error"

listener: ((data) => void)

Parameters

data: BrokerErrorEvent

Returns void

Returns void

Parameters

event: "broker_reconnected"

listener: ((data) => void)

Parameters

data: BrokerReconnectedEvent

Returns void

Returns void

on

Parameters

event: "broker_error"

listener: ((data) => void)

Parameters

data: BrokerErrorEvent

Returns void

Returns void

Parameters

event: "broker_reconnected"

listener: ((data) => void)

Parameters

`Private` constructor

`Private` broker

`Private` eventEmitter

`Private` `Optional` eventSigning

`Private` isConnectedToBroker

`Private` `Optional` lastDisconnectedAt

`Private` logger

`Private` publications

`Private` subscriptions

`Private` `Static` instance

`Private` attachBrokerHandlers

`Private` getPublication

`Private` getSubscription

`Private` onBrokerBlocked

`Private` onBrokerBusy

`Private` onBrokerError

`Private` onBrokerReady

`Private` onBrokerUnblocked

`Private` onBrokerVhostInitialized

type: {
aggregateType: string;
messageType: string;
}

`Optional` envelopeOverrides: MessageEnvelopeOverrides

`Optional` options: PublicationConfig

type: {
aggregateType: string;
messageType: string;
}

`Optional` envelopeOverrides: MessageEnvelopeOverrides

`Optional` commandConfig: PublicationConfig

type: {
aggregateType: string;
messageType: string;
}

`Optional` envelopeOverrides: MessageEnvelopeOverrides

`Optional` eventConfig: PublicationConfig