Now, let's start to implement our system by creating a new module for the ventilator, in a file named ventilator.js:

const zmq = require('zmq'); 
const variationsStream = require('variations-stream'); 
const alphabet = 'abcdefghijklmnopqrstuvwxyz'; 
const batchSize = 10000; 
const maxLength = process.argv[2]; 
const searchHash = process.argv[3]; 
 
const ventilator = zmq.socket('push');             // [1] 
ventilator.bindSync("tcp://*:5000"); 
 
let batch = []; 
variationsStream(alphabet, maxLength) 
  .on('data', combination => { 
    batch.push(combination); 
    if (batch.length === batchSize) {              // [2] 
      const msg = {searchHash: searchHash, variations: batch}; 
      ventilator.send(JSON.stringify(msg)); 
      batch = []; 
    } 
  }) 
  .on('end', () => { 
    //send remaining combinations 
    const msg = {searchHash: searchHash, variations: batch}; 
    ventilator.send(JSON.stringify(msg)); 
  }); 

To avoid generating too many variations, our generator uses only the lowercase letters of the English alphabet and sets a limit on the size of the words generated. This limit is provided in input as a command-line argument (maxLength) together with the hashsum to match (searchHash). We use a library called variations-stream (https://npmjs.org/package/variations-stream) to generate all the variations using a streaming interface.

But the part that we are most interested in analyzing is how we distribute the tasks across the workers:

Now it's time to implement the worker (worker.js):

const zmq = require('zmq'); 
const crypto = require('crypto'); 
const fromVentilator = zmq.socket('pull'); 
const toSink = zmq.socket('push'); 
 
fromVentilator.connect('tcp://localhost:5016'); 
toSink.connect('tcp://localhost:5017'); 
 
fromVentilator.on('message', buffer => { 
  const msg = JSON.parse(buffer); 
  const variations = msg.variations; 
  variations.forEach( word => { 
    console.log(`Processing: ${word}`); 
    const shasum = crypto.createHash('sha1'); 
    shasum.update(word); 
    const digest = shasum.digest('hex'); 
    if (digest === msg.searchHash) { 
      console.log(`Found! => ${word}`); 
      toSink.send(`Found! ${digest} => ${word}`); 
    } 
  }); 
}); 

As we said, our worker represents a transient node in our architecture, therefore, its sockets should connect to a remote node instead of listening for the incoming connections. That's exactly what we do in our worker, we create two sockets:

Besides this, the job done by our worker is very simple: for each message received, we iterate over the batch of words it contains, then for each word we calculate the SHA1 checksum and we try to match it against searchHash passed with the message. When a match is found, the result is forwarded to the sink.

For our example, the sink is a very basic result collector, which simply prints the messages received by the workers to the console. The contents of the file sink.js are as follows:

const zmq  = require('zmq'); 
const sink = zmq.socket('pull'); 
sink.bindSync("tcp://*:5017"); 
 
sink.on('message', buffer => { 
  console.log('Message from worker: ', buffer.toString()); 
}); 

It's interesting to see that the sink (as the ventilator) is also a durable node of our architecture and therefore we bind its PULL socket instead of connecting it explicitly to the PUSH socket of the workers.

We are now ready to launch our application; let's start a couple of workers and the sink:

node worker
node worker
node sink

Then it's time to start the ventilator, specifying the maximum length of the words to generate and the SHA1 checksum that we want to match. The following is a sample list of arguments:

node ventilator 4 f8e966d1e207d02c44511a58dccff2f5429e9a3b

When the preceding command is run, the ventilator will start generating all the possible words that have a length of, at most, four characters, distributing them to the set of workers we started, along with the checksum we provided. The results of the computation, if any, will appear in the terminal of the sink application.

Let's see how to implement such a system, starting from the producer (the variation generator). Its code is identical to the sample we saw in the previous section except for the parts concerning the message exchange. The producer.js file will look as follows:

const amqp = require('amqplib'); 
//... 
 
let connection, channel; 
amqp 
  .connect('amqp://localhost') 
  .then(conn => { 
    connection = conn; 
    return conn.createChannel(); 
  }) 
  .then(ch => { 
    channel = ch; 
    produce(); 
  }) 
  .catch(err => console.log(err)); 
 
function produce() { 
  //... 
  variationsStream(alphabet, maxLength) 
    .on('data', combination => { 
      //... 
      const msg = {searchHash: searchHash, variations: batch}; 
      channel.sendToQueue('jobs_queue', 
        new Buffer(JSON.stringify(msg))); 
      //... 
    }) 
  //... 
} 

As we can see, the absence of any exchange or binding makes the setup of an AMQP communication much simpler. In the preceding code, we didn't even need a queue, as we are interested only in publishing a message.

The most important detail, though, is the channel.sendToQueue() API, which is actually new to us. As its name says, that's the API responsible for delivering a message straight to a queue—jobs_queue in our example—bypassing any exchange or routing.

On the other side of jobs_queue, we have the workers listening for the incoming tasks. Let's implement their code in a file called worker.js, as follows:

const amqp = require('amqplib'); 
//... 
 
let channel, queue; 
amqp 
  .connect('amqp://localhost') 
  .then(conn => conn.createChannel()) 
  .then(ch => { 
    channel = ch; 
    return channel.assertQueue('jobs_queue'); 
  }) 
  .then(q => { 
    queue = q.queue; 
    consume(); 
  }) 
 
//... 
 
function consume() { 
  channel.consume(queue, msg => { 
    //... 
    variations.forEach(word => { 
      //... 
      if(digest === data.searchHash) { 
        console.log(`Found! => ${word}`); 
        channel.sendToQueue('results_queue', 
          new Buffer(`Found! ${digest} => ${word}`)); 
      } 
     //... 
    }); 
    channel.ack(msg); 
  }); 
}; 

Our new worker is also very similar to the one we implemented in the previous section using ØMQ, except for the part related to the message exchange. In the preceding code, we can see how we first make sure that jobs_queue exists and then we start listening for incoming tasks using channel.consume(). Then, every time a match is found, we send the result to the collector via results_queue, using again a point-to-point communication.

If multiple workers are started, they will all listen on the same queue, resulting in the messages being load balanced between them.

The results collector is again a trivial module, simply printing any message received to the console. This is implemented in the collector.js file, as follows:

//... 
  .then(ch => { 
    channel = ch; 
    return channel.assertQueue('results_queue'); 
  }) 
  .then(q => { 
    queue = q.queue; 
    channel.consume(queue, msg => { 
      console.log('Message from worker: ', msg.content.toString()); 
    }); 
  }) 
//... 

Now everything is ready to give our new system a try, we can start by running a couple of workers, which will both connect to the same queue (jobs_queue), so that every message will be load balanced between them:

node worker
node worker

Then, we can run the collector module and then producer (by providing the maximum word length and the hash to crack):

node collector
node producer 4 f8e966d1e207d02c44511a58dccff2f5429e9a3b

With this, we implemented a message pipeline and the competing consumers pattern using nothing but AMQP.