DEV Community: Krisjanis Kallings

Redis Streams + NestJS: Part 3 | Consumer groups

Krisjanis Kallings — Wed, 22 Feb 2023 17:00:06 +0000

Intro

This is part 3 of a 3-part series, where we will explore how to use Redis streams with NestJS.

It is structured in 3 parts:

Part 1 - Setting up the NestJS application and connecting to Redis
Part 2 - Populating Redis streams and reading from in fan-out mode
Part 3 - Using consumer groups to handle one stream from multiple actors in a way that one message is sent to and processed only by a single actor (consumer)

By the end of this series, you will have the knowledge and tools necessary to create your own NestJS app that utilizes Redis streams to handle real-time data.

Full code is available on the github

What we have

In the previous parts, we created a NestJS application and connected it to the Redis server. Then, we implemented functionality for adding messages to the Redis stream and reading from it in a fan-out fashion. Finally, we looked at how we could use the Redis stream data structure for real-time data stream processing by continuously fetching messages from the Redis stream.

In this post, we will build upon this idea by using consumer groups, thus adding the ability for distributed data processing.

Understanding consumer groups

In Redis, consumer groups are a way to distribute messages or tasks across multiple consumers while ensuring that each message is processed only once.
When you have many messages to process, a single consumer might need help to keep up with the incoming message rate. You can create multiple consumer instances to share the workload in this scenario.

Redis consumer groups enable you to distribute messages to multiple consumers in a coordinated, fault-tolerant way.
Each consumer in the group is assigned a unique name and a subset of the stream's messages to process. Redis uses a mechanism called "message acknowledgment" to ensure that each message is processed only once, even if there are multiple consumers in the group.

Using consumer groups

We used the XREAD command to fetch data from a stream. If we want to use consumer groups, we will need to use a different command - XREADGROUP. You can find general info on Redis Docs

Reading as a consumer

Let's create a method in RedisService for reading data in a consumer group manner:

// redis.service.ts

 public async readConsumerGroup({
    streamName,
    group,
    consumer,
    blockMs,
    count,
  }: CosnumeStreamParams): Promise<RedisStreamMessage[] | null> {
    let response: RedsXReadGroupResponse = null;
    try {
      response = await this.redis.xReadGroup(
        commandOptions({ isolated: true }), // uses new connection from pool not to block other redis calls
        group,
        consumer,
        {
          key: streamName,
          id: '>',
        },
        { BLOCK: blockMs, COUNT: count },
      );
    } catch (error) {
      if (error instanceof ClientClosedError) {
        console.log(`${error.message} ...RECONNECTING`);
        await this.connectToRedis();
        return null;
      }
      if (error.message.includes('NOGROUP')) {
        console.log(`${error.message} ...CREATING GROUP`);
        await this.createConsumerGroup(streamName, group);
        return null;
      }
      console.error(
        `Failed to xReadGroup from Redis stream: ${error.message}`,
        error,
      );

      return null;
    }

    const messages = response?.[0]?.messages; // returning first stream (since only 1 stream used)
    return messages || null;
  }

We have introduced a new interface for params CosnumeStreamParams. Since this shares many similarities with ReadStreamParams, we extracted those to a base interface, StreamParamsBase, and extended them from both of these interfaces.

// interfaces.ts

export interface StreamParamsBase {
  /** Name of stream to read from */
  streamName: string;
  /** Max time in ms for how long to block Redis connection before returning
   * If 0 is passed, it will block until at least one message is fetched, or timeout happens
   * */
  blockMs: number;
  /** Max how many messages to fetch at a time from Redis */
  count: number;
}

export interface ReadStreamParams extends StreamParamsBase {
  /** ID of last fetched message */
  lastMessageId: string;
}

export interface CosnumeStreamParams extends StreamParamsBase {
  /** Name of consumer group */
  group: string;
  /** Name of consumer, must be unique within group */
  consumer: string;
}
// --snip--

The resulting stream message - RedisStreamMessage stays the same as we had for reading the stream.
Similarly, as we did for XREAD, we have extracted a response type RedsXReadGroupResponse

// redis-client.type.ts

export type RedsXReadGroupResponse = Awaited<
  ReturnType<RedisClient['xReadGroup']>

The code should be straight forward:

// redis.service.ts

// --snip--

    let response: RedsXReadGroupResponse = null;
    try {
      response = await this.redis.xReadGroup(
        commandOptions({ isolated: true }), // uses new connection from pool not to block other redis calls
        group,
        consumer,
        {
          key: streamName,
          id: '>',
        },
        { BLOCK: blockMs, COUNT: count },
      );
    }
// --snip--

We call xReadGroup method of node-redis client, and wrap it in try-catch to handle any error we might get from RedisClient.

Same as with xRead, when using blocking commands, we want to utilize connection pool and execute these commands in isolation via commandOptions({ isolated: true })

Then we pass the name of the consumer group this consumer belongs to and is sharing messages with.
Next, we need to identify the consumer - entity reading this message.

Next, we specify from which stream we want to fetch a stream by key parameter and starting from which message ID. > is a special symbol meaning we want to fetch only messages that were never delivered to any other consumer. If we used 0 or any other valid ID, we would risk also getting messages that have been given to other users but have yet to be acknowledged.

NOTE!
Here, we are specifying only 1 stream. However, we could read from multiple streams by adding an array of objects containing key and id values.

Finally, we set read options. BLOCK for max blocking time in ms and COUNT for the max count of messages. These work exactly the same as with XREAD option.

Next we have some error handling to do:

// redis.service.ts

// --snip--
   } catch (error) {
      if (error instanceof ClientClosedError) {
        console.log(`${error.message} ...RECONNECTING`);
        await this.connectToRedis();
        return null;
      }
      if (error.message.includes('NOGROUP')) {
        console.log(`${error.message} ...CREATING GROUP`);
        await this.createConsumerGroup(streamName, group);
        return null;
      }
      console.error(
        `Failed to xReadGroup from Redis stream: ${error.message}`,
        error,
      );

      return null;
    }
// --snip--

As with other Redis calls, we check for a closed connection - ClientClosedError and try to reconnect in this case.
Then We check for an error with a message including NOGROUP
The full error might look similar to this - NOGROUP No such key 'example-stream' or consumer group 'example-group' in XREADGROUP with GROUP option, but if we see NOGROUP, we know that this stream does not have the group we want our consumer to be part of.
So we need to create this group.:
await this.createConsumerGroup(streamName, group);

