Installing

npm install @db3dev/activemq

ActiveMQ

Connection Configuration Object

host: String for message broker uri. Do not include protocol or port

port: Number for port to connect to on message broker

username: String for login credentials

password: String password for login credentials

ssl: Boolean indicating that messages should be encrypted

reconnectDelay: Number overriding default reconnection retry wait time

debugCallback: Function to call for debug messages. Should not fill console for production

stompExceptionCallback: Function to call when STOMP client fails

websocketExceptionCallback: Function to call when Websocket Connection fails

API

When constructing the client object optionally a http.Agent interface can be passed in. This allows the support and configuration of a proxy server in the connection.

connect() --- Triggers websocket and STOMP protocols to activate and maintain connection to message queue.

disconnect() --- Triggers websocket and STOMP protocols to deactivate (close) and not attempt to maintain a connection to message queue.

publish(msg, queue) --- Serializes and sends message to designated queue.

rpc(msg, queue, responseCallback, replyTo?) --- Wraps publish functionality but subscribes to a replyTo queue (randomly named if undefined) that will pass the response to the responseCallback when received.

subscribe(queue, listener) --- Subscribes to designated queue name and passes all messages received to listener function. Return a promised unsubscribe function.

Implementing into NestJS as Listener Example

classApplicationModule{ ... }awaitNestFactory.createMicroservice(ApplicationModule,{strategy: newclassActiveMQextendsServerimplementsCustomTransportStrategy{constructor(privatereadonlyclient: ActiveMQClient,privatereadonlyqueue: string){super();}listen(callback: ()=>void){this.client.connect();this.client.subscribe(// Defines the queue the strategy will subscribe to{hash: this.queue,queueName: this.queue,prefetchSize: 1},// The ActiveMQ Client is written to parse the incoming message.// NestJS will then use its handler map to execute decorated function that matches patternasync(msg: {pattern: any,data: any,replyTo?: string})=>{constpattern=JSON.stringify(msg.pattern);consthandler=this.getHandlerByPattern(pattern);if(!handler){return;}returnawaithandler(msg.data);}).then(()=>callback());}close(){returnthis.client.disconnect()}}(newActiveMQClient(config),'developmentQueue')});

Using as NestJS Client Example

classActiveMQextendsClientProxy{constructor(privatereadonlyclient: ActiveMQClient,privatereadonlyqueue: string){super();}connect(): Promise<any>{returnthis.client.connect();}close(){returnthis.client.disconnect();}protectedpublish(packet: ReadPacket<any>,callback: (packet: WritePacket<any>)=>void): Function{// NestJS callback is wrapped with a new function that adds RPC response to NestJS observer responsethis.client.rpc(packet,this.queue,(msg: any)=>{callback({err: null,response: msg,isDisposed: true})});// No-Op code, usually handles connection cleanup but library handles this for us.return()=>{};}protecteddispatchEvent<T=any>(packet: ReadPacket<any>): Promise<T>{returnthis.client.publish(packet,this.queue)asPromise<any>;}}constclient=newActiveMQ(newActiveMQClient({ ... }),'developmentQueue');// Send an RPC Requestclient.send('patternA','msg').subscribe((msg)=>console.log(JSON.stringify(msg)));// Publish a one way message expecting no responseclient.emit('patternB','msg');