Custom transports
If you want to use a message queue that's not officially supported by the transport list, you can provide an adapter layer by implementing the Transport interface from @node-ts/bus-core.
The following interface definition lists the functionality that must be implemented in order for the transport to be compatible.
/**
* A transport adapter interface that enables the service bus to use a messaging technology.
*/
export interface Transport<TransportMessageType = {}> {
/**
* Publishes an event to the underlying transport. This is generally done to a topic or some other
* mechanism that consumers can subscribe themselves to
* @param event A domain event to be published
* @param messageOptions Options that control the behaviour around how the message is sent and
* additional information that travels with it.
*/
publish<TEvent extends Event> (event: TEvent, messageOptions?: MessageAttributes): Promise<void>
​
/**
* Sends a command to the underlying transport. This is generally done to a topic or some other
* mechanism that consumers can subscribe themselves to
* @param command A domain command to be sent
* @param messageOptions Options that control the behaviour around how the message is sent and
* additional information that travels with it.
*/
send<TCommand extends Command> (command: TCommand, messageOptions?: MessageAttributes): Promise<void>
​
/**
* Forwards @param transportMessage to the dead letter queue. The message must have been read in from the
* queue and have a receipt handle.
*/
fail (transportMessage: TransportMessage<unknown>): Promise<void>
​
/**
* Forwards @param transportMessage to the dead letter queue. The message must have been read in from the
* queue and have a receipt handle.
*/
fail (transportMessage: TransportMessage<unknown>): Promise<void>
​
/**
* Fetch the next message from the underlying queue. If there are no messages, then `undefined`
* should be returned.
*
* @returns The message construct from the underlying transport, that inclues both the raw message envelope
* plus the contents or body that contains the `@node-ts/bus-messages` message.
*/
readNextMessage (): Promise<TransportMessage<TransportMessageType> | undefined>
​
/**
* Removes a message from the underlying transport. This will be called once a message has been
* successfully handled by any of the message handling functions.
* @param message The message to be removed from the transport
*/
deleteMessage (message: TransportMessage<TransportMessageType>): Promise<void>
​
/**
* Returns a message to the queue for retry. This will be called if an error was thrown when
* trying to process a message.
* @param message The message to be returned to the queue for reprocessing
*/
returnMessage (message: TransportMessage<TransportMessageType>): Promise<void>
​
/**
* An optional function that is called before startup that will provide core dependencies
* to the transport. This can be used to fetch loggers, registries etc that are used
* in initialization steps
* @param coreDependencies
*/
prepare (coreDependencies: CoreDependencies): void
​
/**
* An optional function that will be called on startup. This gives a chance for the transport
* to establish any connections to the underlying infrastructure.
*/
connect? (): Promise<void>
​
/**
* An optional function that will be called on shutdown. This gives a chance for the transport
* to close any connections to the underlying infrastructure.
*/
disconnect? (): Promise<void>
​
/**
* An optional function that will be called when the service bus is starting. This is an
* opportunity for the transport to see what messages need to be handled so that subscriptions
* to the topics can be created.
* @param handlerRegistry The list of messages being handled by the bus that the transport needs to subscribe to.
*/
initialize? (handlerRegistry: HandlerRegistry): Promise<void>
​
/**
* An optional function that will be called when the service bus is shutting down. This is an
* opportunity for the transport to close out any open requests to fetch messages etc.
*/
dispose? (): Promise<void>
}
​
Once your transport has implemented this interface, it can be provided to the bus on configuration
import { Bus } from '@node-ts/bus-core'
import { MyTransport } from './my-transport'
​
const myTransport = new MyTransport()
await Bus
.configure()
.withTransport(myTransport)
.intialize()
​
Copy link