make BLOCK time configurable

Author: darrachequesne

lib/adapter.ts

@@ -31,6 +31,12 @@ export interface RedisStreamsAdapterOptions {
    * @default 100
    */
   readCount?: number;
+  /**
+   * The number of ms before timing 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:"
@@ -61,6 +67,7 @@ export function createAdapter(
       streamName: "socket.io",
       maxLen: 10_000,
       readCount: 100,
+      blockTimeInMs: 200,
       sessionKeyPrefix: "sio:session:",
       heartbeatInterval: 5_000,
       heartbeatTimeout: 10_000,
@@ -77,7 +84,8 @@ export function createAdapter(
         redisClient,
         options.streamName,
         offset,
-        options.readCount
+        options.readCount,
+        options.blockTimeInMs
       );
 
       if (response) {

lib/util.ts

@@ -57,17 +57,15 @@ function mapResult(result) {
   };
 }
 
-// https://redis.io/docs/latest/commands/xread/#blocking-for-data
-const BLOCK_DURATION_IN_MS = 200;
-
 /**
  * @see https://redis.io/commands/xread/
  */
 export function XREAD(
   redisClient: any,
   streamName: string,
   offset: string,
-  readCount: number
+  readCount: number,
+  blockTimeInMs: number
 ) {
   if (isNodeRedisClient(redisClient)) {
     return redisClient.xRead(
@@ -79,14 +77,14 @@ export function XREAD(
       ],
       {
         COUNT: readCount,
-        BLOCK: BLOCK_DURATION_IN_MS,
+        BLOCK: blockTimeInMs,
       }
     );
   } else {
     return redisClient
       .xread(
         "BLOCK",
-        BLOCK_DURATION_IN_MS,
+        blockTimeInMs,
         "COUNT",
         readCount,
         "STREAMS",

test/util.ts

@@ -178,7 +178,9 @@ export function setup({
 
       const httpServer = createServer();
       const io = new Server(httpServer, {
-        adapter: createAdapter(redisClient),
+        adapter: createAdapter(redisClient, {
+          readCount: 1, // return as soon as possible
+        }),
         ...serverOptions,
       });
       httpServer.listen(() => {