Step 6 - Add queue worker

Author: mattheworiordan

package.json

@@ -12,6 +12,7 @@
     },
     "dependencies": {
         "ably": ">=0.9.0-beta",
+        "amqplib": "^0.4",
         "express": ">=4.14.0"
     }
 }

server.js

@@ -1,6 +1,7 @@
 const Ably = require('ably');
 const Express = require('express');
 const ServerPort = 3000;
+const worker = require('./worker');
 
 const ApiKey = 'INSERT-YOUR-API-KEY-HERE'; /* Add your API key here */
 if (ApiKey.indexOf('INSERT') === 0) { throw('Cannot run without an API key. Add your key to server.js'); }
@@ -27,4 +28,6 @@ app.get('/auth', function (req, res) {
 app.use(Express.static('public'));
 app.listen(3000);
 
-console.log('Web server listening on port', ServerPort);
+worker.start(ApiKey, 'wolfram:answers', 'wolfram', 'us-east-1-a-queue.ably.io:5671/shared');
+
+console.log('Open the Wolfram demo in your browser: https://localhost:' + ServerPort + '/');

worker.js

@@ -0,0 +1,52 @@
+'use strict';
+
+const amqp = require('amqplib/callback_api');
+const Ably = require('ably');
+
+/* Start the worker that consumes from the AMQP QUEUE */
+exports.start = function(apiKey, answersChannelName, queueName, queueEndpoint) {
+  const appId = apiKey.split('.')[0];
+  const queue = appId + ":" + queueName;
+  const endpoint = queueEndpoint;
+  const url = 'amqps://' + apiKey + '@' + endpoint;
+  const rest = new Ably.Rest({ key: apiKey });
+  const answersChannel = rest.channels.get(answersChannelName);
+
+  /* Connect to Ably queue */
+  amqp.connect(url, (err, conn) => {
+    if (err) {
+      console.error('worker:', 'Queue error!', err);
+      return;
+    }
+    console.log('worker:', 'Connected to AMQP endpoint', endpoint);
+
+    /* Create a communication channel */
+    conn.createChannel((err, ch) => {
+      if (err) {
+        console.error('worker:', 'Queue error!', err);
+        return;
+      }
+      console.log('worker:', 'Waiting for messages');
+
+      /* Wait for messages published to the Ably Reactor queue */
+      ch.consume(queue, (item) => {
+        const decodedEnvelope = JSON.parse(item.content);
+
+        const messages = Ably.Realtime.Message.fromEncodedArray(decodedEnvelope.messages);
+        messages.forEach(function(message) {
+          console.log('worker:', 'Received question', message.data, ' - about to ask Wolfram');
+          answersChannel.publish('answer', 'Placeholder until Wolfram connected', function(err) {
+            if (err) {
+              console.error('worker:', 'Failed to publish question', message.data, ' - err:', JSON.stringify(err));
+            }
+          })
+        });
+
+        /* Remove message from queue */
+        ch.ack(item);
+      });
+    });
+
+    conn.on('error', function(err) { console.error('worker:', 'Connection error!', err); });
+  });
+};