image-domain-exchange-notification-consumer.js

const AMQPHutch = require('amqp-hutch');
const log = require('../logger');

const options = {
  exchange: {
    name: 'images-domain-v1-rabbitmq',
    type: 'topic'
  },
  queue: {
    name: 'images.publish',
    prefetch: 1,
    durable: false
  }
};
const consumer = (message, done, fail) => {
  const entityUri = getEntityUri(JSON.parse(message.content.toString('utf-8')));

  if(entityUri){
    //TODO: retrirve image metadata from image repo using entity uri & begin ingestion process
  }

  done();
};

function getEntityUri(notification) {
  if (notification && notification.hasOwnProperty('entity') && notification.entity.uri) {
    return notification.entity.uri
  }

  return null;
}

let hutch = new AMQPHutch();

/**
 * class for consuming and processing image domain exchange notifications.
 */
class ImageDomainExchangeNotificationConsumer {

  constructor(connectionString, retryWait) {
    hutch.initialise({
                       connectionString: connectionString,
                       retryWait: retryWait
                     });
  }

  /**
   * consume and process notifications from image domain exchange.
   *
   */
  processImageNotifications() {
    hutch.on('ready', () => {
      log.info('Established RabbitMQ connection');
      hutch.consume(options, consumer, (err) => {
        if (err) {
          log.error(err);
        } else {
          log.info("Successfully setup notification consumer for queue: [" + options.queue.name + "]");
        }
      });
    });

    hutch.on('close', (err) => {
      log.error(err.message + ' RabbitMQ closing connection');
    });

    hutch.on('error', (err) => {
      log.error(err.message + ' RabbitMQ connection error');
    });
  }
}

module.exports = ImageDomainExchangeNotificationConsumer;