Class BrokerProxy

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

Hierarchy

BrokerProxy

Constructors

`Private` constructor

new BrokerProxy(): BrokerProxy
Returns BrokerProxy

Properties

`Private` attachBrokerHandlers

attachBrokerHandlers: any

`Private` broker

broker: any

`Private` eventEmitter

eventEmitter: any

`Private` `Optional` eventSigning

eventSigning?: any

`Private` getPublication

getPublication: any

`Private` getSubscription

getSubscription: any

`Private` isConnectedToBroker

isConnectedToBroker: any

`Private` `Optional` lastDisconnectedAt

lastDisconnectedAt?: any

`Private` logger

logger: any

`Private` onBrokerBlocked

onBrokerBlocked: any

`Private` onBrokerBusy

onBrokerBusy: any

`Private` onBrokerError

onBrokerError: any

`Private` onBrokerReady

onBrokerReady: any

`Private` onBrokerUnblocked

onBrokerUnblocked: any

`Private` onBrokerVhostInitialized

onBrokerVhostInitialized: any

`Private` publications

publications: any

`Private` subscriptions

subscriptions: any

`Private` `Static` instance

instance: any

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
- Optional 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): 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, listener): void
Unsubscribes from the 'error' event.
Parameters
- event: "error"
  
  The event name ('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 'reconnected' event.
Parameters
- event: "reconnected"
  
  The event name ('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 'error' event, which is emitted when the broker encounters an error.
Parameters
- event: "error"
  
  The event name ('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 'reconnected' event, which is emitted when the broker successfully reconnects.
Parameters
- event: "reconnected"
  
  The event name ('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

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<BrokerProxy>
Initializes a rascal broker instance using a rascal config and components
Parameters
- config: BrokerConfig
- logger: Logger
- Optional 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<BrokerProxy>

Class BrokerProxy

Hierarchy

Index

Constructors

Properties

Methods

Constructors

Private constructor

Returns BrokerProxy

Properties

Private attachBrokerHandlers

Private broker

Private eventEmitter

Private Optional eventSigning

Private getPublication

Private getSubscription

Private isConnectedToBroker

Private Optional lastDisconnectedAt

Private logger

Private onBrokerBlocked

Private onBrokerBusy

Private onBrokerError

Private onBrokerReady

Private onBrokerUnblocked

Private onBrokerVhostInitialized

Private publications

Private subscriptions

Private Static instance

Methods

addMessageHandler

Type Parameters

TContent

Parameters

handler: MessageHandler<TContent>

Optional middleware: OnMessageMiddleware<unknown>[]

Returns Promise<SubscriberSessionAsPromised>

bounce

Returns Promise<void>

connect

Parameters

key: string

Returns Promise<Connection>

isConnected

Returns boolean

nuke

Returns Promise<void>

off

Parameters

event: "error"

listener: ((data) => void)

Parameters

data: BrokerErrorEvent

Returns void

Returns void

Parameters

event: "reconnected"

listener: ((data) => void)

Parameters

data: BrokerReconnectedEvent

Returns void

Returns void

on

Parameters

event: "error"

listener: ((data) => void)

Parameters

data: BrokerErrorEvent

Returns void

Returns void

Parameters

event: "reconnected"

listener: ((data) => void)

Parameters

data: BrokerReconnectedEvent

Returns void

Returns void

publish

Type Parameters

T

Parameters

`Private` constructor

`Private` attachBrokerHandlers

`Private` broker

`Private` eventEmitter

`Private` `Optional` eventSigning

`Private` getPublication

`Private` getSubscription

`Private` isConnectedToBroker

`Private` `Optional` lastDisconnectedAt

`Private` logger

`Private` onBrokerBlocked

`Private` onBrokerBusy

`Private` onBrokerError

`Private` onBrokerReady

`Private` onBrokerUnblocked

`Private` onBrokerVhostInitialized

`Private` publications

`Private` subscriptions

`Private` `Static` instance

`Optional` middleware: OnMessageMiddleware<unknown>[]

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

`Optional` components: Record<string, unknown>