Aleno: resubscribe after reconnect (#3819)

Author: dskloetc

.changeset/pretty-goats-add.md

@@ -0,0 +1,5 @@
+---
+'@chainlink/aleno-adapter': patch
+---
+
+Bug fix: Recreate subscriptions after reconnecting

packages/sources/aleno/src/transport/price-socketio.ts

@@ -1,14 +1,14 @@
-import { io, Socket } from 'socket.io-client'
+import { EndpointContext } from '@chainlink/external-adapter-framework/adapter'
 import { TransportDependencies } from '@chainlink/external-adapter-framework/transports'
-import { makeLogger, sleep } from '@chainlink/external-adapter-framework/util'
 import {
   StreamingTransport,
   SubscriptionDeltas,
 } from '@chainlink/external-adapter-framework/transports/abstract/streaming'
+import { makeLogger, sleep } from '@chainlink/external-adapter-framework/util'
 import { TypeFromDefinition } from '@chainlink/external-adapter-framework/validation/input-params'
-import { EndpointContext } from '@chainlink/external-adapter-framework/adapter'
-import { BaseEndpointTypes } from '../endpoint/price'
+import { io, Socket } from 'socket.io-client'
 import { config } from '../config'
+import { BaseEndpointTypes } from '../endpoint/price'
 
 const logger = makeLogger('SocketIOTransport')
 
@@ -220,6 +220,7 @@ export class SocketIOTransport extends StreamingTransport<SocketIOTransportTypes
 
       this.socket.on('connect', () => {
         logger.info({ msg: 'Connection open' })
+        this.confirmedSubscriptions = new Set<string>()
       })
 
       this.socket.on('disconnect', (reason, details) => {

packages/sources/aleno/test/unit/adapter-socket.test.ts

@@ -135,6 +135,51 @@ describe('SocketIOTransport', () => {
     expect(subscribeSpy).toBeCalledTimes(1)
   })
 
+  it('should resubscribe after reconnect', async () => {
+    const subscribeCalls: SubscribeCall[] = []
+    const subscribeSpy = jest
+      .fn()
+      .mockImplementation((subscriptions: string[], callback: SubscribeCallback) => {
+        subscribeCalls.push({ subscriptions, callback })
+      })
+    mockSocket.clientMock.on('subscribe', subscribeSpy)
+
+    const transport = new SocketIOTransport()
+
+    transport.streamHandler(context, emptySubscriptions)
+    await jest.advanceTimersByTimeAsync(BACKGROUND_EXECUTE_MS_SSE)
+
+    mockSocket.clientMock.emit('connect')
+
+    const subscriptions = [{ base: 'FRAC', quote: 'USD' }]
+    const streamHandlerPromise = transport.streamHandler(context, {
+      desired: subscriptions,
+      new: subscriptions,
+      stale: [],
+    })
+
+    expect(subscribeSpy).toBeCalledTimes(1)
+    expect(subscribeSpy).toBeCalledWith(['FRAC/USD'], expect.any(Function))
+
+    subscribeCalls[0].callback({
+      status: 'ok',
+      involvedSubscriptions: ['frac/usd'],
+      subscriptionsAfterUpdate: ['frac/usd'],
+    })
+    await jest.advanceTimersByTimeAsync(BACKGROUND_EXECUTE_MS_SSE)
+    await streamHandlerPromise
+
+    // Simulate a reconnect
+    mockSocket.clientMock.emit('connect')
+
+    transport.streamHandler(context, { desired: subscriptions, new: [], stale: [] })
+    await jest.advanceTimersByTimeAsync(BACKGROUND_EXECUTE_MS_SSE)
+
+    // Resubscribed
+    expect(subscribeSpy).toBeCalledTimes(2)
+    expect(subscribeSpy).toHaveBeenNthCalledWith(2, ['FRAC/USD'], expect.any(Function))
+  })
+
   it('should unsubscribe', async () => {
     const subscribeCalls: SubscribeCall[] = []
     const subscribeSpy = jest.fn().mockImplementation((subscriptions, callback) => {