// Import the queue module.
const QueueManager = require('@plgworks/queue');

// Config Strategy for PLG Works Queue.
configStrategy = {
	"rabbitmq": {
        "username": "guest",
        "password": "guest",
        "host": "127.0.0.1",
        "port": "5672",
        "heartbeats": "30",
        "enableRabbitmq": 1
    }
};

// Create instance
const queueManagerInstance = await QueueManager.getInstance(configStrategy);

// Config Strategy for PLG Works Queue.
configStrategy = {
	"rabbitmq": {
        "username": "guest",
        "password": "guest",
        "host": "127.0.0.1",
        "port": "5672",
        "heartbeats": "30",
        "enableRabbitmq": 1
    }
};

// Import the queue module.
const QueueManager = require('@plgworks/queue');
let unAckCount = 0; // Number of unacknowledged messages.

const subscribe = async function() {
  let queueManagerInstance = await QueueManager.getInstance(configStrategy);
  queueManagerInstance.subscribeEvent.rabbit(
    ["event.PublicTestEvent"], // List of events
    {
      queue: 'testQueue',
      ackRequired: 1, // When set to 1, all delivered messages MUST get acknowledge.
      broadcastSubscription: 1, // When set to 1, it will subscribe to broadcast channel and receive all broadcasted messages. 
      prefetch:10
    }, 
    function(msgContent){
      // Please make sure to return promise in callback function. 
      // On resolving the promise, the message will get acknowledged.
      // On rejecting the promise, the message will be re-queued (noAck)
      return new Promise(async function(onResolve, onReject) {
        // Incrementing unacknowledged message count.
        unAckCount++;
        console.log('Consumed message -> ', msgContent);
        response = await processMessage(msgContent);
        
        // Complete the task and in the end of all tasks done
        if(response == success){
          // The message MUST be acknowledged here.
          // To acknowledge the message, call onResolve
          // Decrementing unacknowledged message count.
          unAckCount--;
          onResolve();   
        } else {
          //in case of failure to requeue same message.
          onReject();
        }
       
      })
    
    });
};
// Gracefully handle SIGINT, SIGTERM signals.
// Once SIGINT/SIGTERM signal is received, programme will stop consuming new messages. 
// But, the current process MUST handle unacknowledged queued messages.
process.on('SIGINT', function () {
  console.log('Received SIGINT, checking unAckCount.');
  const f = function(){
    if (unAckCount === 0) {
      process.exit(1);
    } else {
      console.log('waiting for open tasks to be done.');
      setTimeout(f, 1000);
    }
  };
  setTimeout(f, 1000);
});

function rmqError(err) {
  console.log('rmqError occured.', err);
  process.emit('SIGINT');
}
// Event published from package in case of internal error.
process.on('rmq_error', rmqError);
subscribe();

// Config Strategy for PLG Works Queue.
configStrategy = {
	"rabbitmq": {
        "username": "guest",
        "password": "guest",
        "host": "127.0.0.1",
        "port": "5672",
        "heartbeats": "30",
        "enableRabbitmq": 1
    }
};

// Import the queue module.
const QueueManager = require('@plgworks/queue');
const subscribeMultiple = async function() {
  let queueManagerInstance = await QueueManager.getInstance(configStrategy);
  queueManagerInstance.subscribeEvent.rabbit(
    ["event.PublicTestEvent1", "event.PublicTestEvent2"],
    {}, 
    function(msgContent){
      console.log('Consumed message -> ', msgContent)
    });
  };
subscribeMultiple();

// Config Strategy for PLG Works Queue.
configStrategy = {
	"rabbitmq": {
        "username": "guest",
        "password": "guest",
        "host": "127.0.0.1",
        "port": "5672",
        "heartbeats": "30",
        "connectionTimeoutSec": "60",
        "enableRabbitmq": 1
    }
};

// Import the Queue module.
const QueueManager = require('@plgworks/queue');
const publish = async function() {
  let queueManagerInstance = await QueueManager.getInstance(configStrategy);
  queueManagerInstance.publishEvent.perform(
    {
      topics:["event.PublishTestEvent"],
      broadcast: 1, // When set to 1 message will be broadcasted to all channels.
      publishAfter: 1000, // message to be sent after milliseconds.
      publisher: 'MyPublisher',
      message: {
  	    kind: "event_received",
  	    payload: {
  		   // Custom payload for message
  	    }
  	  }
    });
};
publish();

// Config Strategy for PLG Works Queue.
let configStrategy = {
	"rabbitmq": {
        "username": "guest",
        "password": "guest",
        "host": "127.0.0.1",
        "port": "5672",
        "heartbeats": "30",
        "enableRabbitmq": 1
    }
};

let queueConsumerTag = null;
// Import the queue module.
const QueueManager = require('@plgworks/queue');
const subscribePauseRestartConsume = async function() {
  let queueManagerInstance = await QueueManager.getInstance(configStrategy);
  queueManagerInstance.subscribeEvent.rabbit(
      ["event.PublicTestEvent1", "event.PublicTestEvent2"],
      {}, 
      function(msgContent){
        console.log('Consumed message -> ', msgContent);
        
        if(some_failure_condition){
          process.emit('CANCEL_CONSUME', queueConsumerTag);
        }
        
        if(failure_resolve_detected){
          process.emit('RESUME_CONSUME', queueConsumerTag);
        }
      },
      function(consumerTag) {
        queueConsumerTag = consumerTag;
      }
    );
  };
subscribePauseRestartConsume();