// redis.service.ts

  private async createConsumerGroup(streamName: string, group: string) {
    try {
      await this.redis.xGroupCreate(
        streamName,
        group,
        '0', // use 0 to create group from the beginning of the stream, use '$' to create group from the end of the stream
        {
          MKSTREAM: true,
        },
      );
    } catch (error) {
      if (error.message.includes('BUSYGROUP')) {
        // Consumer group already exists
        return;
      }
      if (error instanceof ClientClosedError) {
        console.log(`${error.message} ...RECONNECTING`);
        await this.connectToRedis();
        return null;
      }
      console.error(`Failed to xGroupCreate: ${error.message}`);
      return null;
    }
  }

We are calling XGROUP CREATE command providing streamName, the group name we want to create.
We can also choose the start position where we want to start consuming messages. Here it's set to 0, which means the latest message in the stream. It could be any valid ID or a special character $ - this would start consuming messages added only after group creation.

We also set MKSTREAM option to true, so in case we don't have this stream yet for some reason, e.g., no messages have been added, Redis gets restarted/switched, or in case of bad memory management, the stream has evicted, we will create a new stream with this name (as well as add a consumer group).

We added error handling for BUSYGROUP - meaning that this group (and stream) exists after all ( e.g., in case it has been just created by another group consumer)
The rest of the error handing stays the same as for other Redis calls.

Finally, jumping back to readConsumerGroup, after we have fetched messages and done our error handling, we extract the first stream's messages (since we added only one key and id object) and return them.

// redis.service.ts

// --snip--
    const messages = response?.[0]?.messages; // returning first stream (since only 1 stream used)
    return messages || null;
// --snip--

Message acknowledgement

Consumer groups is a great way to distribute data on stream to different consumers only once per group. But how to guarantee that a message sent out by the Redis server was actually received and processed by the consumer it was sent to? The answer is "acknowledgment".

When a message is sent to a consumer, as a side effect, Redis puts the message ID and that consumer in the Pending Entries List (PEL) of that stream consumer group.

The consumer must send an acknowledgment (also known as an "ack") to the Redis server to signal that it has successfully processed the message. This acknowledgment is sent using the XACK command, which takes the name of the consumer group, the stream's name, and the IDs of the messages that were processed.

Once the Redis server receives an acknowledgment from a consumer, it updates its internal tracking data to mark the message as processed by that consumer. The server also removes the message from the PEL for that consumer.

If a consumer fails to send an acknowledgment within a configurable period, the message is considered unacknowledged. The Redis server will then deliver the message again to another consumer in the same consumer group. This ensures that messages are not lost if a consumer fails or crashes before it can process a message.

Implementing acknowledgement

Let's handle acknowledgement using the xAck method:

// redis.service.ts

  public async acknowledgeMessages({
    streamName,
    group,
    messageIds,
  }: AcknowledgeMessageParams) {
    try {
      await this.redis.xAck(streamName, group, messageIds);
    } catch (error) {
      if (error instanceof ClientClosedError) {
        console.log(`${error.message} ...RECONNECTING`);
        await this.connectToRedis();
        return null;
      }
      console.error(`Failed to xAck from Redis stream: ${error.message}`);
      return null;
    }
  }

For this we have create yet another interface - AcknowledgeMessageParams

// interfaces.ts

export interface AcknowledgeMessageParams {
  /** Name of stream to acknowledge message in */
  streamName: string;
  /** Name of consumer group */
  group: string;
  /** ID of messages to acknowledge */
  messageIds: string[];
}

XACK takes single or multiple message IDs, so we can batch multiple acks in one call.

The rest is straightforward - we send the command and handle errors as we do for other Redis methods.

(Auto)Claiming unacknowledged messages

XAUTOCLAIM is a command in Redis that automates the process of claiming unacknowledged messages from a consumer group. It is similar to the XPENDING and XCLAIM commands, but it simplifies the process by automatically allowing Redis to claim unacknowledged messages on behalf of a consumer.

A consumer sends a XAUTOCLAIM command with the consumer group's name, the consumer's name, and the name of the stream to consume from. Redis returns unacknowledged messages for the consumer to claim, which can be processed and acknowledged like a message received from XREADGROUP. Another consumer can claim unacknowledged messages if the acknowledgment window expires. On subsequent XAUTOCLAIM commands, Redis only returns unacknowledged messages that have not been previously returned to the consumer and have yet reached the maximum delivery attempts.

Implementing `xAutoClaim`

Here the code is almost the same as we had for xReadGroup:

// redis.service.ts

  public async autoClaimMessage({
    streamName,
    group,
    consumer,
    minIdleTimeMs,
    count,
  }: AutoclaimMessageParams) {
    let response: RedsXAutoClaimResponse = null;
    try {
      response = await this.redis.xAutoClaim(
        streamName,
        group,
        consumer,
        minIdleTimeMs,
        '0-0', // use 0-0 to claim all messages. In case of multiple consumers, this will be used to claim messages from other consumers
        {
          COUNT: count,
        },
      );
    } catch (error) {
      if (error instanceof ClientClosedError) {
        console.log(`${error.message} ...RECONNECTING`);
        await this.connectToRedis();
        return null;
      }
      console.error(`Failed to xAutoClaim from Redis stream: ${error.message}`);
      return null;
    }
    return response?.messages || null;
  }

We have created a params interface AutoclaimMessageParams

// interfaces.ts

export interface AutoclaimMessageParams {
  streamName: string;
  group: string;
  consumer: string;
  minIdleTimeMs: number;
  count: number;
}

Here the distinct params is minIdleTimeMs - this is a minimum idle time in ms for message to be eligible for auto-claim.

Utilizing generators to consume messages

We have the functionality to consume, ack, and even re-claim messages. Let's add all of that together and abstract all of that logic inside a generator so that the StreamHandlerService clients don't need to worry about all of these mechanisms.

