import { MessageError } from "../errors/MessageError" ; import { logger } from "../lib/logger" ; export abstract class MessageManager { private queue : string ; static connection ; constructor ( queue ) { this . queue = queue ; this . subscribe ( ) . then ( ( ) => { logger . debug ( 'Message queue subscribed' ) ; } , ( ) => { logger . error ( 'Message queue subscription failed' ) ; } ) ; } public abstract async handleMessage ( message ) ; public async publish ( content ) { try { const channel = await MessageManager . connection . createChannel ( ) ; channel . assertQueue ( this . queue , { durable : true } ) . then ( ( ) => { channel . sendToQueue ( this . queue , Buffer . from ( JSON . stringify ( content ) ) , { persistent : true , contentType : 'application/json' } ) ; return true ; } , ( error ) => { throw error ; } ) ; } catch ( error ) { throw new MessageError ( error . message ) ; } } public async subscribe ( ) { try { const channel = await MessageManager . connection . createChannel ( ) ; channel . assertQueue ( this . queue , { durable : true } ) . then ( ( ) => { channel . prefetch ( 1 ) ; channel . consume ( this . queue , messageData => { if ( messageData === null ) { return ; } const message = JSON . parse ( messageData . content . toString ( ) ) ; this . handleMessage ( message ) . then ( ( ) => { return channel . ack ( messageData ) ; } , ( ) => { return channel . nack ( messageData ) ; } ) ; } ) ; } , ( error ) => { throw error ; } ) ; } catch ( error ) { throw new MessageError ( error . message ) ; } } }