Basic tutorial of how to use rabbitmq with nodejs

Author: learnercys

.gitignore

@@ -0,0 +1 @@
+node_modules

how-to-run.md

@@ -0,0 +1,72 @@
+# How To Run The Examples
+
+### Dependencies
+* RabbitMQ Host.
+* NodeJS.
+
+#### How to install RabbitMQ using docker
+Install docker run this command:
+```bash
+docker run -d --hostname my-rabbit --name some-rabbit -p 15672:15672 -p 5672:5672 rabbitmq:3-management
+```
+
+After that, you will be able to access to the management on `0.0.0.0:15672` or `localhost:15672` and have access to the services on the port `5672`.
+
+### Tutorial 1
+
+1. Open a terminal to run the consumer:
+```bash
+# this will show: [*] Waiting for messages. To exit press CTRL+C
+$ node t1-basic-consumer.js
+```
+
+2. Open other terminal to run the producer:
+```bash
+# this will show:  [x] Sent 'hello world'
+$ node t1-basic.js 'hello world.'
+```
+
+3. Wait 3 seconds and then you will see `[x] Received 'hello world'` on the consumers terminal.
+
+### Tutorial 2
+1. Open the terminal to run the consumer:
+```bash
+# this will show: [*] Waiting for logs. To exit press CTRL+C.
+$ node t2-direct-consumer.js SEVERITY
+```
+
+2. Open other terminal to run the producer:
+```bash
+# this will show: [x] Sent SEVERITY:'hellow world'
+$ node t2-direct.js SEVERITY 'hello world'
+```
+
+### Tutorial 3
+1. Open the terminal to run the consumer:
+```bash
+# This command will consume all the messages with a key that match with `logs.#`: `logs.info`, `logs.error`, `logs.warning`
+$ node t3-topics-consumer.js 'logs.#'
+```
+
+2. Open other terminal to run the producer:
+```bash
+$ node t3-topics.js 'logs.info' 'information message'
+$ node t3-topics.js 'logs.error' 'error message'
+$ node t3-topics.js 'logs.warning' 'warning message'
+```
+### Tutotial 4
+1. Open the terminal to run the consumer:
+```bash
+# this consumer will process events only that match with `logs.info`. Also
+# this consumer will only accept the messages equal to `hello` and reject the others.
+# All the messages rejected will be stored on `queue.dl`.
+$ node t4-deadletter-consumer.js 'logs.info'
+```
+
+2. Open other terminal to runt the producer:
+```bash
+# this message is accepted
+$ node t4-deadletter.js 'logs.info' 'hello'
+# and this is rejected.
+$ node t4-deadletter.js 'logs.info' 'hellos'
+```

package.json

@@ -0,0 +1,19 @@
+{
+  "name": "amqp-tutorial",
+  "version": "0.0.1",
+  "description": "amqp tutorial using rabbitmq",
+  "main": "index.js",
+  "scripts": {
+    "test": "echo \"Error: no test specified\" && exit 1"
+  },
+  "keywords": [
+    "rabbitmq",
+    "amqp"
+  ],
+  "author": "learnercys <[email protected]>",
+  "license": "ISC",
+  "dependencies": {
+    "amqplib": "^0.4.2",
+    "when": "^3.7.7"
+  }
+}

readme.md

@@ -0,0 +1,43 @@
+# AMQP -> RabbitMQ
+
+Advanced Message Queuing Protocol
+
+***
+
+### Producer
+Manage the messages to be published so they can be processed.
+
+### Consumer
+Receive the messages to process.
+
+***
+
+### Exchange
+Entity that receives the messages from the produces and enrute to its linked queues.
+
+##### Exchanges Types
+
+* Direct: Send messages to the linked queues based on the _routing key(Unicast)._
+* Fanout: Send messages to all the queues linked to the exchange.
+* Topic: Send messages to the linked queues based on the _routing key(Multicast)._
+* Headers: Designed to enrute multiples attributes(headers).
+
+***
+
+### Queue
+Store the messages consumed by the application.
+
+### Binding
+Rules used for the exchange to send messages to its linked queues.
+
+***
+
+### How Run the Examples
+See the document [how-to-run.md](how-to-run.md)
+
+### Reference:
+1. https://www.rabbitmq.com/tutorials/amqp-concepts.html
+2. http://www.squaremobius.net/amqp.node/channel_api.html
+3. https://github.com/squaremo/amqp.node
+4. https://www.rabbitmq.com/getstarted.html
+5. https://www.cloudamqp.com/blog/2015-05-18-part1-rabbitmq-for-beginners-what-is-rabbitmq.html

t1-basic-consumer.js

@@ -0,0 +1,24 @@
+var amqp = require('amqplib');
+
+amqp.connect('amqp://localhost').then(function (conn) {
+	return conn.createChannel().then(function (ch) {
+		// queue to consume
+		var q = 'queuet1';
+
+		// we need to make sure that the queue exists
+		var ok = ch.assertQueue(q, { durable: false});
+
+		ok = ok.then(function () {
+			// initialize the consume event 
+			return ch.consume(q, function (msg) {
+				setTimeout(function () {
+					console.log(" [x] Received '%s'", msg.content.toString());		
+				}, 3000);
+			}, { noAck: true });
+		});
+
+		return ok.then(function (_consumeOk) {
+			console.log(' [*] Waiting for messages. To exit press CTRL+C');
+		});
+	});
+}).then(null, console.warn);

t1-basic.js

@@ -0,0 +1,37 @@
+// This is a basic example of how we should publish messages
+// into rabbitmq, here we will not use directly an exchange but 
+// rabbitmq always use an exchange to process data, so the default 
+// exchange will be `direct` -> `amqp.direct`.=
+
+// RabbitMQ client
+var amqp = require('amqplib');
+var when = require('when');
+
+var args = process.argv.slice(2);
+
+// and this will be the message to send.
+var msg = args[0] || 'hello world';
+
+amqp.connect('amqp://localhost').then(function(conn) {
+	// here we use `when` to ensure that the connection is closed 
+	// after the `sendToQueue` event
+	return when(conn.createChannel().then(function(ch) {
+		// this will be the name of our new queue.
+		var q = 'queuet1';
+
+		// we need to be sure that the queue `queuet1` exists
+		// if not, the `assertQueue` method create it.
+		var ok = ch.assertQueue(q, {durable: false});
+
+		return ok.then(function(_qok) {
+			// after create | validate the existence of the queue
+			// we are ready to publish messages to it.
+			ch.sendToQueue(q, new Buffer(msg));
+			console.log(" [x] Sent '%s'", msg);
+
+			// we shouldn't need the channel anymore.
+			return ch.close();
+		});
+	})).ensure(function () { conn.close(); });
+}).then(null, console.warn);
+

t2-direct-consumer.js

@@ -0,0 +1,39 @@
+
+var amqp = require('amqplib');
+
+var args = process.argv.slice(2);
+var severity = args[0] || 'info';
+
+amqp.connect('amqp://localhost').then(function(conn) {
+	process.once('SIGINT', function() { conn.close(); });
+
+	return conn.createChannel().then(function(ch) {
+		var ex = 'direct_logs';
+
+		var ok = ch.assertExchange(ex, 'direct', {durable: false});
+
+		ok = ok.then(function() {
+			return ch.assertQueue('', {exclusive: true});
+		});
+
+		ok = ok.then(function(qok) {
+			var queue = qok.queue;
+			
+			return ch.bindQueue(queue, ex, severity).then(function() { return queue; });
+		});
+
+		ok = ok.then(function(queue) {
+			return ch.consume(queue, logMessage, {noAck: true});
+		});
+
+		return ok.then(function() {
+			console.log(' [*] Waiting for logs. To exit press CTRL+C.');
+		});
+
+		function logMessage(msg) {
+			console.log(" [x] %s:'%s'",
+									msg.fields.routingKey,
+									msg.content.toString());
+		}
+	});
+}).then(null, console.warn);

t2-direct.js

@@ -0,0 +1,24 @@
+
+var amqp = require('amqplib');
+var when = require('when');
+
+var args = process.argv.slice(2);
+
+// severity is the way that the consumer filter an event
+var severity = args[0] || 'info';
+var message = args[1] || 'Hello World!';
+
+amqp.connect('amqp://localhost').then(function(conn) {
+	return when(conn.createChannel().then(function(ch) {
+		// the exchange name used to receive all the messages
+		var ex = 'direct_logs';
+
+		var ok = ch.assertExchange(ex, 'direct', {durable: false});
+
+		return ok.then(function() {
+			ch.publish(ex, severity, new Buffer(message));
+			console.log(" [x] Sent %s:'%s'", severity, message);
+			return ch.close();
+		});
+	})).ensure(function() { conn.close(); });
+}).then(null, console.warn);

t3-topics-consumer.js

@@ -0,0 +1,34 @@
+
+var amqp = require('amqplib');
+
+var key = process.argv[2] || 'log.info';
+
+amqp.connect('amqp://localhost').then(function (conn) {
+	process.once('SIGINT', function() { conn.close(); });
+
+	return conn.createChannel().then(function(ch) {
+		var ex = 'topic_logs';
+		var ok = ch.assertExchange(ex, 'topic', {durable: false});
+		
+		ok = ok.then(function() {
+			return ch.assertQueue('', {exclusive: true});
+		});
+		
+		ok = ok.then(function(qok) {
+			var queue = qok.queue;
+			return ch.bindQueue(queue, ex, key).then(function() { return queue; });
+		});
+		
+		ok = ok.then(function(queue) {
+			return ch.consume(queue, logMessage, {noAck: true});
+		});
+
+		return ok.then(function() {
+			console.log(' [*] Waiting for logs. To exit press CTRL+C.');
+		});
+		
+		function logMessage(msg) {
+			console.log(" [x] %s:'%s'", msg.fields.routingKey, msg.content.toString());
+		}
+	});
+}).then(null, console.warn);

t3-topics.js

@@ -0,0 +1,23 @@
+
+var amqp = require('amqplib');
+var when = require('when');
+
+var args = process.argv.slice(2);
+var key = args[0] || 'log.info';
+var message = args[1] || 'Hello World!';
+
+amqp.connect('amqp://localhost').then(function(conn) {
+	return when(conn.createChannel().then(function(ch) {
+		var ex = 'topic_logs';
+		var ok = ch.assertExchange(ex, 'topic', {durable: false});
+
+		ok = ok.then(function() {
+			return ch.publish(ex, key, new Buffer(message));
+		});
+
+		return ok.then(function() {
+			console.log(" [x] Sent %s:'%s'", key, message);
+			return ch.close();
+		});
+	})).ensure(function() { conn.close(); })
+}).then(null, console.log);

t4-deadletter-consumer.js

@@ -0,0 +1,63 @@
+
+var amqp = require('amqplib');
+var basename = require('path').basename;
+
+var key = process.argv[2] || 'log.info';
+
+amqp.connect('amqp://localhost').then(function (conn) {
+	process.once('SIGINT', function() { conn.close(); });
+
+	return conn.createChannel().then(function (ch) {
+		var ex = 'topic_exchange';
+		var queue = 'queue';
+		var exdl = 'topic_exchange.dl';
+		var queuedl = 'queue.dl';
+
+		// dead letter exchange configs
+		var ok = ch.assertExchange(exdl, 'topic', {durable: true});
+
+		ok = ok.then(function() {
+			return ch.assertQueue(queuedl, {durable: true});
+		});
+
+		ok = ok.then(function() {
+			return ch.bindQueue(queuedl, exdl, key);
+		});
+
+		// exchange configs
+		ok = ok.then(function() {
+			return ch.assertExchange(ex, 'topic', {durable: true});
+		});
+
+		ok = ok.then(function () {
+			return ch.assertQueue(queue, {
+				durable: true,
+				arguments: {
+					'x-dead-letter-exchange': exdl
+				}
+			});
+		});
+
+		ok = ok.then(function () {
+			return ch.bindQueue(queue, ex, key).then(function() { return queue; });
+		});
+
+		ok = ok.then(function(queue) {
+			return ch.consume(queue, logMessage, {noAck: false});
+		});
+
+		return ok.then(function() {
+			console.log(' [*] Waiting for logs. To exit press CTRL+C.');
+		});
+
+		function logMessage(msg) {
+			console.log(" [x] %s:'%s'", msg.fields.routingKey, msg.content.toString());
+
+			if(msg.content.toString() !== 'hello') {
+				ch.reject(msg, false);
+			} else {
+				ch.ack(msg);
+			}
+		}
+	});
+}).then(null, console.warn);