// stream-handler.service

  public async *getConsumerMessageGenerator({
    streamName,
    group,
    consumer,
    count,
    autoClaimMinIdleTimeMs,
    autoAck = true,
  }: ReadConsumerGroupParams): AsyncRedisStreamGenerator {
    let fetchNewMessages = true; // Toggle for switching between fetching new messages and auto claiming messages
    while (this.isAlive) {
      let response: RedisStreamMessage[];
      if (fetchNewMessages) {
        response = await this.redisService.readConsumerGroup({
          streamName,
          group,
          consumer,
          blockMs: 0, // 0 = infinite blocking until at least one message is fetched, or timeout happens
          count,
        });
      } else {
        // Try to auto claim messages that are idle for a certain amount of time
        response = await this.redisService.autoClaimMessage({
          streamName,
          group,
          consumer,
          count,
          minIdleTimeMs:
            autoClaimMinIdleTimeMs || StreamHandlerService.DEFAULT_IDLE_TIME_MS,
        });
      }

      // Acknowledge messages if autoAck is enabled
      if (autoAck && response?.length > 0) {
        await this.redisService.acknowledgeMessages({
          streamName,
          group,
          messageIds: response.map((m) => m.id),
        });
      }

      // Toggle between fetching new messages and auto claiming messages
      fetchNewMessages = !fetchNewMessages;

      // If no messages returned, continue to next iteration without yielding
      if (!response || response.length === 0) {
        continue;
      }
      for (const message of response) {
        yield message;
      }
    }
  }

Here's the new params interface ReadConsumerGroupParams

// interfaces.ts

export interface ReadConsumerGroupParams {
  streamName: string;
  group: string;
  consumer: string;
  count: number;
  autoClaimMinIdleTimeMs?: number;
  autoAck?: boolean;
}

We will expect streamName, consumer group name, and a unique consumer name. A count of how many messages to try to fetch in one call, as well as
autoClaimMinIdleTimeMs - minimum idle time in ms for the message to be eligible for auto-claim, re-claim from another consumer, if not acknowledged.

And finally, autoAck - Should messages be acknowledged automatically after being read. If set to false, the client must manually acknowledge messages using acknowledgeMessage

We will try to alternate between fetching new messages and auto claiming forgotten ones. For this a variable fetchNewMessages is introduced.

