examples/src/example-amqp-consumer-batch.js
// And finally, we can set up, let's create a main method to hold our logic...
And finally, we can set up, let's create a main method to hold our logic
And finally we can set up let ' s create a main method to hold our logic
async function main() {
  await amqpCacoon.getConsumerChannel();
  await amqpCacoon.registerConsumerBatch(
    amqpConfig.exampleQueue,
    async (channelWrapper, msgBatch) => {
      try {
        logger.info(`Message Batch Count: ${msgBatch.messages.length}`);
        for (let msg of msgBatch.messages) {
          logger.info(`Message content: ${msg.content.toString()}`);
        }
        msgBatch.ackAll();
      } catch (e) {
        msgBatch.nackAll();
      }
    },
    {
      batching: {
        maxSizeBytes: 1000, 
        maxTimeMs: 5000, 
      },
    }
  );
}
async function main ( ) { await amqpCacoon . getConsumerChannel ( ) ; await amqpCacoon . registerConsumerBatch ( amqpConfig . exampleQueue , async ( channelWrapper , msgBatch ) => { try { logger . info ( ` ${ msgBatch . messages . length } ` ) ; for ( let msg of msgBatch . messages ) { logger . info ( ` ${ msg . content . toString ( ) } ` ) ; } msgBatch . ackAll ( ) ; } catch ( e ) { msgBatch . nackAll ( ) ; } } , { batching : { maxSizeBytes : 1000 , maxTimeMs : 5000 , } , } ) ; }
And finally, we can set up, let's create a main method to hold our logic...
And finally we can set up let ' s create a main method to hold our logic ...
// Connects and sets up a subscription channelWrapper // Register a consumer to consume single message at a time // ... Do other processing here // Loop through the batch and handle each as needed // Once processing is done, ACK them all! // This requires a bit of planning on your application, of course, to avoid duplicate processing! // Some error happened in our handling of the message batch. // The bet practice is to NACK all the messages so that some other process retries them! // This requires a bit of planning on your application, of course, to avoid duplicate processing! // A total of 1,000 Bytes will force consume // 5000ms = 5s elapsed will for a consume