@coolgk/amqp

@coolgk/amqp

a javascript / typescript module

npm install @coolgk/amqp

a simple RabbitMQ (amqp wrapper) class for publishing and consuming messages

constructor(options)

• Parameters
  • options
    • {string} options.url - connection string e.g. amqp://localhost
    • {string} [options.sslPem] - pem file path
    • {string} [options.sslCa] - sslCa file path
    • {string} [options.sslPass] - password
• Return Value
  • void

closeConnection()

close the connection

• Return Value
  • void

publish(message, callback, { route = '#', exchangeName = 'defaultExchange' } = {})

• Parameters
  • {*} message - message any type that can be JSON.stringify'ed
  • {function} [callback] - callback(message) for processing response from consumers
  • {object} [options]
    • {string} [options.route='#'] - route name
    • {string} [options.exchangeName='defaultExchange'] - exchange name
• Return Value
  • Promise

consume(callback, { route = '#', queueName = 'defaultQueue', exchangeName = 'defaultExchange', exchangeType = 'topic', priority = 0, prefetch = 0 } = {})

• Parameters
  • {function} callback - consumer(message) function should returns a promise
  • {object} [options]
  • {string} [options.route='#'] - exchange route
  • {string} [options.queueName='defaultQueue'] - queue name for processing request
  • {string} [options.exchangeName='defaultExchange'] - exchange name
  • {string} [options.exchangeType='topic'] - exchange type
  • {number} [options.priority=0] - priority, larger numbers indicate higher priority
  • {number} [options.prefetch=0] - 1 or 0, if to process request one at a time
• Return Value
  • Promise

Examples

import { Amqp } from '@coolgk/amqp';
// OR
// const { Amqp } = require('@coolgk/amqp');

const amqp = new Amqp({
    url: 'amqp://localhost/vhost'
});

const message = {
    a: 1,
    b: 'b'
};

// CONSUMER MUST BE STARTED FIRST BEFORE PUSHLISHING ANY MESSAGE

// consumer.js
// consume message and return (send) a response back to publisher
amqp.consume(({rawMessage, message}) => {
    console.log('consumer received', message); // consumer received ignore response
                                               // consumer received { a: 1, b: 'b' }
    return {
        response: 'response message'
    }
});

// publisher.js
// publish a message, no response from consumer
amqp.publish('ignore response');

// publish a message and handle response from consumer
amqp.publish(message, ({rawResponseMessage, responseMessage}) => {
    console.log('response from consumer', responseMessage); // response from consumer { response: 'response message' }
});

Amqp

Kind: global class

• Amqp
  • new Amqp(options)
  • .closeConnection() ⇒ void
  • .publish(message, [callback], [options]) ⇒ promise.<boolean>
  • .consume(callback, [options]) ⇒ promise

new Amqp(options)

Param	Type	Description
options	object
options.url	string	connection string e.g. amqp://localhost
[options.sslPem]	string	pem file path
[options.sslCa]	string	sslCa file path
[options.sslPass]	string	password

amqp.closeConnection() ⇒ void

Kind: instance method of Amqp

amqp.publish(message, [callback], [options]) ⇒ promise.<boolean>

Kind: instance method of Amqp

Param	Type	Default	Description
message	*		message any type that can be JSON.stringify'ed
[callback]	function		callback(message) for processing response from consumers
[options]	object
[options.route]	string	"'#'"	route name
[options.exchangeName]	string	"'defaultExchange'"	exchange name

amqp.consume(callback, [options]) ⇒ promise

Kind: instance method of Amqp

Param	Type	Default	Description
callback	function		consumer(message) function should returns a promise
[options]	object
[options.route]	string	"'#'"	exchange route
[options.queueName]	string	"''"	queue name for processing request
[options.exchangeName]	string	"'defaultExchange'"	exchange name
[options.exchangeType]	string	"'topic'"	exchange type
[options.priority]	number	0	priority, larger numbers indicate higher priority
[options.prefetch]	number	0	1 or 0, if to process request one at a time