node-redis/examples/stream-consumer-group.js

// A sample stream consumer using the blocking variant of XREADGROUP.
// This consumer works in collaboration with other instances of itself
// in the same consumer group such that the group as a whole receives
// every entry from the stream.
//
// This consumes entries from a stream created by stream-producer.js
//
// Run this as follows:
//
// $ node stream-consumer-group.js <consumerName>
//
// Run multiple instances with different values of <consumerName>
// to see them processing the stream as a group:
//
// $ node stream-consumer-group.js consumer1
//
// In another terminal:
//
// $ node stream-consumer-group.js consumer2

import { createClient, commandOptions } from 'redis';

async function streamConsumerGroup() {
  const client = createClient();

  if (process.argv.length !== 3) {
    console.log(`usage: node stream-consumer-group.js <consumerName>`);
    process.exit(1);
  }

  const consumerName = process.argv[2];

  await client.connect();

  // Create the consumer group (and stream) if needed...
  try {
    await client.xGroupCreate('mystream', 'myconsumergroup', '0', {
      MKSTREAM: true
    });
    console.log('Created consumer group.');
  } catch (e) {
    console.log('Consumer group already exists, skipped creation.');
  }

  console.log(`Starting consumer ${consumerName}.`);

  while (true) {
    try {
      let response = await client.xReadGroup(
        commandOptions({
          isolated: true
        }),
        'myconsumergroup', 
        consumerName, [
          // XREADGROUP can read from multiple streams, starting at a
          // different ID for each...
          {
            key: 'mystream',
            id: '>' // Next entry ID that no consumer in this group has read
          }
        ], {
          // Read 1 entry at a time, block for 5 seconds if there are none.
          COUNT: 1,
          BLOCK: 5000
        }
      );

      if (response) {
        // Response is an array of streams, each containing an array of
        // entries:
        //
        // [
        //   {
        //      "name": "mystream",
        //      "messages": [
        //        {
        //          "id": "1642088708425-0",
        //          "message": {
        //            "num": "999"
        //          }
        //        }
        //      ]
        //    }
        //  ]
        console.log(JSON.stringify(response));

        // Use XACK to acknowledge successful processing of this
        // stream entry.
        const entryId = response[0].messages[0].id;
        await client.xAck('mystream', 'myconsumergroup', entryId);

        console.log(`Acknowledged processing of entry ${entryId}.`);
      } else {
        // Response is null, we have read everything that is
        // in the stream right now...
        console.log('No new stream entries.');
      }
    } catch (err) {
      console.error(err);
    }
  }
}

streamConsumerGroup();