// --snip--
    let fetchNewMessages = true;
    while (this.isAlive) {
// --snip--

As for our fan-out generator, we will use a while loop that will become invalid when the module is destroyed.

Then we alternate between new messages and auto-claim and assigning the result to a response variable.

// --snip--
      let response: RedisStreamMessage[];
      if (fetchNewMessages) {
        response = await this.redisService.readConsumerGroup({
          streamName,
          group,
          consumer,
          blockMs: 0, // 0 = infinite blocking until at least one message is fetched, or timeout happens
          count,
        });
      } else {
        // Try to auto claim messages that are idle for a certain amount of time
        response = await this.redisService.autoClaimMessage({
          streamName,
          group,
          consumer,
          count,
          minIdleTimeMs:
            autoClaimMinIdleTimeMs || StreamHandlerService.DEFAULT_IDLE_TIME_MS,
        });
      }
// --snip--

Next, we'll handle the auto-ack functionality. This will allow us to remember to ack message after processing.
In production applications, it might or might not make sense. For example, if you need to ensure that a message has been received and successfully processed, you will need to do ack manually after processing.

// --snip--
      if (autoAck && response?.length > 0) {
        await this.redisService.acknowledgeMessages({
          streamName,
          group,
          messageIds: response.map((m) => m.id),
        });
// --snip--

Then we want to toggle the "fetching mode"

// --snip--
      fetchNewMessages = !fetchNewMessages;
// --snip--

And finally, only if there are any messages, we want to yield them:

// --snip--
      if (!response || response.length === 0) {
        continue;
      }
      for (const message of response) {
        yield message;
      }
    }

Consuming messages

The hard part is over. Now we can happily read the Redis stream via consumer groups.
Let's do that from our AppService:

// app.service.ts

  public async consumeMessageFromGroup(
    group: string,
    consumer: string,
    count: number,
  ) {
    const generator = this.streamService.getConsumerMessageGenerator({
      streamName: EXAMPLE_STREAM_NAME,
      group,
      consumer,
      count,
    });
    const messages: Record<string, string>[] = [];
    let counter = 0;
    for await (const messageObj of generator) {
      messages.push(this.parseMessage(messageObj.message));
      counter++;
      if (counter >= count) {
        break;
      }
    }
    return {
      group,
      consumer,
      messages,
    };
  }

The code is almost the same as for multiple new messages reads in a fan-out mode that we create in part 2. The only addition is group and consumer.

Let's also add an simple endpoint from which we can consume messages and specify the group, consumer name and count:

// app.controller

  @Get('consume/:group/:consumer/:count')
  consumeMessages(
    @Param('group') group: string,
    @Param('consumer') consumer: string,
    @Param('count') count: number,
  ) {
    return this.appService.consumeMessageFromGroup(group, consumer, count);
  }

In this case, we won't create multiple instances of our service, but it would work the same.
Here is an example:

If we open our service with http://localhost:8081/consume/example-group/example-consumer/3, we get messages from the stream beginning for a group named example-group and for a consumer named example-consumer.

If we try a different consumer, e.g., example-consumer-2 we will get new messages that are not pending and have not been ack by any of the consumers yet.

However, if we try to fetch messages from a different group, they will be fetched from the very beginning, no matter if another group or even the same consumer in another group has acknowledged them already.

Here is an example with example-consumer consuming messages from the same stream, but on example-group-2

Notice that these messages are the same as first fetched by example-group. That is because they were consumed and acknowledged by that group only. Different groups do not share the PEL.

That's all, floks!

Over this 3-part article series, we have successfully built a NestJS application that connects to a Redis server and utilizes the Redis stream data structure.
In part 1, we set up the Redis server, installed the Redis client library, and established a connection between the NestJS application and Redis.

In part 2, we explored the basics of Redis streams and learned how to use the Redis client to add messages to the stream. We also implemented fan-out mode, which allows multiple clients to consume messages from a single stream.

In part 3, we took our implementation to the next level by using Redis consumer groups. We learned how to create a consumer group and use it to consume messages in a more scalable and reliable way.

We also implemented some error handling to ensure our application can recover from errors and continue processing messages without interruption.

Overall, this article series has provided a comprehensive guide to building a NestJS application that utilizes Redis streams and consumer groups. By following along and experimenting with the code samples, you should now understand how to implement real-time systems that can easily handle large amounts of data, thanks to the power and flexibility of Redis and NestJS.

Redis Streams + NestJS: Part 2 | Reading from Stream

Krisjanis Kallings — Wed, 22 Feb 2023 16:58:57 +0000

Intro

This is part 2 of a 3-part series, where we will explore how to use Redis streams with NestJS.

It is structured in 3 parts:

Part 1 - Setting up the NestJS application and connecting to Redis
Part 2 - Populating Redis streams and reading from in fan-out mode
Part 3 - Using consumer groups to handle one stream from multiple actors in a way that one message is sent to and processed only by a single actor (consumer)

By the end of this series, you will have the knowledge and tools necessary to create your own NestJS app that utilizes Redis streams to handle real-time data.

Full code is available on the github

What we have

In part 1, we have created a working NestJS application that can connect to the Redis server. We have also created an endpoint /ping-redis for our application that will call a ping command on the Redis server and return its response.
For our development convenience, we run our application and Redis server in a docker. We are using Docker Compose to set up our docker-copose.yml file for simplicity.

Now we are ready to focus on the main topic - using Redis streams.

What are Redis streams

Before we start with our implementation, it's essential to understand what Redis streams are and clear some misconceptions that might have arisen about what Redis streams are not.

Data structure

At its core, Redis streams are a data structure that holds data - messages, in order by the message IDs.
By default, this ID is a timestamp on when this data was added to the stream (followed by sequence number, in case multiple entries happened at the same timestamp).

Not a list

This all sounds very similar to a list. However, the difference here is that while a list can be popped, streams can't. This is because reading the stream does not mutate it. So when calling LPOP or RPOP on a list, the element returned is removed. In streams, XREAD leaves the item in the stream and can be read multiple times.

Not a socket - there is no "subscribe"

Just by the name "stream," it's easy to think that it's a continuous "stream" of data - something that you would connect to and send you new data once it's appended.
It's a stream in the sense that the data represents some direction - older data with smaller IDs and newer data with larger IDs.
But to access it, we will need to read from it manually, just like with other more superficial data structures, like sets or lists. Once a request has been executed and data is sent from the Redis to the application, the command has been completed, and no new data will be received unless it is called again.
So it would also be wrong to think about streams in the form of Pub-Sub. With Pub-sub, once you subscribe to channels, the connection is blocked, and all new messages published will be delivered. No need to re-subscribe.

Basic functionality

We mostly add to stream (XADD) and read from it (XREAD, XRANGE XREADGROUP). There is also functionality for deletion (XDEL), but in most cases, we think of the stream as append-only.

Managing memory

To avoid using up all of the space and evicting data randomly, the length of streams is managed by trimming them. It can be done manually via XTRIM, or we can move the responsibility to the Redis by using capped streams - adding MAXLEN option when using XADD.

Adding to streams

Finally, let's get back into implementing streams in our NestJS application.
Let's create a method to add a message to the stream:

// redis.service.ts

// --snip--
  public async addToStream({
    fieldsToStore,
    streamName,
  }: AddToStreamParams): Promise<string> {
    // Converting object to record to store in redis
    const messageObject = Object.entries(fieldsToStore).reduce(
      (acc, [key, value]) => {
        if (typeof value === 'undefined') {
          return acc;
        }
        acc[key] = typeof value === 'string' ? value : JSON.stringify(value);

        return acc;
      },
      {} as Record<string, string>,
    );

    // Adding to stream with trimming - approximately max 100 messages
    return this.redis.xAdd(streamName, '*', messageObject, {
      TRIM: {
        strategy: 'MAXLEN',
        strategyModifier: '~',
        threshold: 100,
      },
    });
  }

As you can see, we have defined the interface for the parameters of this method:

// interfaces.ts

export interface AddToStreamParams {
  fieldsToStore: Record<string, any>;
  streamName: string;
}

We will take in the name of the stream we want to add data to and specify data as POJO named fieldsToStore

Next, we have a transformation code that transforms our object of any depth into key-value pairs of string type. Later, when reading data, we will parse this back to our original POJO:

// redis.service.ts

// --snip--
    const messageObject = Object.entries(fieldsToStore).reduce(
      (acc, [key, value]) => {
        if (typeof value === 'undefined') {
          return acc;
        }
        acc[key] = typeof value === 'string' ? value : JSON.stringify(value);

        return acc;
      },
      {} as Record<string, string>,
    );
// --snip--

Then we execute a call to Redis service with the following parameters:

// redis.service.ts

// --snip--
return this.redis.xAdd(streamName, '*', messageObject, {
      TRIM: {
        strategy: 'MAXLEN',
        strategyModifier: '~',
        threshold: 100,
      },
    });
// --snip--

streamName - the name of our stream that we want to append to
* - asterisk specifies Redis to auto-generate a unique ID (based on timestamp). Here we could set our own ID if we want.
messageObject - key-value pairs that contain the data we want to store
Options:
- TRIM - as mentioned before, we don't want to handle trimming manually, so we specify that we wish Redis to trim the stream by itself. This includes additional options:
  - strategy MAXLEN - here, we specify that we want to trim when the stream has reached the maximum length of a threshold. Other option would be MINID, where the threshold would specify the smallest ID to keep. The rest are marked for eviction.
  - strategyModifier ~ - this tells Redis that we don't want to be strict in size, and Redis can go above the threshold for a little while, but the trim will eventually happen. This optimizes Redis to execute eviction at the most convenient time, not immediately.
  - threshold 100 - that is the max length of the stream we want. In the case of MINID, it would be the smallest ID to keep. And since we have strategy modifier ~, this is an approximate value.

So we would read it like this: Trim stream if the length (MAXLEN) of the stream has reached approximately (~) 100.

Stream handler

Since we will do more stuff with our stream data and we want to keep the basic functionality of Redis server calls on RedisService, we will create a new service called StreamHandlerService for additional handling of the streams.
For now, we will just add "pass-through" method addToStream that will call RedisService addToStream method.
To generate it, we'll call: nest g service redis/stream-handler

// stream-handler.service.ts

import { Injectable } from '@nestjs/common';
import { RedisService } from './redis.service';

@Injectable()
export class StreamHandlerService {

constructor(private readonly redisService: RedisService) {}

  public addToStream(fieldsToStore: Record<string, any>, streamName: string) {
    return this.redisService.addToStream({ fieldsToStore, streamName });
  }
}

And since we don't want to expose our RedisService to other modules and will want them to interact with StreamHandlerService, we will move the ping call functionality to it and remove export from module.

// stream-handler.service.ts

// --snip--
  public ping() {
    return this.redisService.ping();
  }
// --snip--

To use it in other modules, we need to export it:

// redis.module.ts

import { Module } from '@nestjs/common';
import { redisClientFactory } from './redis-client.factory';
import { RedisService } from './redis.service';
import { StreamHandlerService } from './stream-handler.service';

@Module({
  providers: [redisClientFactory, RedisService, StreamHandlerService],
  exports: [StreamHandlerService], // Removed RedisService from exports
})
export class RedisModule {}

And in AppService, we are going to call ping from StreamHandlerService instead:

// app.service.ts

// --snip--
  constructor(private readonly streamService: StreamHandlerService) {} // changed from RedisService to StreamHandlerService

  redisPing() {
    return this.streamService.ping(); // changed here as well
  }
// --snip--

In theory, other modules should not care whether it is Redis, RabbitMQ, Kafka, or even something else we are calling. . It should only know that it is a stream, and we get the specific data by calling methods made public on exported services.
This way, we can easily change our implementation of streams without needing to change other parts of code - they become de-coupled.
Since this is a project about Redis and Redis streams, we have called this module RedisModule, but expanding on the previous taught, IRL it would make more sense to call it StreamModule or MessagingModule, or something else, depending on what these streams are going to be used for.

Populating Redis with messages

Let's use our addToStream method to populate our Redis.
We could go the same route as with ping and expose it to our API endpoint, but this time, since we want to "simulate" data flow, we'll call this method in an interval so that we have an ongoing flow of data in our stream.
Let's add it to our AppService:

// app.service.ts

// --snip--

@Injectable()
export class AppService implements OnModuleInit, OnModuleDestroy {
  private interval: NodeJS.Timeout = null;

// --snip--

  private populateStream() {
    this.interval = setInterval(() => {
      this.streamService.addToStream(
        {
          hello: 'world',
          date: new Date(),
          nestedObj: { num: Date.now() % 100 },
        },
        EXAMPLE_STREAM_NAME,
      );
    }, 1000);
  }

  async onModuleInit() {
    this.populateStream();
  }

  onModuleDestroy() {
    clearInterval(this.interval);
  }
}

As you can see, we have created a populateStream method that will add some dummy data into our stream.
EXAMPLE_STREAM_NAME is just a constant of our stream name we have defined in constants.ts file and imported it here:

// constants.ts
export const EXAMPLE_STREAM_NAME = 'example-stream';

We created an interval that we call on onModuleInit lifestyle event, and we clear it in our onModuleDestroy lifecycle event so that we don't leave dangling references.

This will add a message to our Redis stream on every 1000 ms.
To verify that, let's now create a way to read from stream.

Reading from stream

To read from Redis streams, we can use either XREAD or XRANGE (and REVXRANGE) commands. Since the goal of this project is to read new messages, both commands could be used to achieve this goal. We are going to use XREAD in this article.
Let's add a method that wraps XREAD command call and does some extra error handling for us:

// redis.service.ts

// --snip--

  public async readStream({
    streamName,
    blockMs,
    count,
    lastMessageId,
  }: ReadStreamParams): Promise<RedisStreamMessage[] | null> {
    try {
      const response = await this.redis.xRead(
        commandOptions({ isolated: true }), // uses new connection from pool not to block other redis calls
        [
          {
            key: streamName,
            id: lastMessageId,
          },
        ],
        { BLOCK: blockMs, COUNT: count },
      );

      const { messages } = response?.[0]; // returning first stream (since only 1 stream used)

      return messages || null;
    } catch (error) {
      if (error instanceof ClientClosedError) {
        console.log(`${error.message} ...RECONNECTING`);
        await this.connectToRedis();
        return null;
      }
      console.error(
        `Failed to xRead from Redis Stream: ${error.message}`,
        error,
      );
      return null;
    }
  }

Let's break down this code:

First, notice that we have added a parameters interface ReadStreamParams:

// interfaces.ts

export interface ReadStreamParams {
  streamName: string;
  blockMs: number;
  lastMessageId: string;
}
// --snip--

as well as we have defined our response message RedisStreamMessage. Since node-redis is scarce with its type exports, we need to extract it on our own:



// redis-client.types

// --snip--

export type RedisStreamMessage = Awaited<
  ReturnType<RedisClient['xRead']>
>[number]['messages'][number];

// --snip--

In the try block, we have our command call:


// --snip--
      const response = await this.redis.xRead(
        commandOptions({ isolated: true }), // uses new connection from pool not to block other redis calls
        [
          {
            key: streamName,
            id: lastMessageId,
          },
        ],
        { BLOCK: blockMs, COUNT: count },
      );

      const { messages } = response?.[0]; // returning first stream (since only 1 stream used)

      return messages || null;
    } catch (error) {
// --snip--

We first use the call command option to run it in isolation commandOptions({ isolated: true }).
This is needed because we are going to BLOCK the connection. This means that once this command call takes a connection to the Redis server if the results won't be on the Redis at the time the call arrives, it will wait for the specified amount of time (BLOCK parameter) and, if during that period, data arrives that matches the request, it will be returned immediately, or it will wait till the end of the specified time and return nothing.
During this time, the connection is used-up, and no other commands can be called. Essentially, our command BLOCKS the connection.

If we would like to create multiple calls, we would quickly run into delays and performance problems. For this, we can use isolated execution. This means that node-redis will create multiple connections using Generic resource pool under the hood, so we'll have multiple connections to use in parallel.

Next, we say which streams we want to read and starting from what ID. We need to pass it in an array, since we could call multiple streams in a single call. We'll handling of multiple streams in one request out of the scope of this article:

        [
          {
            key: streamName,
            id: lastMessageId,
          },
        ],

Next we pass optional parameters:

        { BLOCK: blockMs, COUNT: count },

BLOCK - is blocking time in milliseconds. *If 0 is specified, it will block indefinitely until at least 1 message can be returned. *
COUNT - this is a maximum count we will get returned. If there are fewer messages available on Redis server, it will not wait for the count to match but return that many messages how many there are.

Next we unravel the data:

      const { messages } = response?.[0];

      return messages || null;

Since we called only 1 stream, the response array will always be of length 1.
All messages returned in an messages object.
And finally, we return either the stream messages array or null.

Here is also the place we can handle some initial errors:

// --- snip---
    } catch (error) {
      if (error instanceof ClientClosedError) {
        console.log(`${error.message} ...RECONNECTING`);
        await this.connectToRedis();
        return null;
      }
      console.error(
        `Failed to xRead from Redis Stream: ${error.message}`,
        error,
      );
      return null;
    }

