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` `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` 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>

`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 Optional eventSigning

Private isConnectedToBroker

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>

Private onBrokerBlocked

Parameters

reason: string

connection: ConnectionDetails

Returns void

Private onBrokerBusy

Parameters

stats: BrokerStats

Returns void

Private onBrokerError

Parameters

error: Error

connection: ConnectionDetails

Returns void

Private onBrokerReady

Parameters

stats: BrokerStats

Returns void

Private onBrokerUnblocked

Parameters

reason: string

connection: ConnectionDetails

Returns void

Private onBrokerVhostInitialized

Parameters

connection: ConnectionDetails

Returns void

publish

Type Parameters

`Private` constructor

`Private` broker

`Private` `Optional` eventSigning

`Private` isConnectedToBroker

`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

`Optional` overrides: SubscriptionConfig

`Static` create