Redis Streams - model data as a log or a stream of events

Redis Streams is a feature that allows you to model data as a log or a stream of events. You can use it for various purposes, including event-driven architecture, message queuing, and real-time data processing. Here's an example of how you can use Redis Streams to implement a simple message queue:

const redis = require('redis');
const { promisify } = require('util');

const redisClient = redis.createClient();
const xaddAsync = promisify(redisClient.xadd).bind(redisClient);
const xreadgroupAsync = promisify(redisClient.xreadgroup).bind(redisClient);

const STREAM_NAME = 'message_stream';
const CONSUMER_GROUP_NAME = 'message_consumers';

// Create a stream (if it doesn't exist)
async function createStreamIfNotExists() {
  try {
    await redisClient.xgroup('CREATE', STREAM_NAME, CONSUMER_GROUP_NAME, '$', 'MKSTREAM');
  } catch (err) {
    // Ignore if the stream already exists
    if (!err.message.includes('BUSYGROUP Consumer Group name already exists')) {
      throw err;
    }
  }
}

// Produce a message to the stream
async function produceMessage(message) {
  const messageId = await xaddAsync(STREAM_NAME, '*', 'message', message);
  console.log(`Produced message with ID: ${messageId}`);
}

// Consume messages from the stream
async function consumeMessages() {
  const consumerName = 'consumer-1';

  while (true) {
    try {
      const messages = await xreadgroupAsync(
        'GROUP',
        CONSUMER_GROUP_NAME,
        consumerName,
        'BLOCK',
        0,
        'COUNT',
        10,
        'STREAMS',
        STREAM_NAME,
        '>',
      );

      for (const [stream, messageData] of messages) {
        for (const [messageId, message] of messageData) {
          console.log(`Received message with ID ${messageId}: ${message}`);
          // Process the message here

          // Acknowledge the message to remove it from the stream
          await redisClient.xack(STREAM_NAME, CONSUMER_GROUP_NAME, messageId);
        }
      }
    } catch (err) {
      console.error('Error consuming messages:', err);
    }
  }
}

(async () => {
  await createStreamIfNotExists();

  // Start consuming messages
  consumeMessages();

  // Produce some example messages
  for (let i = 1; i <= 10; i++) {
    await produceMessage(`Message ${i}`);
    await new Promise((resolve) => setTimeout(resolve, 1000)); // Delay between messages
  }
})();

In this example:

We create a Redis stream named message_stream and a consumer group named message_consumers.
The produceMessage function is used to produce messages to the stream, and the consumeMessages function consumes messages from the stream.
Messages are produced to the stream with unique IDs, and consumers can read and acknowledge (ack) them. Once a message is acknowledged, it's considered processed and removed from the stream.
The consumer script runs continuously, waiting for new messages to arrive in the stream. It processes each message and acknowledges it.

Redis Streams is a versatile feature that can be used for more complex use cases beyond simple message queues, including real-time event processing and log aggregation. You can customize and extend this example to fit your specific requirements.