In case of an error, we will return null value.
If we have ClientClosedError, there is no use tying to fetch anything further, so we try to reconnect to the Redis.

// --- snip---
  private async connectToRedis() {
    try {
      // Try to reconnect only if the connection socket is closed. Else let it be handled by reconnect strategy.
      if (!this.redis.isOpen) {
        await this.redis.connect();
      }
    } catch (error) {
      console.error(
        `[${error.name}] ${error.message}`,
        error,
      );
    }
  }

We check for isOpen to see if the socket is open. If it is, we will trust reconnect strategy to do its job and reconnect to the client. If it's not, then we are trying to do it ourselves.

And that's how we can fetch a single message from a stream. Let's expand on this and add the ability to read multiple messages.

Exposing read to our API

Exposing the Redis stream to our API is similar to what we did with ping.
Let's add another method to our StreamHandlerService:

//stream-handler.service.ts

// --- snip --
  public readFromStream(streamName, count) {
    return this.redisService.readStream({
      streamName,
      blockMs: 0, // 0 = infinite blocking until at least one message is fetched, or timeout happens
      count, // max how many messages to fetch at a time
      lastMessageId: '$',
    });
  }

Simple method where we pass streamName of the stream we want to read and count of messages we would like to receive. As mentioned before, COUNT actually is "max count" since, if there are less or no messages on stream, Redis will return from 1 to count number of messages.
We are using special ID $ - this represents "anything that comes after the XREAD request.

