complainer.js

let messageQueue = require('./messageQueue');

function saySomething() {
  messageQueue.publish('sync', {hello: process.argv[2] || 'world', when: Date.now()});

  setTimeout(saySomething, 1000);
}

saySomething();

messageQueue.js

const amqplib = require('amqplib');

class MessageQueue {
  constructor() {
    this.connection = amqplib.connect(process.env.CLOUDAMQP_URL || 'amqp://localhost');
    this.channels   = {};
  }

  channel(queue) {
    if (!this.channels[queue]) {
      this.channels[queue] = this.connection
        .then(connection => connection.createChannel())
        .tap(channel => channel.assertQueue(queue))
        .tap(channel => channel.prefetch(1));
    }

    return this.channels[queue];
  }

  publish(queue, message) {
    this.channel(queue).then(channel => {
      channel.sendToQueue(queue, Buffer.from(JSON.stringify(message)));
    });
  }

  consume(queue, callback) {
    this.channel(queue).then(channel => {
      channel.consume(queue, message => {
        let decoded = message;

        if (decoded !== null) {
          decoded = JSON.parse(decoded.content.toString());
        }

        callback(decoded, () => {
          channel.ack(message);
        });
      });
    });
  }
}

module.exports = new MessageQueue();

package.json

{
  "name": "knijntje",
  "version": "1.0.0",
  "description": "",
  "main": "index.js",
  "scripts": {
    "test": "echo \"Error: no test specified\" && exit 1"
  },
  "keywords": [],
  "author": "",
  "license": "ISC",
  "dependencies": {
    "amqplib": "^0.5.1"
  }
}

worker.js

let messageQueue = require('./messageQueue');

messageQueue.consume('sync', (message, done) => {
  console.log(message);

  setTimeout(() => {
    console.log('ready for next message');

    done();
  }, 1500);
});