import { Bundle } from "@bluelibs/core" ; import { EJSON } from "@bluelibs/ejson" ; import { Channel , connect , Connection , Options } from "amqplib" ; import { RabbitMQBundleConfigType } from "./defs" ; export class RabbitMQBundle extends Bundle < RabbitMQBundleConfigType , RabbitMQBundleConfigType > { public connection : Connection ; public channel : Channel ; protected defaultConfig = { url : "amqp://localhost:5672/" , consume : true , } ; protected assertedQueues : string [ ] = [ ] ; async init ( ) { this . connection = await connect ( this . config . url ) ; this . channel = await this . connection . createChannel ( ) ; } public async assertQueue ( queue , options ? ) { return this . channel . assertQueue ( queue , options ) ; } param queue param message param options public publish ( queue , message , options ? ) { return this . channel . sendToQueue ( queue , Buffer . from ( EJSON . stringify ( message ) ) , options ) ; } param queue param handler param options public consume ( queue , handler , options ? ) { if ( ! this . config . consume ) { return ; } this . channel . consume ( queue , async ( msg ) => { await handler ( EJSON . parse ( msg . content . toString ( ) ) ) ; if ( ! options . noAck ) { this . channel . ack ( msg ) ; } } , options ) ; } async shutdown ( ) { await this . channel . close ( ) ; await this . connection . close ( ) ; } }