Now let's handle reading a single message:

// app.service.ts

// --snip--
  public getSingleNewMessage() {
    return this.streamService.readFromStream(EXAMPLE_STREAM_NAME, 1);
  }
// --snip--

Again, using EXAMPLE_STREAM_NAME - same stream that we populate, and setting count to 1.

And finally, let's add a new endpoint to our controller:

// app.controller.ts

// --snip--
  @Get('message')
  getMessage() {
    return this.appService.getSingleNewMessage();
  }
// --snip--

Now calling the endpoint we should see single message generated by our populateStream method:

Handling results as a stream

Getting a single message is relatively straight forward.
But how do we get multiple messages? What if we want to get a specific number of messages, e.g., precisely 3?
At first, it might seem simple - just set a count to the number of messages you want. But remember - COUNT does not guarantee that we will get the exact number of messages. It's more of a "max count". So you might be tempted to call XREAD multiple times. But, since we are passing $ as our last message ID, we might miss some messages that come in between our calls.
That means that we need to remember our last id.

And what if we want to read from a stream continuously and indefinitely?

This all adds additional complexity that we need to handle.

Generators

Generators are a valuable tool to have in our toolbox. Here they come in handy when we want to fetch the data from some external resource (Redis) lazily.
I will not focus on the "what and how" of generators.
If you haven't used them before, I encourage you to check JavaScript Generators reference and Dr. Axel Rauschmayer's book's "JavaScript for impatient programmers" on asynchronous generators.

The main feature we can do is create an infinite loop that we can stop and restart whenever we want and need.

Let's add a new method to our StreamHandlerService:

// stream-handler.service.ts

export class StreamHandlerService implements OnModuleDestroy {

  private isAlive = true;

  onModuleDestroy() {
    this.isAlive = false;
  }

// --snip--
  public async *getStreamMessageGenerator(
    streamName: string,
    count: number,
  ): AsyncRedisStreamGenerator {
    // Start with latest data
    let lastMessageId = '$';
    while (this.isAlive) {
      const response = await this.redisService.readStream({
        streamName,
        blockMs: 0, // 0 = infinite blocking until at least one message is fetched, or timeout happens
        count, // max how many messages to fetch at a time
        lastMessageId,
      });

      // If no messages returned, continue to next iteration without yielding
      if (!response || response.length === 0) {
        continue;
      }
      // Update last message id to be the last message returned from redis
      lastMessageId = response[response.length - 1].id;
      for (const message of response) {
        yield message;
      }
    }
  }
// --snip--

Let's break down what we have written:

First, you will notice that we have added a new boolean isAlive that we will initially set to true, and only during the destruction of the module we will set it to false. This will serve as a while-true loop that we can exit from when the application ends.

Also, for our convenience, we have created a new type AsyncRedisStreamGenerator

// redis-client.type.ts

// --snip--
export type AsyncRedisStreamGenerator = AsyncGenerator<
  RedisStreamMessage,
  void,
  unknown
>;
// --snip--

This is an async generator that will yield or produce RedisStreamMessage values, will return nothing (void) when it returns and will accept unknown messages, since we won't pass anything to next method of our generator.

We are going to call readStream method in an infinite loop:

    let lastMessageId = '$';
    while (this.isAlive) {
      const response = await this.redisService.readStream({
        streamName,
        blockMs: 0, // 0 = infinite blocking until at least one message is fetched, or timeout happens
        count, // max how many messages to fetch at a time
        lastMessageId,
      });
// -- snip --
    }

As in our previous example, we will start to read data only when the generator is created and calls the stream by using $ as lastMessageId.
The rest is also the same as in the previous example, except we are calling it in a loop.

Next, we will handle the case where we get an empty response. We will simply go into the next iteration without yielding any results

   // If no messages returned, continue to next iteration without yielding
      if (!response || response.length === 0) {
        continue;
      }

If we have some responses, we are going to handle them:

      // Update last message id to be the last message returned from redis
      lastMessageId = response[response.length - 1].id;
      for (const message of response) {
        yield message;
      }

To avoid gaps in messages and read all messages from the time we started this generator, we will update lastMessageId variable with the last message we received.

Since we have count, we will get an array of sizes ranging from 1 to count with messages. However, this generator produces only a single message at a time. So we will iterate through the array and yield only one message at once.

That is for the generator part. Let's see how we can use it to solve our previous questions:

Reading multiple results

Now let's use our generator to generate multiple results by creating a new method in AppService

// app.service.ts

// --snip--
  public async getMultipleNewMessages(count: number) {
    const generator = this.streamService.getStreamMessageGenerator(
      EXAMPLE_STREAM_NAME,
      count,
    );
    const messages: Record<string, string>[] = [];
    let counter = 0;
    for await (const messageObj of generator) {
      messages.push(this.parseMessage(messageObj.message));
      counter++;
      if (counter >= count) {
        break;
      }
    }
    return messages;
  }
// --snip--

First, we create a generator:

