import { AmqpOptions , DEFAULT_AMQP_OPTIONS } from "./options" ; import { ResourcePool } from "../containers" ; import { MessageBroker } from "../message_broker" ; import { TaskMessage } from "../messages" ; import { isNullOrUndefined , promisifyEvent } from "../utility" ; import * as AmqpLib from "amqplib" ; /**
 * `AmqpBroker` implements `MessageBroker` using the `RabbitMQ` message broker.
 * Messages are, by default, durable, and will survive a broker restart.
 */ export class AmqpBroker implements MessageBroker { private channels : ResourcePool < AmqpLib . Channel > ; private readonly connection : Promise < AmqpLib . Connection > ; private readonly options : AmqpOptions ; param options returns public constructor ( options ? ) { this . options = ( ( ) => { if ( isNullOrUndefined ( options ) ) { return DEFAULT_AMQP_OPTIONS ; } return options ; } ) ( ) ; this . connection = Promise . resolve ( AmqpLib . connect ( this . options ) ) ; this . channels = new ResourcePool ( async ( ) => { const connection = await this . connection ; return connection . createChannel ( ) ; } , async ( channel ) => { await channel . close ( ) ; return "closed" ; } , 2 ) ; } returns see public async disconnect ( ) { await this . end ( ) ; } returns public async end ( ) { await this . channels . destroyAll ( ) ; const connection = await this . connection ; await connection . close ( ) ; } param message returns public async publish ( message ) { const exchange = message . properties . delivery_info . exchange ; const routingKey = message . properties . delivery_info . routing_key ; const body = AmqpBroker . getBody ( message ) ; const options = AmqpBroker . getPublishOptions ( message ) ; return this . channels . use ( async ( channel ) => { await AmqpBroker . assert ( { channel , exchange , routingKey } ) ; return AmqpBroker . doPublish ( { body , channel , exchange , options , routingKey } ) ; } ) ; } /**
     * Converts a task message's body into a `Buffer`.
     */ private static getBody ( message ) { return Buffer . from ( message . body , message . properties . body_encoding ) ; } param message returns private static getPublishOptions ( message ) { return { contentEncoding : message [ "content-encoding" ] , contentType : message [ "content-type" ] , correlationId : message . properties . correlation_id , deliveryMode : message . properties . delivery_mode , headers : message . headers , priority : message . properties . priority , replyTo : message . properties . reply_to , } ; } param channel param exchange param routingKey returns private static async assert ( { channel , exchange , routingKey } ) { const assertion = channel . assertQueue ( routingKey ) ; if ( exchange === "" ) { await assertion ; } else { await Promise . all ( [ assertion , channel . assertExchange ( exchange , "direct" ) , ] ) ; } } param body param channel param exchange param options param routingKey returns private static async doPublish ( { body , channel , exchange , options , routingKey } ) { const publish = ( ) => channel . publish ( exchange , routingKey , body , options ) ; while ( ! publish ( ) ) { await promisifyEvent < void > ( channel , "drain" ) ; } return "flushed to write buffer" ; } }