Working with Redis Streams: Adding and Reading Events | CodeSignal Learn

Working with Redis Streams: Adding and Reading Events | CodeSignal Learn

In this lesson, we'll learn about streams in Redis and how they can be used to handle continuous data flows. We'll learn how to create streams, add events to them, and read events from them.

Streams are a powerful data structure that allows you to process real-time data efficiently. Here are a few real-world scenarios where Redis Streams can be useful:

Chat Applications: Streams can be used to handle messages in real time.
Monitoring Systems: Streams can be used to process logs and events.
User Activity Tracking: Streams can be used to track user actions in real time.

Let's dive into the details!

Here’s a quick preview:

To add an event to a stream, you can use the xadd command.
To read events from a stream, you can use the xrange command.

Let's see how these commands work in practice.

JavaScript

import Redis from 'ioredis';

// Connect to Redis
const client = new Redis('redis://localhost:6379');

client.on('error', (err) => {
  console.error('Error connecting to Redis', err);
});

async function run() {
  // Add events to the stream
  await client.xadd('mystream', '*', 'event', 'login', 'user', 'Alice');
  await client.xadd('mystream', '*', 'event', 'purchase', 'user', 'Bob', 'amount', '100');
  await client.xadd('mystream', '*', 'event', 'add_to_cart', 'user', 'Alice', 'product', 'laptop');

  // Read up to 2 messages from the stream
  const messages = await client.xrange('mystream', '-', '+', 'COUNT', 2);
  console.log(`Stream messages: ${JSON.stringify(messages)}`);

  // Access the first message's data
  const firstMessage = messages[0];
  // firstMessage is an array: [id, [field1, value1, field2, value2, ...]]
  const id = firstMessage[0];
  const fieldsArray = firstMessage[1];
  // Convert fieldsArray to an object for easier use
  const messageObject = {};
  for (let i = 0; i < fieldsArray.length; i += 2) {
    messageObject[fieldsArray[i]] = fieldsArray[i + 1];
  }
  console.log(`First message (ID: ${id}): ${JSON.stringify(messageObject)}`); // {"event":"login", "user":"Alice"}

  await client.quit();
}

run();

The above code snippet demonstrates how to add events to a Redis stream called mystream. Each event contains key-value pairs representing different actions by users.

It's important to understand that the '*' used as the second argument in xadd is a placeholder for the message ID, which tells Redis to auto-generate a unique ID based on the current timestamp. If you omit '*' or provide a manually defined ID, you risk potential ID conflicts or ordering issues, especially in distributed environments. Also, note that these auto-generated IDs ensure messages are appended in the order they are received, which is essential for stream processing use cases.

This code reads up to the first two messages from mystream and prints them. The '-' and '+' arguments to xrange indicate reading messages from the beginning to the end of the stream, while the 'COUNT', 2 option limits the number of messages retrieved.

Notice that each message returned by xrange is an array where the first element is the message ID and the second element is an array of field-value pairs. To access the actual event data, you can convert this array into an object as shown above. This conversion is necessary because Redis does not store the message content as a plain object—it stores it as an ordered array of field-value pairs. By iterating over the array in steps of 2, we reconstruct the key-value mapping in a JavaScript object, making it easier to work with in typical application logic. This pattern is essential whenever dealing with Redis stream responses programmatically.