    const generator = this.streamService.getStreamMessageGenerator(
      EXAMPLE_STREAM_NAME,
      count,
    );

We are passing it our count for "optimistic" fetch. We will try to get everything in a single Redis call. But if that does not happen, we will call Redis as often as needed.
We create an array to collect our messages:

    const messages: Record<string, string>[] = [];

and define a counter where we will count how many we have received.

One convenient feature of generators is that they are iterable as they implement @@iterator method. So we can call for..of on them. In this case, since the generator is async it's for await .. of

    for await (const messageObj of generator) {
      messages.push(this.parseMessage(messageObj.message));
      counter++;
      if (counter >= count) {
        break;
      }
    }
    return messages;

Once the count is reached, we simply break out of the loop and then return our collected messages.

Since Redis stores all values as strings, we also have created a generic parser method:

//app.service.ts

// --snip--
  private parseMessage(message: Record<string, string>) {
    return Object.entries(message).reduce((acc, [key, value]) => {
      try{
      acc[key] = JSON.parse(value);
      }catch(e){
        acc[key] =value
      }
      return acc;
    }, {});
  }
// --snip--

To check that we indeed are getting specified amount of messages, we'll create a new endpoint /messages where we will fetch 3 messages :

// app.controller.ts

// --snip--
  @Get('messages')
  getMessages() {
    return this.appService.getMultipleNewMessages(3);
  }
// --snip--

Again, when we call our newly created endpoint, we indeed, have 3 messages fetched:

Continuous reading of the stream

The biggest power of a Redis stream comes from using it as a continuous data stream. To do that, we can re-use our generator:

// app.service.ts

// --snip--
  private isAlive = true;
// --snip--
  async onModuleInit() {
// --snip--
    this.continuousReadMessages();
  }
  onModuleDestroy() {
// --snip--

    this.isAlive = false;
  }
// --snip--
  private async continuousReadMessages() {
    const generator = this.streamService.getStreamMessageGenerator(
      EXAMPLE_STREAM_NAME,
      3,
    );
    for await (const messageObj of generator) {
      console.log(
        `Got message with ID: ${messageObj.id}`,
        JSON.stringify(this.parseMessage(messageObj.message), undefined, 2),
      );
      if (!this.isAlive) {
        break;
      }
    }
  }
// --snip--

Same as with multiple message fetch, we start by creating a generator:

    const generator = this.streamService.getStreamMessageGenerator(
      EXAMPLE_STREAM_NAME,
      10,
    );

We give it an arbitrary number for count since we want to try to fetch multiple responses in one Redis call, if possible.

And then we just do whatever we need to do with the data in the loop. So here it's simply logging the message.

And breaking the loop once the module is destroyed, we don't have any dangling references and can stop our loop.

    for await (const messageObj of generator) {
      console.log(
        `Got message with ID: ${messageObj.id}`,
        JSON.stringify(this.parseMessage(messageObj.message), undefined, 2),
      );
      if (!this.isAlive) {
        break;
      }
    }

And we call this method once on our module init, the same as we did with populating stream:


  async onModuleInit() {
    this.populateStream();
    this.continuousReadMessages();
  }

And in our console, we should see that we are fetching the messages continuously

...
app      | Got message with ID: 1675377159049-0 {
app      |   "hello": "world",
app      |   "date": "2023-02-02T22:32:39.048Z",
app      |   "nestedObj": {
app      |     "num": 48
app      |   }
app      | }
app      | Got message with ID: 1675377160050-0 {
app      |   "hello": "world",
app      |   "date": "2023-02-02T22:32:40.050Z",
app      |   "nestedObj": {
app      |     "num": 50
app      |   }
app      | }
...

Refactor for a single message

Since we are using the generator to handle stream messages, let's refactor our single message fetch example.

// app.service.ts

// --snip--
  public async getSingleNewMessage() {
    const generator = this.streamService.getStreamMessageGenerator(
      EXAMPLE_STREAM_NAME,
      1,
    );
    const messageObj = await generator.next();
    if (!messageObj.done && messageObj.value) {
      return this.parseMessage(messageObj.value.message);
    }
  }
// --snip--

Same as for all examples, we are creating a generator on our EXAMPLE_STREAM_NAME stream, this time with count 1.

    const generator = this.streamService.getStreamMessageGenerator(
      EXAMPLE_STREAM_NAME,
      1,
    );

Since we need only a single message, we do not need to use a loop. Instead, we call next method manually:

    const messageObj = await generator.next();

and then handle the response - if the generator has not returned/ended - it's not done and we have a message t return, we parse message and return it to the client:

    if (!messageObj.done && messageObj.value) {
      return this.parseMessage(messageObj.value.message);
    }

Everything else stays the same for the client, and the same result is returned.

This is the end of part 2 of the 3-part series.
We built on what we created in part 1. As a result, you should have an application that can fetch single or multiple responses and continuously fetch new results from the Redis stream.

In part 3, we are going to look at how we can use consumer groups to divide messages to multiple consumers so that each one gets a unique set of messages (not like we have now, where if we connected more clients, all clients would get getting all of the messages).

Redis Streams + NestJS: Part 1 | Setup

Krisjanis Kallings — Wed, 22 Feb 2023 16:58:39 +0000

Intro

This is part 1 of a 3-part series, where we will explore how to use Redis streams with NestJS.

It is structured in 3 parts:

Part 1 - Setting up the NestJS application and connecting to Redis
Part 2 -Populating Redis streams and reading from in fan-out mode
Part 3 - Using consumer groups to handle one stream from multiple actors in a way that one message is sent to and processed only by a single actor (consumer)

By the end of this series, you will have the knowledge and tools necessary to create your own NestJS app that utilizes Redis streams to handle real-time data.

Full code is available on the github

Assumed prior knowledge

Before diving into this post, it is assumed that you have a basic understanding of NestJS and Redis.
Familiarity with JavaScript generators is also recommended, as they will be used in the examples to listen to the Redis stream continuously.
We will also use Docker and Docker Compose to create and run our Redis server and application.
NestJS is a typescript framework, so you should be familiar with it too.
To familiarize yourself with these things, I recommend:

NestJS docs
TypeScript
Redis:
- Redis streams intro
- The tutorial
Generators:
- JavaScript Generators reference
- Dr. Axel Rauschmayer's book's "JavaScript for impatient programmers" sections on synchronous and asynchronous generators (available free online)
Docker:
- Docker
- Docker Compose

Side note

It's also important to note that while this post explicitly covers the integration of Redis streams with NestJS, the concepts and techniques discussed here can also be applied to other frameworks and languages. Redis streams is a versatile tool that can be used in a wide range of applications and architectures. Therefore, even if you are not using NestJS, the general concepts and approaches outlined in this post can still apply to your projects and stack.

Setup

Creating a new NestJS application

To get started, we will need to create a new NestJS application. This can be done using the NestJS CLI, which can be installed via npm by running npm i -g @nestjs/cli. Once installed, you can create a new application by running nest new redis-streams, where "redis-streams" is the name of your application.
This will create a new directory with the same name, containing the basic structure of a NestJS application.

Docker setup

To simplify things, we are going to use docker-compose.yml file to include both our app and Redis server:

# docker-compose.yml

version: '3'

services:
  app:
    container_name: app
    image: node:18
    environment:
      HTTPS_METHOD: noredirect
    ports:
      - 8081:3000
    volumes:
      - ./:/usr/src/app/
    working_dir: /usr/src/app
    command: npm run start:dev

