feat(runtime): resubscribe websocket topics (#6)

Author: ark-65

CHANGELOG.md

@@ -4,4 +4,5 @@ English | [中文](./CHANGELOG.zh-CN.md)
 
 ## [Unreleased]
 
+- feat(runtime): resubscribe Socket.IO runtime page topics after reconnect with replay cursor support.
 - feat(runtime): add a Socket.IO WebSocket manager for BFF runtime page subscriptions.

CHANGELOG.zh-CN.md

@@ -4,4 +4,5 @@
 
 ## [Unreleased]
 
+- feat(runtime): Socket.IO runtime page topic 重连后自动重订阅，并支持携带 replay cursor。
 - feat(runtime): 新增用于 BFF runtime page 订阅的 Socket.IO WebSocket manager。

projects/runtime-adapter/src/lib/runtime-adapter.spec.ts

@@ -207,6 +207,7 @@ describe('createSocketIoWebSocketManager', () => {
     expect(socket.connectCalls).toBe(1);
     expect(socket.disconnectCalls).toBe(1);
     expect(socket.listenerCount('runtimeManagerExecuted')).toBe(0);
+    expect(socket.listenerCount('connect')).toBe(0);
   });
 
   it('subscribes to BFF runtime page topics without cursor', async () => {
@@ -239,6 +240,62 @@ describe('createSocketIoWebSocketManager', () => {
     ]);
   });
 
+  it('resubscribes active topics after socket reconnect', async () => {
+    const socket = new FakeRuntimeSocket();
+    const manager = createSocketIoWebSocketManager({
+      socketFactory: () => socket,
+    });
+    const topic = 'tenant.tenant-a.page.orders.instance.instance-1';
+
+    await manager.connect();
+    await manager.subscribe(topic, vi.fn());
+    socket.receive('connect', undefined);
+
+    expect(socket.emits).toEqual([
+      {
+        event: 'subscribePage',
+        payload: {
+          tenantId: 'tenant-a',
+          pageId: 'orders',
+          pageInstanceId: 'instance-1',
+        },
+      },
+      {
+        event: 'subscribePage',
+        payload: {
+          tenantId: 'tenant-a',
+          pageId: 'orders',
+          pageInstanceId: 'instance-1',
+        },
+      },
+    ]);
+  });
+
+  it('resubscribes a topic once when multiple handlers share it', async () => {
+    const socket = new FakeRuntimeSocket();
+    const manager = createSocketIoWebSocketManager({
+      socketFactory: () => socket,
+    });
+    const topic = 'tenant.tenant-a.page.orders.instance.instance-1';
+
+    await manager.connect();
+    await manager.subscribe(topic, vi.fn());
+    await manager.subscribe(topic, vi.fn());
+    socket.emits.length = 0;
+    socket.receive('connect', undefined);
+
+    expect(socket.emits).toEqual([
+      {
+        event: 'subscribePage',
+        payload: {
+          tenantId: 'tenant-a',
+          pageId: 'orders',
+          pageInstanceId: 'instance-1',
+        },
+      },
+    ]);
+  });
+
   it('subscribes with replay cursor options', async () => {
     const socket = new FakeRuntimeSocket();
     const manager = createSocketIoWebSocketManager({
@@ -269,6 +326,76 @@ describe('createSocketIoWebSocketManager', () => {
     ]);
   });
 
+  it('uses the latest replay cursor when resubscribing after reconnect', async () => {
+    const socket = new FakeRuntimeSocket();
+    const manager = createSocketIoWebSocketManager({
+      socketFactory: () => socket,
+    });
+    const topic = 'tenant.tenant-a.page.orders.instance.instance-1';
+    const handler = vi.fn();
+
+    await subscribeWithOptions(manager, topic, handler, {
+      afterReplayId: '42-0',
+    });
+    socket.receive('runtimeManagerExecuted', {
+      type: 'runtime.manager.executed',
+      topic,
+      page: {
+        tenantId: 'tenant-a',
+        pageId: 'orders',
+        pageInstanceId: 'instance-1',
+      },
+      replayId: '43-0',
+      patchState: {},
+      refreshedDatasourceIds: [],
+      runActionIds: [],
+    });
+    socket.emits.length = 0;
+    socket.receive('connect', undefined);
+
+    expect(socket.emits).toEqual([
+      {
+        event: 'subscribePage',
+        payload: {
+          tenantId: 'tenant-a',
+          pageId: 'orders',
+          pageInstanceId: 'instance-1',
+          afterReplayId: '43-0',
+        },
+      },
+    ]);
+  });
+
+  it('falls back to the original replay cursor when no newer event was received', async () => {
+    const socket = new FakeRuntimeSocket();
+    const manager = createSocketIoWebSocketManager({
+      socketFactory: () => socket,
+    });
+
+    await subscribeWithOptions(
+      manager,
+      'tenant.tenant-a.page.orders.instance.instance-1',
+      vi.fn(),
+      {
+        afterReplayId: '42-0',
+      },
+    );
+    socket.emits.length = 0;
+    socket.receive('connect', undefined);
+
+    expect(socket.emits).toEqual([
+      {
+        event: 'subscribePage',
+        payload: {
+          tenantId: 'tenant-a',
+          pageId: 'orders',
+          pageInstanceId: 'instance-1',
+          afterReplayId: '42-0',
+        },
+      },
+    ]);
+  });
+
   it('dispatches manager events only to matching topic handlers', async () => {
     const socket = new FakeRuntimeSocket();
     const manager = createSocketIoWebSocketManager({
@@ -309,6 +436,8 @@ describe('createSocketIoWebSocketManager', () => {
     await manager.connect();
     await manager.subscribe(topic, handler);
     await manager.unsubscribe(topic, handler);
+    socket.emits.length = 0;
+    socket.receive('connect', undefined);
     socket.receive('runtimeManagerExecuted', {
       type: 'runtime.manager.executed',
       topic,
@@ -323,6 +452,26 @@ describe('createSocketIoWebSocketManager', () => {
     });
 
     expect(handler).not.toHaveBeenCalled();
+    expect(socket.emits).toEqual([]);
+  });
+
+  it('clears subscriptions and replay cursors on disconnect', async () => {
+    const socket = new FakeRuntimeSocket();
+    const manager = createSocketIoWebSocketManager({
+      socketFactory: () => socket,
+    });
+    const topic = 'tenant.tenant-a.page.orders.instance.instance-1';
+
+    await subscribeWithOptions(manager, topic, vi.fn(), {
+      afterReplayId: '42-0',
+    });
+    await manager.disconnect();
+    socket.emits.length = 0;
+    socket.receive('connect', undefined);
+
+    expect(socket.emits).toEqual([]);
+    expect(socket.listenerCount('runtimeManagerExecuted')).toBe(0);
+    expect(socket.listenerCount('connect')).toBe(0);
   });
 });
 

projects/runtime-adapter/src/lib/runtime-adapter.ts

@@ -269,6 +269,8 @@ export function createSocketIoWebSocketManager(
   options: SocketIoWebSocketManagerOptions = {},
 ): NgxLowcodeWebSocketManager {
   const subscriptions = new Map<string, Set<NgxLowcodeWebSocketEventHandler>>();
+  const originalReplayIdsByTopic = new Map<string, string>();
+  const latestReplayIdsByTopic = new Map<string, string>();
   const socketUrl = resolveSocketUrl(options);
   const socketOptions: Partial<ManagerOptions & SocketOptions> = {
     autoConnect: false,
@@ -283,6 +285,10 @@ export function createSocketIoWebSocketManager(
       return;
     }
 
+    if (event.replayId) {
+      latestReplayIdsByTopic.set(event.topic, event.replayId);
+    }
+
     const handlers = subscriptions.get(event.topic);
     if (!handlers) {
       return;
@@ -293,6 +299,12 @@ export function createSocketIoWebSocketManager(
     }
   };
 
+  const handleConnect = (): void => {
+    for (const channel of subscriptions.keys()) {
+      emitSubscribePage(channel);
+    }
+  };
+
   return {
     connect: () => {
       getSocket().connect();
@@ -304,25 +316,31 @@ export function createSocketIoWebSocketManager(
     ) => {
       const page = parseRuntimePageTopic(channel);
       const handlers = subscriptions.get(channel) ?? new Set<NgxLowcodeWebSocketEventHandler>();
+      const isFirstTopicHandler = handlers.size === 0;
       handlers.add(handler);
       subscriptions.set(channel, handlers);
-      getSocket().emit('subscribePage', {
-        ...page,
-        ...(subscribeOptions?.afterReplayId
-          ? { afterReplayId: subscribeOptions.afterReplayId }
-          : {}),
-      });
+      if (subscribeOptions?.afterReplayId) {
+        originalReplayIdsByTopic.set(channel, subscribeOptions.afterReplayId);
+      }
+      if (isFirstTopicHandler) {
+        emitSubscribePage(channel, page);
+      }
     },
     unsubscribe: (channel: string, handler: NgxLowcodeWebSocketEventHandler) => {
       const handlers = subscriptions.get(channel);
       handlers?.delete(handler);
       if (handlers && handlers.size === 0) {
         subscriptions.delete(channel);
+        originalReplayIdsByTopic.delete(channel);
+        latestReplayIdsByTopic.delete(channel);
       }
     },
     disconnect: () => {
       subscriptions.clear();
+      originalReplayIdsByTopic.clear();
+      latestReplayIdsByTopic.clear();
       socket?.off('runtimeManagerExecuted', handleManagerExecuted);
+      socket?.off('connect', handleConnect);
       socket?.disconnect();
       socket = undefined;
       listenerAttached = false;
@@ -333,10 +351,19 @@ export function createSocketIoWebSocketManager(
     socket ??= socketFactory(socketUrl, socketOptions);
     if (!listenerAttached) {
       socket.on('runtimeManagerExecuted', handleManagerExecuted);
+      socket.on('connect', handleConnect);
       listenerAttached = true;
     }
     return socket;
   }
+
+  function emitSubscribePage(channel: string, page = parseRuntimePageTopic(channel)): void {
+    const replayId = latestReplayIdsByTopic.get(channel) ?? originalReplayIdsByTopic.get(channel);
+    getSocket().emit('subscribePage', {
+      ...page,
+      ...(replayId ? { afterReplayId: replayId } : {}),
+    });
+  }
 }
 
 export function parseRuntimePageTopic(topic: string): RuntimePageTopicRef {