Welcome! In this unit, we will explore how to use Redis streams for event logging. This is an important part of our Redis-based backend system project. By the end of this lesson, you will know how to log events and retrieve them using Redis streams. Remember, you've already learned how to manage user data and handle transactions. Now, we're adding another layer to our system by using streams.

In this unit, we will focus on the following tasks:

1. Adding entries to a stream: We will log user activities in a Redis stream.
2. Reading entries from a stream: You will see how to read the logged events from the stream.

To log user activities, we use the xadd command in Redis. In our Java implementation, this is handled by the addToStream method in the Main class. Here's how it works:

// Function to add entry to stream with pipeline
public static void addToStream(Pipeline pipeline, String streamName, User user) {
    pipeline.xadd(
        streamName,
        XAddParams.xAddParams(), // Auto-generate an ID
        Map.of(
            "event", "user_added",
            "username", user.getUsername(),
            "data", gson.toJson(user.getData()),
            "score", String.valueOf(user.getScore())
        )
    );
}

This method logs an event to a Redis stream with fields such as event, username, data, and score. Each event is auto-assigned a unique ID by Redis.

In the Main class, this method is called for multiple users within a pipeline:

try (Pipeline pipeline = jedis.pipelined()) {
    for (User user : new User[]{user1, user2, user3}) {
        addToStream(pipeline, streamName, user);
    }
    pipeline.sync();
}

This ensures that all events are logged efficiently in one batch operation.

To retrieve logged events, we use the xread command in Redis. The readFromStream method in the Main class demonstrates this:

// Function to read from stream
public static List<Entry<String, List<StreamEntry>>> readFromStream(Jedis jedis, String streamName) {
    // We want to read from the beginning (0-0)
    Map<String, StreamEntryID> streams = Map.of(streamName, new StreamEntryID("0-0"));
    // Provide XReadParams (e.g., count, block) if needed. Here, we use default.
    return jedis.xread(XReadParams.xReadParams(), streams);
}

This method reads entries from the specified stream starting from the beginning (0-0). The XReadParams object allows for customization, such as setting a maximum count of entries to read.

In the Main class, the retrieved entries are printed as follows:

List<Entry<String, List<StreamEntry>>> streamEntries = readFromStream(jedis, streamName);
System.out.println("Stream entries:");
for (Entry<String, List<StreamEntry>> entry : streamEntries) {
    for (StreamEntry se : entry.getValue()) {
        System.out.println("Stream: " + entry.getKey()
                + ", ID: " + se.getID()
                + ", Fields: " + se.getFields());
    }
}

This code iterates through the retrieved entries and prints each event, including its stream, ID, and fields.