Nostradamus Prophecy Machine

    @moxiedotxyz/queue

    1.0.0 • Public • Published

    Queue

    Latest version Downloads per month

    Moxie Queue helps publish critical events using EventEmitter and RabbitMQ. All events get published using node EventEmitter and, if configured, events are also published through RabbitMQ, using topic-based exchange.

    Install

    npm install @moxiedotxyz/queue --save

    Examples:

    Subscribe to events published through RabbitMQ:

    • Basic example on how to listen a specific event. Arguments passed are:
      • Events [Array] (mandatory) - List of events to subscribe to
      • Options [object] (mandatory) -
        • queue [string] (optional) - Name of the queue on which you want to receive all your subscribed events. These queues and events, published in them, have TTL of 6 days. If a queue name is not passed, a queue with a unique name is created and is deleted when the subscriber gets disconnected.
        • ackRequired [number] - (optional) - The delivered message needs ack if passed 1 ( default 0 ). if 1 passed and ack not done, message will redeliver.
        • prefetch [number] (optional) - The number of messages released from queue in parallel. In case of ackRequired=1, queue will pause unless delivered messages are acknowledged.
      • Callback [function] (mandatory) - Callback method will be invoked whenever there is a new notification
    // Config Strategy for Moxie Queue.
    configStrategy = {
    	"rabbitmq": {
            "username": "guest",
            "password": "guest",
            "host": "127.0.0.1",
            "port": "5672",
            "heartbeats": "30",
            "enableRabbitmq": 1
        }
    };
    // Import the queue module.
    const MoxieQueueManager = require('@moxiedotxyz/queue');
    let unAckCount = 0; // Number of unacknowledged messages.
    
    const subscribe = async function() {
      let moxieQueueManagerInstance = await MoxieQueueManager.getInstance(configStrategy);
      moxieQueueManagerInstance.subscribeEvent.rabbit(
        ["event.ProposedBrandedToken"],
        {
          queue: 'myQueue',
          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 moxieRmqError(err) {
      console.log('moxieRmqError occured.', err);
      process.emit('SIGINT');
    }
    // Event published from package in case of internal error.
    process.on('moxie_rmq_error', moxieRmqError);
    subscribe();
    • Example on how to listen to multiple events with one subscriber.
    // Config Strategy for Moxie Queue.
    configStrategy = {
    	"rabbitmq": {
            "username": "guest",
            "password": "guest",
            "host": "127.0.0.1",
            "port": "5672",
            "heartbeats": "30",
            "enableRabbitmq": 1
        }
    };
    // Import the queue module.
    const MoxieQueueManager = require('@moxiedotxyz/queue');
    const subscribeMultiple = async function() {
      let moxieQueueManagerInstance = await MoxieQueueManager.getInstance(configStrategy);
      moxieQueueManagerInstance.subscribeEvent.rabbit(
        ["event.ProposedBrandedToken", "obBoarding.registerBrandedToken"],
        {}, 
        function(msgContent){
          console.log('Consumed message -> ', msgContent)
        });
      };
    subscribeMultiple();

    Subscribe to local events published through EventEmitter:

    • Basic example on how to listen a specific event. Arguments passed are:
      • Events (mandatory) - List of events to subscribe to
      • Callback (mandatory) - Callback method will be invoked whenever there is a new notification
    // Config Strategy for Moxie Queue.
    configStrategy = {
    	"rabbitmq": {
            "username": "guest",
            "password": "guest",
            "host": "127.0.0.1",
            "port": "5672",
            "heartbeats": "30",
            "enableRabbitmq": 1
        }
    };
    // Import the queue module.
    const MoxieQueueManager = require('@moxiedotxyz/queue');
    const subscribeLocal = async function() {
      let moxieQueueManagerInstance = await MoxieQueueManager.getInstance(configStrategy);
      moxieQueueManagerInstance.subscribeEvent.local(["event.ProposedBrandedToken"], 
      function(msgContent){
        console.log('Consumed message -> ', msgContent)
      });
      };
    subscribeLocal();

    Publish Notifications:

    • All events are by default published using EventEmitter and if configured, through RabbitMQ as well.
    // Config Strategy for Moxie 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 MoxieQueueManager = require('@moxiedotxyz/queue');
    const publish = async function() {
      let moxieQueueManagerInstance = await MoxieQueueManager.getInstance(configStrategy);
      moxieQueueManagerInstance.publishEvent.perform(
        {
          topics:["event.ProposedBrandedToken"],
          broadcast: 1, // When set to 1 message will be broadcasted to all channels. 'topics' parameter should not be sent.
          publishAfter: 1000, // message to be sent after milliseconds.
          publisher: 'MyPublisher',
          message: {
      	  kind: "event_received",
      	  payload: {
      		event_name: 'ProposedBrandedToken',
      		params: {
      		  //params of the event
      		},
              contract_address: 'contract address',
              chain_id: 'Chain id',
              chain_kind: 'kind of the chain'
      	  }
      	}
        });
    };
    publish();

    Pause and Restart queue consumption:

    • We also support pause and start queue consumption. According to your logical condition, you can fire below events from your process to cancel or restart consumption respectively.
    // Config Strategy for Moxie 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 MoxieQueueManager = require('@moxiedotxyz/queue');
    const subscribePauseRestartConsume = async function() {
      let moxieQueueManagerInstance = await MoxieQueueManager.getInstance(configStrategy);
      moxieQueueManagerInstance.subscribeEvent.rabbit(
        ["event.ProposedBrandedToken", "obBoarding.registerBrandedToken"],
        {}, 
        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();

    Install

    npm i @moxiedotxyz/queue

    DownloadsWeekly Downloads

    23

    Version

    1.0.0

    License

    LGPL-3.0

    Unpacked Size

    69.8 kB

    Total Files

    27

    Last publish

    Collaborators

    • avatar
    • avatar
    • avatar
    • avatar