add wrapper that can reconnect

Author: goFrendiAsgard

labs/publisher.js

@@ -0,0 +1,38 @@
+const amqp = require('amqp-connection-manager');
+
+const QUEUE_NAME = 'amqp-connection-manager-sample2'
+const EXCHANGE_NAME = 'amqp-connection-manager-sample2-ex';
+
+// Create a connetion manager
+const connection = amqp.connect(["amqp://root:[email protected]:5672/"], {json: true});
+connection.on('connect', () => console.log('Connected!'));
+connection.on('disconnect', params => console.log('Disconnected.', params.err.stack));
+
+// Create a channel wrapper
+const channelWrapper = connection.createChannel({
+    json: true,
+    setup: channel => channel.assertExchange(EXCHANGE_NAME, 'topic')
+});
+
+// Send messages until someone hits CTRL-C or something goes wrong...
+function sendMessage() {
+    const msg = {time: Date.now()};
+    channelWrapper.publish(EXCHANGE_NAME, "test", msg, { contentType: 'application/json', persistent: true })
+    .then(function() {
+        console.log("Message sent", msg);
+    })
+    .then(() => {
+        return new Promise((resolve, reject) => {
+            setTimeout(() => resolve(), 1000);
+        });
+    })
+    .then(() => sendMessage())
+    .catch(err => {
+        console.log("Message was rejected:", err.stack);
+        channelWrapper.close();
+        connection.close();
+    });
+};
+
+console.log("Sending messages...");
+sendMessage();

labs/subscriber.js

@@ -0,0 +1,34 @@
+const amqp = require('amqp-connection-manager');
+
+const QUEUE_NAME = 'amqp-connection-manager-sample2'
+const EXCHANGE_NAME = 'amqp-connection-manager-sample2-ex';
+
+// Handle an incomming message.
+const onMessage = data => {
+    var message = JSON.parse(data.content.toString());
+    console.log("subscriber: got message", message);
+    channelWrapper.ack(data);
+}
+
+// Create a connetion manager
+const connection = amqp.connect(["amqp://root:[email protected]:5672/"], {json: true});
+connection.on('connect', () => console.log('Connected!'));
+connection.on('disconnect', params => console.log('Disconnected.', params.err.stack));
+
+// Set up a channel listening for messages in the queue.
+var channelWrapper = connection.createChannel({
+    setup: channel =>
+        // `channel` here is a regular amqplib `ConfirmChannel`.
+        Promise.all([
+            channel.assertQueue(QUEUE_NAME, { exclusive: true, autoDelete: true }),
+            channel.assertExchange(EXCHANGE_NAME, 'topic'),
+            channel.prefetch(1),
+            channel.bindQueue(QUEUE_NAME, EXCHANGE_NAME, '#'),
+            channel.consume(QUEUE_NAME, onMessage)
+        ])
+});
+
+channelWrapper.waitForConnect()
+.then(function() {
+    console.log("Listening for messages");
+});

labs/test1.js

@@ -0,0 +1,69 @@
+const q = 'tasks';
+
+const { execSync } = require("child_process");
+const amqplib = require("amqplib");
+
+async function createConnection() {
+    const id = Math.round(Math.random() * 1000);
+    console.log(`* connect ${id}`);
+    const pconn = amqplib.connect("amqp://root:[email protected]:5672/").then((conn) => {
+        conn.on("close", (error) => {
+            console.log(`* connection closed ${id}:`, error);
+        });
+        conn.on("error", (error) => {
+            console.log(`* connection error ${id}:`, error);
+        });
+        return conn;
+    });
+    console.log(`* connected ${id}`);
+    return pconn;
+}
+
+async function createChannel(conn) {
+    const id = Math.round(Math.random() * 1000);
+    console.log(`* create channel ${id}`);
+    const channel = conn.createChannel({durable: false});
+    console.log(`* channel created ${id}`);
+    return channel;
+}
+
+async function createConnectionAndChannel() {
+    const conn = await createConnection();
+    const channel = await createChannel(conn);
+    return channel;
+}
+
+async function main(channels) {
+
+    if (!channels) {
+        channels = await Promise.all([createConnectionAndChannel(), createConnectionAndChannel()]);
+    }
+    const [publisherChannel, consumerChannel] = channels;
+
+    // publisher
+    console.log("publish");
+    await publisherChannel.assertQueue(q).then(function(ok) {
+        console.log("published");
+        return publisherChannel.sendToQueue(q, Buffer.from("something to do"));
+    });
+
+    // consumer
+    console.log("consume");
+    await consumerChannel.assertQueue(q).then(function(ok) {
+        console.log("consumed");
+        return consumerChannel.consume(q, function(msg) {
+            if (msg !== null) {
+                console.log(msg.content.toString());
+                consumerChannel.ack(msg);
+            }
+        });
+    });
+
+    await new Promise((resolve, reject) => {
+        setTimeout(resolve, 1000);
+    });
+
+    await main(channels);
+}
+
+main();