import { EventEmitter } from "events" ; import { MessageQueue , MessageQueueMessage , DEFAULT_RES_TIMEOUT } from "./Types" ; import micromatch from "micromatch" ; import { v4 as uuid } from "uuid" ; import Metrics from "../Metrics" ; export class LocalMQ extends EventEmitter implements MessageQueue { private subs : Set < string > ; constructor ( ) { super ( ) ; this . subs = new Set ( ) ; } public subscribe ( eventGlob ) { this . subs . add ( eventGlob ) ; } public unsubscribe ( eventGlob ) { this . subs . delete ( eventGlob ) ; } public async push < T > ( message ) { Metrics . messageQueuePushes . inc ( { event : message . eventName } ) ; if ( ! micromatch . match ( [ ... this . subs ] , message . eventName ) ) { return ; } if ( ! message . messageId ) { message . messageId = uuid ( ) ; } this . emit ( message . eventName , message ) ; } public async pushWait < T , X > ( message , timeout = DEFAULT_RES_TIMEOUT ) { let resolve ; let timer ; const p = new Promise < X > ( ( res , rej ) => { resolve = res ; timer = setTimeout ( ( ) => { rej ( new Error ( "Timeout waiting for message queue response" ) ) ; } , timeout ) ; } ) ; const awaitResponse = ( response ) => { if ( response . messageId === message . messageId ) { clearTimeout ( timer ) ; this . removeListener ( `response.${ message . eventName }` , awaitResponse ) ; resolve ( response . data ) ; } } ; this . addListener ( `response.${ message . eventName }` , awaitResponse ) ; this . push ( message ) ; return p ; } }