  redis:
    container_name: redis
    image: redis:7
    restart: always

Now you can run docker-compose up -d to build and run your services. To see console output, use docker-compose logs
You should see the following message:

 LOG [NestApplication] Nest application successfully started +7ms

Connecting to Redis

Setting up `node-redis` client library

To call Redis from we are going to use the official NodeJS Redis client library node-redis

$ npm i redis

There are also other libraries, e.g. ioredis is noteworthy alternative. You can see the list of clients on the Redis website

Redis module

Finally, we can start working with Redis from our application.
First, will create a new module for Redis-related services.

$ nest g module redis

You should see this:

// redis.module.ts

import { Module } from '@nestjs/common';

@Module({
  providers: [],
  exports: [],
})
export class RedisModule {}

RedisClient factory

To use RedisClient from node-redis library we are going to create a factory provider for it:

// redis-client.factory.ts

import { FactoryProvider } from '@nestjs/common';
import { createClient } from 'redis';
import { RedisClient, REDIS_CLIENT } from './redis-client.type';

export const redisClientFactory: FactoryProvider<Promise<RedisClient>> = {
  provide: REDIS_CLIENT,
  useFactory: async () => {
    const client = createClient({ url: 'redis://redis:6379/0' });
    await client.connect();
    return client;
  },
};

Let's break it down:
We create a FactoryProvider that will call async function provided in useFactory:

// redis-client.factory.ts
// --snip--
  useFactory: async () => {
    const client = createClient({ url: 'redis://redis:6379/0' });
    await client.connect();
    return client;
  },
// --snip--

We call createClient function from redis library, and pass it the URL consisting of {protocol}://{host}:{port}/{database}, where:
- protocol= redis
- host = redis - this is specified in docker-compose.yml with container_host: redis. Usually, you would create an environment variable with your Redis instance IP and use it here.
- port = 6379 - default Redis port
- database = 0 default database
We connect to the Redis server await client.connect();
Return the created & connected client.

You may have noticed that we did not provide an instance of RedisClient type but REDIS_CLIENT, which is our injection token. Also, RedisClient is our custom type, not redis.
This is due to node-redis on v4 not exporting the RedisClient type, so we need to create our own in /redis-client.type.ts file:

export type RedisClient = ReturnType<typeof createClient>;
export const REDIS_CLIENT = Symbol('REDIS_CLIENT');

All that is left is to add this to our module:

// redis.module.ts

//  --snip--
@Module({
  providers: [redisClientFactory],
})
export class RedisModule {}

Creating RedisService

Next, let's add a RedisService that will create a layer of abstraction from RedisClient.

$ nest g service redis

In there we will inject our RedisClient

// redis.service.ts

// --snip--
@Injectable()
export class RedisService implements OnModuleDestroy {
  public constructor(
    @Inject(REDIS_CLIENT) private readonly redis: RedisClient,
  ) {}

  onModuleDestroy() {
    this.redis.quit();
  }
}

Not to leave hanging connections, we are going to close the connection to the Redis by calling this.redis.quit() on OnModuleDestroy lifecycle event.

Ping for Redis server

To check that we have successfully connected to Redis, let's add an API endpoint that calls call Redis ping

// redis.service.ts

  ping() {
    return this.redis.ping();
  }

Let's export RedisService so that we can use it in other modules:

// redis.module.ts

// --snip--
  exports: [RedisService],
// --snip--

Now we will import it into AppService and pass through our call to ping redis:

// app.service.ts

@Injectable()
export class AppService {
// --snip--
  constructor(
    private readonly redisService: RedisService,
  ) {}

  redisPing() {
    return this.redisService.ping();
  }
// --snip--

Finally, we can add a new endpoint to AppController that will execute a ping to the Redis server and send the response to the user:

// app.controller.ts

// --snip--

@Controller()
export class AppController {
  constructor(private readonly appService: AppService) {}

  @Get('redis-ping')
  redisPing() {
    return this.appService.redisPing();
  }
}

Now, with our app and Redis server running, we can open our /redis-ping endpoint http://localhost:8081/redis-ping, and you should see the response:

Congratulations! We have finished part 1 of the 3-part series and created a NestJS application with a working connection to our Redis server! In part 2 we are going to create, populate and read from Redis streams.

DEV Community: Krisjanis Kallings

Redis Streams + NestJS: Part 3 | Consumer groups

Intro

What we have

Understanding consumer groups

Using consumer groups

Reading as a consumer

Message acknowledgement

Implementing acknowledgement

(Auto)Claiming unacknowledged messages

Implementing xAutoClaim

Utilizing generators to consume messages

Consuming messages

That's all, floks!

Redis Streams + NestJS: Part 2 | Reading from Stream

Intro

What we have

What are Redis streams

Data structure

Not a list

Not a socket - there is no "subscribe"

Basic functionality

Managing memory

Adding to streams

Stream handler

Populating Redis with messages

Reading from stream

Exposing read to our API

Handling results as a stream

Generators

Reading multiple results

Continuous reading of the stream

Refactor for a single message

Redis Streams + NestJS: Part 1 | Setup

Intro

Assumed prior knowledge

Side note

Setup

Creating a new NestJS application

Docker setup

Connecting to Redis

Setting up node-redis client library

Redis module

RedisClient factory

Creating RedisService

Ping for Redis server

Implementing `xAutoClaim`

Setting up `node-redis` client library