feat: make BLOCK timeout configurable

Author: darrachequesne

README.md

@@ -133,6 +133,7 @@ io.listen(3000);
 | `streamCount`       | The number of streams to use to scale horizontally.                                                                   | `1`            |
 | `maxLen`            | The maximum size of the stream. Almost exact trimming (~) is used.                                                    | `10_000`       |
 | `readCount`         | The number of elements to fetch per XREAD call.                                                                       | `100`          |
+| `blockTimeInMs`     | The number of ms before the XREAD call times out.                                                                     | `5_000`        |
 | `sessionKeyPrefix`  | The prefix of the key used to store the Socket.IO session, when the connection state recovery feature is enabled.     | `sio:session:` |
 | `heartbeatInterval` | The number of ms between two heartbeats.                                                                              | `5_000`        |
 | `heartbeatTimeout`  | The number of ms without heartbeat before we consider a node down.                                                    | `10_000`       |

lib/adapter.ts

@@ -52,6 +52,12 @@ export interface RedisStreamsAdapterOptions {
    * @default 100
    */
   readCount?: number;
+  /**
+   * The number of ms before the XREAD call times out.
+   * @default 200
+   * @see https://redis.io/docs/latest/commands/xread/#blocking-for-data
+   */
+  blockTimeInMs?: number;
   /**
    * The prefix of the key used to store the Socket.IO session, when the connection state recovery feature is enabled.
    * @default "sio:session:"
@@ -117,7 +123,8 @@ function startPolling(
         redisClient,
         streamName,
         offset,
-        options.readCount
+        options.readCount,
+        options.blockTimeInMs
       );
 
       if (response) {
@@ -163,6 +170,7 @@ export function createAdapter(
       streamCount: 1,
       maxLen: 10_000,
       readCount: 100,
+      blockTimeInMs: 5_000,
       sessionKeyPrefix: "sio:session:",
       heartbeatInterval: 5_000,
       heartbeatTimeout: 10_000,

lib/util.ts

@@ -72,7 +72,8 @@ export function XREAD(
   redisClient: any,
   streamName: string,
   offset: string,
-  readCount: number
+  readCount: number,
+  blockTimeInMs: number
 ) {
   if (isRedisV4Client(redisClient)) {
     return redisClient.xRead(
@@ -84,12 +85,20 @@ export function XREAD(
       ],
       {
         COUNT: readCount,
-        BLOCK: 5000,
+        BLOCK: blockTimeInMs,
       }
     );
   } else {
     return redisClient
-      .xread("BLOCK", 5000, "COUNT", readCount, "STREAMS", streamName, offset)
+      .xread(
+        "BLOCK",
+        blockTimeInMs,
+        "COUNT",
+        readCount,
+        "STREAMS",
+        streamName,
+        offset
+      )
       .then((results) => {
         if (results === null) {
           return null;