@appolo/bus

Appolo Bus Module

bus module for appolo built with rabbot

Installation

npm i @appolo/bus

Options

key	Description	Type	Default
id	injection id	string	busProvider
connection	AMQP connection string	string	null
auto	true to auto initialize busProvider and start listen to events	boolean	true
listener	true to register queue event handlers	boolean	true
exchangeName	name of the exchange	string
queueName	name of the queue	string
appendEnv	append env name to queueName and exchangeName	boolean	true
exchange	exchange options	object	{}
queue	queue options	object	{}
requestQueue	request queue options	object	{}
replayQueue	request queue options or false to disable	object	{}

Exchange Options

key	Description	Type	Default
type	request queue options or false to disable	string	topic
autoDelete	delete when consumer count goes to 0	boolean	false
durable	survive broker restarts	boolean	true
persistent	persistent delivery, messages saved to disk	boolean	true
alternate	define an alternate exchange	string
publishTimeout	timeout in milliseconds for publish calls to this exchange	2^32
replyTimeout	timeout in milliseconds to wait for a reply	2^32
limit	the number of unpublished messages to cache while waiting on connection	2^16
noConfirm	prevents rabbot from creating the exchange in confirm mode	boolean	false

Queue Options

key	Description	Type	Default
autoDelete	delete when consumer count goes to 0	boolean	false
durable	survive broker restarts	boolean	true
subscribe	auto-start the subscription	boolean	false
limit	max number of unacked messages allowed for consumer	2^16	1
noAck	the server will remove messages from the queue as soon as they are delivered	boolean	false
noBatch	causes ack, nack & reject to take place immediately	boolean	false
noCacheKeys	disable cache of matched routing keys to prevent unbounded memory growth	boolean	false
queueLimit	max number of ready messages a queue can hold	2^32
messageTt	time in ms before a message expires on the queue	2^32
expires	time in ms before a queue with 0 consumers expires	2^32
in config/modules/all.ts

import {PubSubModule} from '@appolo/pubsub';

export = async function (app: App) {
   await app.module(new BusModule({redis:"amqp://connection-string"}));
}

Usage

Publisher

import {define, singleton} from 'appolo'
import {publisher} from "@appolo/bus";

@define()
@singleton()
export class SomePublisher {

    @publisher("test")
    async publish(data: any): Promise<any> {
        return data
    }
}

Or with BusProvider

@define()
@singleton()
export class SomePublisher {

    inject() busProvider:BusProvider

    publish(data:any): Promise<any> {
        return this.busProvider.publish("test",data)
    }
}

Handler

if you don not call msg ack or nack it will be called on handler return msg.ack() or msg.nack() on error

import {define, singleton} from 'appolo'
import {handler} from "@appolo/bus";

@define()
@singleton()
export class SomeHandler {

    @handler("test")
    handle(msg: IMessage<data>) {
       //do something
    }

    @handler("someName")
    handle(msg: IMessage<data>) {

        try{
           //do some thing

           msg.ack();
        }
        catch(){
            msg.nack();
        }
    }
}

Request

import {define, singleton} from 'appolo'
import {request} from "@appolo/bus";

@define()
@singleton()
export class SomePublisher {

    @request("test")
    async getData(data: any): Promise<any> {
        return data
    }

    public async handleData(){
        let data = await this.getData({userId:1})
    }


}

Or with BusProvider

@define()
@singleton()
export class SomePublisher {

    inject() busProvider:busProvider

    publish(data:any): Promise<any> {
        let data = await  this.busProvider.request("test",data)

        return data;
    }
}

Reply

import {define, singleton} from 'appolo'
import {handler} from "@appolo/bus";

@define()
@singleton()
export class SomeHandler {

    inject() busProvider:busProvider


    @reply("test")
    handle(msg: IMessage<data>) {
        return {userId:1}
    }

    // or reply methods
    @reply("someName")
    handle(msg: IMessage<data>) {

        try{
            //get some data
         msg.replySuccess(msg,{userId:1})
        }
        catch(){
            msg.replyError(msg,e)
        }
    }
}

IMessage

each handler and reply handler called with message object

{
  // metadata specific to routing & delivery
  fields: {
    consumerTag: "", // identifies the consumer to rabbit
    deliveryTag: #, // identifies the message delivered for rabbit
    redelivered: true|false, // indicates if the message was previously nacked or returned to the queue
    exchange: "" // name of exchange the message was published to,
    routingKey: "" // the routing key (if any) used when published
  },
  properties:{
    contentType: "application/json", // see serialization for how defaults are determined
    contentEncoding: "utf8", // rabbot's default
    headers: {}, // any user provided headers
    correlationId: "", // the correlation id if provided
    replyTo: "", // the reply queue would go here
    messageId: "", // message id if provided
    type: "", // the type of the message published
    appId: "" // not used by rabbot
  },
  content: { "type": "Buffer", "data": [ ... ] }, // raw buffer of message body
  body: , // this could be an object, string, etc - whatever was published
  type: "" // this also contains the type of the message published
}

message.ack()

Enqueues the message for acknowledgement.

message.nack()

Enqueues the message for rejection. This will re-enqueue the message.

message.reject()

Rejects the message without re-queueing it. Please use with caution and consider having a dead-letter-exchange assigned to the queue before using this feature.

message.reply( data:any )

Acknowledges the messages and sends the message back to the requestor.

message.replySuccess( data:T )

reply the message with json object {success: true,data}

message.replyError( e: RequestError<T> )

reply the message with json object {success: false,message: e.message, data:e.data}

BusProvider

initialize()

initialize busProvider and start listen to events if not in in auto mode

publish(type: string, data: any, expire?: number): Promise<void>

publish event

• type - event name
• data - any data
• expire - timeout until the message is expired in the queue

request<T>(type: string, data: any, expire?: number): Promise<T>

request data by event return promise with event response

• type - event name
• data - any data
• expire - timeout until the request is rejected

close<T>(): Promise<void>

close the connection and clean all handlers

getQueueMessagesCount(): Promise<number>

return number of pending events in the queue