Add reconnect test

packages/transport-http/src/subscriptions/stream-controller.test.ts

@@ -97,4 +97,108 @@ describe("StreamController", () => {
     subscription.unsubscribe()
     await serverPromise
   })
+
+  test("reconnects to stream on close", async () => {
+    const config = {
+      hostname: "wss://localhost:8080",
+      reconnectInterval: 1000,
+      reconnectAttempts: 1,
+    }
+    const streamController = new StreamController(config)
+    const topic = "topic" as SubscriptionTopic
+    const args = {key: "value"} as any
+    const onData = jest.fn()
+    const onError = jest.fn()
+
+    let mockWs = new WS("wss://localhost:8080")
+
+    let serverPromise = (async () => {
+      await mockWs.connected
+
+      const msg = (await mockWs.nextMessage) as string
+      const data = JSON.parse(msg) as SubscribeMessageRequest
+      expect(data).toEqual({
+        action: "subscribe",
+        topic,
+        arguments: args,
+      })
+
+      const response: SubscribeMessageResponse = {
+        id: "id1",
+        action: "subscribe",
+        success: true,
+        topic,
+      }
+      mockWs.send(JSON.stringify(response))
+    })()
+
+    const [subscription] = await Promise.all([
+      streamController.subscribe({
+        topic,
+        args,
+        onData,
+        onError,
+      }),
+      serverPromise,
+    ])
+
+    expect(subscription).toBeDefined()
+    expect(subscription.unsubscribe).toBeInstanceOf(Function)
+
+    serverPromise = (async () => {
+      const data = {
+        id: "id1",
+        data: {key: "value"},
+      } as SubscriptionDataMessage
+      mockWs.send(JSON.stringify(data))
+    })()
+
+    await serverPromise
+
+    expect(onData).toHaveBeenCalledTimes(1)
+    expect(onData).toHaveBeenCalledWith({key: "value"})
+    expect(onError).toHaveBeenCalledTimes(0)
+
+    // Close the connection and create a new one
+    mockWs.close()
+    mockWs = new WS("wss://localhost:8080")
+
+    serverPromise = (async () => {
+      await mockWs.connected
+
+      const msg = (await mockWs.nextMessage) as string
+      const data = JSON.parse(msg) as SubscribeMessageRequest
+      expect(data).toEqual({
+        action: "subscribe",
+        topic,
+        arguments: args,
+      })
+
+      const response: SubscribeMessageResponse = {
+        id: "id2",
+        action: "subscribe",
+        success: true,
+        topic,
+      }
+      mockWs.send(JSON.stringify(response))
+    })()
+
+    await serverPromise
+
+    // Wait for client to register the new subscription
+    await new Promise(resolve => setTimeout(resolve, 0))
+
+    serverPromise = (async () => {
+      const data = {
+        id: "id2",
+        data: {key: "value2"},
+      } as SubscriptionDataMessage
+      mockWs.send(JSON.stringify(data))
+    })()
+
+    await serverPromise
+
+    expect(onData).toHaveBeenCalledTimes(2)
+    expect(onData.mock.calls[1]).toEqual([{key: "value2"}])
+  })
 })

packages/transport-http/src/subscriptions/stream-controller.ts

@@ -50,7 +50,7 @@ export class StreamController implements SubscriptionTransport {
   constructor(config: StreamControllerConfig) {
     this.config = {
       hostname: config.hostname,
-      reconnectInterval: config.reconnectInterval || 1000,
+      reconnectInterval: config.reconnectInterval || 2000,
       reconnectAttempts: config.reconnectAttempts || 10,
     }
   }