Refactor redisReceiver to eliminate global state and prevent memory leaks

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

@@ -0,0 +1,117 @@
+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);
+  await redisClient.quit();
+});
+
+test('calls connect after quit', async () => {
+  await redisClient.connect();
+  expect(redisClient.isOpen).toBe(true);
+  await redisClient.quit();
+
+  await redisClient.connect();
+  expect(redisClient.isOpen).toBe(true);
+  await redisClient.quit();
+});
+
+test('calls quit before connect is resolved', async () => {
+  const client = createClient({ url: process.env.REDIS_URL || 'redis://localhost:6379' });
+  const connectPromise = client.connect();
+  await client.quit();
+  await connectPromise;
+  expect(client.isOpen).toBe(false);
+});
+
+test('multiple connect calls', async () => {
+  const client = createClient({ url: process.env.REDIS_URL || 'redis://localhost:6379' });
+  const connectPromise1 = client.connect();
+  const connectPromise2 = client.connect();
+  await expect(connectPromise2).rejects.toThrow('Socket already opened');
+  await expect(connectPromise1).resolves.toMatchObject({});
+  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/components/ErrorComponent.tsx

@@ -7,6 +7,7 @@ const ErrorComponent = ({ error }: { error: Error }) => {
     <div>
       <h1>Error happened while rendering RSC Page</h1>
       <p>{error.message}</p>
+      <p>{error.stack}</p>
     </div>
   );
 };

react_on_rails_pro/spec/dummy/client/app/ror-auto-load-components/RSCPostsPageOverRedis.jsx

@@ -3,14 +3,14 @@ import RSCPostsPage from '../components/RSCPostsPage/Main';
 import { listenToRequestData } from '../utils/redisReceiver';
 
 const RSCPostsPageOverRedis = ({ requestId, ...props }, railsContext) => {
-  const { getValue, close } = listenToRequestData(requestId);
+  const { getValue, destroy } = listenToRequestData(requestId);
 
   const fetchPosts = () => getValue('posts');
   const fetchComments = (postId) => getValue(`comments:${postId}`);
   const fetchUser = (userId) => getValue(`user:${userId}`);
 
   if ('addPostSSRHook' in railsContext) {
-    railsContext.addPostSSRHook(close);
+    railsContext.addPostSSRHook(destroy);
   }
 
   return () => (

react_on_rails_pro/spec/dummy/client/app/ror-auto-load-components/RedisReceiver.jsx

@@ -34,10 +34,10 @@ const AsyncToggleContainer = async ({ children, childrenTitle, getValue }) => {
 };
 
 const RedisReceiver = ({ requestId, asyncToggleContainer }, railsContext) => {
-  const { getValue, close } = listenToRequestData(requestId);
+  const { getValue, destroy } = listenToRequestData(requestId);
 
   if ('addPostSSRHook' in railsContext) {
-    railsContext.addPostSSRHook(close);
+    railsContext.addPostSSRHook(destroy);
   }
 
   const UsedToggleContainer = asyncToggleContainer ? AsyncToggleContainer : ToggleContainer;