var RabbitMQStream = require("rabbitmq-queue-stream");
var stream         = require("stream");

var options = {
  connection: {
    url: "amqp://user:[email protected]"
  },
  queueStream: {
    name: "myQueue",
    prefetchCount: 100
  }
};

/*
 * Initialize two consumer channels to our queue.
*/
RabbitMQStream.init(2, options, function(err, streamifiedQueues) {
  if(err) {
    return console.error(err);
  }

  /*
   * Each consumer channel comes with a .source and .sink property.
   * 
   * .source is a Readable stream that gives us a stream of objects
   * from the specified queue
   *
   * Every job written to .sink is deleted from the queue. Only object
   * originating from .source should be written to .sink
   *
  */

  streamifiedQueues.channels.forEach(function(channel) {

    var myProcessingStream = new stream.Transform({objectMode: true});
    myProcessingStream._transform(function(data, enc, next) {
      console.log("Doing something with", data);
      this.push(data);
      /*
       * Messages are successfully acked and removed from the queue by default.
       * RabbitMQStream provides methods to requeue and delete messages too.
       *
       * Requeue:
       *     this.push(RabbitMQStream.RequeueMessage(data));
       *
       * Delete:
       *     this.push(RabbitMQStream.DeleteMessage(data));
      */
      next();
    });

    channel.source
      .pipe(myProcessingStream)
      .pipe(channel.sink);
  });

  /* example graceful shutdown routine */
  var gracefulShutdown = function() {
    //stop fetching messages
    streamifiedQueues.unsubscribeConsumers(function(err) {
      if(err) {
        //handle error
      }
      //Wait some time for queues to flush out before closing consumers.
      streamifiedQueues.closeConsumers(function(err) {
        if(err) {
          //handle error
        }
        streamifiedQueues.disconnect(function(err) {
          if(err) {
            //handle error
          }
          process.exit(0);
        });
      });
    });
  };
});

myQueueStream.source.on("parseError", function(err, message) {
  console.error("Problem JSON parsing message", message);
});

var totalDeleted = 0;
myQueueStream.source.on("deleted", function() {
  console.log("Deleted", totalDeleted++);
});

myQueueStream.sink.on("formatError", function(err, message) {
  console.error("Malformatted message written to .sink. Please check your pipeline configuration", message);
});