import * as amqplib from "amqplib" ; import { CeleryBroker } from "." ; import { Message } from "../message" ; class AMQPMessage extends Message { constructor ( payload ) { super ( payload . content , payload . properties . contentType , payload . properties . contentEncoding , payload . properties , payload . properties . headers ) ; } } export default class AMQPBroker implements CeleryBroker { connect : Promise < amqplib . Connection > ; channel : Promise < amqplib . Channel > ; queue : string ; constructor param { string } url param { object } opts param { string } queue constructor ( url , opts , queue = "celery" ) { this . queue = queue ; this . connect = amqplib . connect ( url , opts ) ; this . channel = this . connect . then ( conn => conn . createChannel ( ) ) ; } method returns { Promise } public isReady ( ) { return new Promise ( ( resolve , reject ) => { this . channel . then ( ch => { Promise . all ( [ ch . assertExchange ( "default" , "direct" , { durable : true , autoDelete : true , internal : false , arguments : null } ) , ch . assertQueue ( this . queue , { durable : true , autoDelete : false , exclusive : false , arguments : null } ) ] ) . then ( ( ) => resolve ( ) ) . catch ( reject ) ; } ) ; } ) ; } method returns { Promise } public disconnect ( ) { return this . connect . then ( conn => conn . close ( ) ) ; } method returns { Promise } public publish ( body , exchange , routingKey , headers , properties ) { const messageBody = JSON . stringify ( body ) ; const contentType = "application/json" ; const contentEncoding = "utf-8" ; return this . channel . then ( ch => ch . assertQueue ( routingKey , { durable : true , autoDelete : false , exclusive : false , arguments : null } ) . then ( ( ) => Promise . resolve ( ch ) ) ) . then ( ch => ch . publish ( exchange , routingKey , Buffer . from ( messageBody ) , { contentType , contentEncoding , headers , deliveryMode : 2 , ... properties } ) ) ; } method param { String } queue param { Function } callback returns { Promise } public subscribe ( queue , callback ) { return this . channel . then ( ch => ch . assertQueue ( queue , { durable : true , autoDelete : false , exclusive : false , arguments : null } ) . then ( ( ) => Promise . resolve ( ch ) ) ) . then ( ch => ch . consume ( queue , rawMsg => { ch . ack ( rawMsg ) ; if ( rawMsg . properties . contentType !== "application/json" ) { throw new Error ( `unsupported content type ${ rawMsg . properties . contentType }` ) ; } if ( rawMsg . properties . contentEncoding !== "utf-8" ) { throw new Error ( `unsupported content encoding ${ rawMsg . properties . contentEncoding }` ) ; } callback ( new AMQPMessage ( rawMsg ) ) ; } ) ) ; } }