Fix: Process multiple received streams concurrently. (#13)

tamimaj · Feb 29, 2024 · ce2f7b9 · ce2f7b9
1 parent c53b6b8
commit ce2f7b9
Show file tree

Hide file tree

Showing 2 changed files with 8 additions and 6 deletions.
diff --git a/lib/redis.client.ts b/lib/redis.client.ts
@@ -260,9 +260,10 @@ export class RedisStreamClient extends ClientProxy {
       // if BLOCK time ended, and results are null, listen again.
       if (!results) return this.listenOnStreams();
 
-      const [key, messages] = results[0];
-
-      await this.notifyHandlers(key, messages);
+      for (let result of results) {
+        let [stream, messages] = result;
+        await this.notifyHandlers(stream, messages);
+      }
 
       return this.listenOnStreams();
     } catch (error) {

diff --git a/lib/redis.server.ts b/lib/redis.server.ts
@@ -268,9 +268,10 @@ export class RedisStreamStrategy
       // if BLOCK time ended, and results are null, listen again.
       if (!results) return this.listenOnStreams();
 
-      const [key, messages] = results[0];
-
-      await this.notifyHandlers(key, messages);
+      for (let result of results) {
+        let [stream, messages] = result;
+        await this.notifyHandlers(stream, messages);
+      }
 
       return this.listenOnStreams();
     } catch (error) {