Update redisReceiver to use destroy instead of quit and set TTL for streams

Author: AbanoubGhadban

react_on_rails_pro/packages/node-renderer/tests/redisClient.test.ts

@@ -2,6 +2,15 @@ import { createClient } from 'redis';
 
 const redisClient = createClient({ url: process.env.REDIS_URL || 'redis://localhost:6379' });
 
+interface RedisStreamMessage {
+  id: string;
+  message: Record<string, string>;
+}
+interface RedisStreamResult {
+  name: string;
+  messages: RedisStreamMessage[];
+}
+
 test('Redis client connects successfully', async () => {
   await redisClient.connect();
   expect(redisClient.isOpen).toBe(true);
@@ -35,3 +44,74 @@ test('multiple connect calls', async () => {
   expect(client.isOpen).toBe(true);
   await client.quit();
 });
+
+test('write to stream and read back', async () => {
+  const client = createClient({ url: process.env.REDIS_URL || 'redis://localhost:6379' });
+  await client.connect();
+
+  const streamKey = 'test-stream';
+  await client.del(streamKey);
+  const messageId = await client.xAdd(streamKey, '*', { field1: 'value1' });
+
+  const result = (await client.xRead({ key: streamKey, id: '0-0' }, { COUNT: 1, BLOCK: 2000 })) as
+    | RedisStreamResult[]
+    | null;
+  expect(result).not.toBeNull();
+  expect(result).toBeDefined();
+
+  const [stream] = result!;
+  expect(stream).toBeDefined();
+  expect(stream?.messages.length).toBe(1);
+  const [message] = stream!.messages;
+  expect(message!.id).toBe(messageId);
+  expect(message!.message).toEqual({ field1: 'value1' });
+
+  await client.quit();
+});
+
+test('quit while reading from stream', async () => {
+  const client = createClient({ url: process.env.REDIS_URL || 'redis://localhost:6379' });
+  await client.connect();
+
+  const streamKey = 'test-stream-quit';
+
+  const readPromise = client.xRead({ key: streamKey, id: '$' }, { BLOCK: 0 });
+
+  // Wait a moment to ensure xRead is blocking
+  await new Promise((resolve) => {
+    setTimeout(resolve, 500);
+  });
+
+  client.destroy();
+
+  await expect(readPromise).rejects.toThrow();
+});
+
+it('expire sets TTL on stream', async () => {
+  const client = createClient({ url: process.env.REDIS_URL || 'redis://localhost:6379' });
+  await client.connect();
+
+  const streamKey = 'test-stream-expire';
+  await client.del(streamKey);
+  await client.xAdd(streamKey, '*', { field1: 'value1' });
+
+  const expireResult = await client.expire(streamKey, 1); // 1 second
+  expect(expireResult).toBe(1); // 1 means the key existed and TTL was set
+
+  const ttl1 = await client.ttl(streamKey);
+  expect(ttl1).toBeLessThanOrEqual(1);
+  expect(ttl1).toBeGreaterThan(0);
+
+  const existsBeforeTimeout = await client.exists(streamKey);
+  expect(existsBeforeTimeout).toBe(1); // Key should exist before timeout
+
+  // Wait for 1.1 seconds
+  await new Promise((resolve) => {
+    setTimeout(resolve, 1100);
+  });
+
+  const existsAfterTimeout = await client.exists(streamKey);
+  expect(existsAfterTimeout).toBe(0); // Key should have expired
+
+  await client.quit();
+});

react_on_rails_pro/spec/dummy/client/app/utils/redisReceiver.ts

@@ -23,7 +23,7 @@ interface RedisStreamResult {
  */
 interface RequestListener {
   getValue: (key: string) => Promise<unknown>;
-  destroy: () => Promise<void>;
+  destroy: () => void;
 }
 
 /**
@@ -51,13 +51,13 @@ export function listenToRequestData(requestId: string): RequestListener {
   /**
    * Closes the Redis connection and rejects all pending promises
    */
-  async function close(): Promise<void> {
+  function close() {
     if (isClosed) return;
     isClosed = true;
 
     // Close client - this will cause xRead to throw, which rejects pending promises
     try {
-      await redisClient.quit();
+      redisClient.destroy();
     } finally {
       isConnected = false;
     }
@@ -84,6 +84,7 @@ export function listenToRequestData(requestId: string): RequestListener {
       // And `listenToStream` runs only one promise at a time, so no fear of race condition
       if (!isConnected) {
         await redisClient.connect();
+        await redisClient.expire(streamKey, REDIS_LISTENER_TIMEOUT / 1000); // Set TTL to avoid stale streams
         isConnected = true;
       }
 
@@ -118,7 +119,7 @@ export function listenToRequestData(requestId: string): RequestListener {
 
       // If end message received, close the connection
       if (receivedEndMessage) {
-        await close();
+        close();
       }
     })();
 
@@ -174,19 +175,19 @@ export function listenToRequestData(requestId: string): RequestListener {
   /**
    * Destroys the listener, closing the connection and preventing further getValue calls
    */
-  async function destroy(): Promise<void> {
+  function destroy() {
     if (isDestroyed) return;
     isDestroyed = true;
 
     // Clear global timeout
     clearTimeout(globalTimeout);
 
-    await close();
+    close();
   }
 
   // Global timeout - destroys listener after 15 seconds
   globalTimeout = setTimeout(() => {
-    void destroy();
+    destroy();
   }, REDIS_LISTENER_TIMEOUT);
 
   return { getValue, destroy };