dxFeed handle multi messages (#4398)

* dxFeed handle multi messages

* Update .changeset/witty-experts-exercise.md

---------

Co-authored-by: mmcallister-cll <[email protected]>
Co-authored-by: app-token-issuer-data-feeds[bot] <134377064+app-token-issuer-data-feeds[bot]@users.noreply.github.com>

Author: 3 people

.changeset/witty-experts-exercise.md

@@ -0,0 +1,5 @@
+---
+'@chainlink/dxfeed-adapter': patch
+---
+
+Handle multiple tickers in one message

packages/sources/dxfeed/src/transport/price-ws.ts

@@ -5,14 +5,14 @@ const eventSymbolIndex = 0
 const priceIndex = 6
 
 export const transport = buildWsTransport<BaseEndpointTypes>(
-  (params) => ({ Trade: [params.base.toUpperCase()] }),
+  (params) => [{ Trade: [params.base.toUpperCase()] }],
   (message) => {
     if (message[0].data[0] != 'Trade' && message[0].data[0][0] != 'Trade') {
       return []
     }
 
-    const base = message[0].data[1][eventSymbolIndex]
-    const price = message[0].data[1][priceIndex]
+    const base = message[0].data[1][eventSymbolIndex].toString()
+    const price = Number(message[0].data[1][priceIndex])
 
     return [
       {

packages/sources/dxfeed/src/transport/stock-quotes.ts

@@ -14,52 +14,54 @@ const askSizeIndex = 11
 const dataLength = 12
 
 export const transport = buildWsTransport<BaseEndpointTypes>(
-  (params) => ({ Quote: [params.base.toUpperCase()] }),
+  (params) => [{ Quote: [params.base.toUpperCase()] }],
   (message) => {
     if (message[0].data[0] != 'Quote' && message[0].data[0][0] != 'Quote') {
       return []
     }
 
     const data = message[0].data[1]
 
-    if (data.length != dataLength) {
-      logger.warn(`${JSON.stringify(data)} is invalid since it doesn't have ${dataLength} fields.`)
+    if (data.length % dataLength != 0) {
+      logger.warn(
+        `${JSON.stringify(data)} is invalid since length is not multiple of ${dataLength}.`,
+      )
       return []
     }
 
-    const bidPrice = Number(data[bidPriceIndex])
-    const askPrice = Number(data[askPriceIndex])
+    return Array.from({ length: data.length / dataLength }, (_, i) => i * dataLength).map((i) => {
+      const bidPrice = Number(data[i + bidPriceIndex])
+      const askPrice = Number(data[i + askPriceIndex])
 
-    let midPrice: number
+      let midPrice: number
 
-    if (bidPrice == 0) {
-      midPrice = askPrice
-    } else if (askPrice == 0) {
-      midPrice = bidPrice
-    } else {
-      midPrice = (askPrice + bidPrice) / 2
-    }
+      if (bidPrice == 0) {
+        midPrice = askPrice
+      } else if (askPrice == 0) {
+        midPrice = bidPrice
+      } else {
+        midPrice = (askPrice + bidPrice) / 2
+      }
 
-    return [
-      {
-        params: { base: data[eventSymbolIndex] },
+      return {
+        params: { base: data[i + eventSymbolIndex].toString() },
         response: {
           result: null,
           data: {
             mid_price: midPrice,
             bid_price: bidPrice,
-            bid_volume: Number(data[bidSizeIndex]),
+            bid_volume: Number(data[i + bidSizeIndex]),
             ask_price: askPrice,
-            ask_volume: Number(data[askSizeIndex]),
+            ask_volume: Number(data[i + askSizeIndex]),
           },
           timestamps: {
             providerIndicatedTimeUnixMs: Math.max(
-              Number(data[bidTimeIndex]),
-              Number(data[askTimeIndex]),
+              Number(data[i + bidTimeIndex]),
+              Number(data[i + askTimeIndex]),
             ),
           },
         },
-      },
-    ]
+      }
+    })
   },
 )

packages/sources/dxfeed/src/transport/ws.ts

@@ -10,23 +10,7 @@ type DXFeedMessage = {
   channel: string
   clientId?: string
   id: string
-  data: [
-    string,
-    [
-      string,
-      number,
-      number,
-      number,
-      number,
-      string,
-      number,
-      string,
-      number,
-      number,
-      number,
-      number,
-    ],
-  ]
+  data: [string, (string | number)[]]
   successful?: boolean
   advice?: {
     interval: number
@@ -111,7 +95,7 @@ class DxFeedWebsocketTransport<T extends BaseTransportTypes> extends WebSocketTr
 export function buildWsTransport<T extends BaseTransportTypes>(
   formatTicker: (
     base: TypeFromDefinition<(T & ProviderTypes)['Parameters']>,
-  ) => Record<string, string[]>,
+  ) => Record<string, string[]>[],
   processMessage: (message: DXFeedMessage) => ProviderResult<T & ProviderTypes>[],
 ): WebSocketTransport<T & ProviderTypes> {
   const wsTransport: DxFeedWebsocketTransport<T> = new DxFeedWebsocketTransport({
@@ -153,22 +137,18 @@ export function buildWsTransport<T extends BaseTransportTypes>(
 
     builders: {
       subscribeMessage: (params) => {
-        return [
-          {
-            channel: SERVICE_SUB,
-            data: { add: formatTicker(params) },
-            clientId: wsTransport.connectionClientId,
-          },
-        ]
+        return formatTicker(params).map((ticker) => ({
+          channel: SERVICE_SUB,
+          data: { add: ticker },
+          clientId: wsTransport.connectionClientId,
+        }))
       },
       unsubscribeMessage: (params) => {
-        return [
-          {
-            channel: SERVICE_SUB,
-            data: { remove: formatTicker(params) },
-            clientId: wsTransport.connectionClientId,
-          },
-        ]
+        return formatTicker(params).map((ticker) => ({
+          channel: SERVICE_SUB,
+          data: { remove: ticker },
+          clientId: wsTransport.connectionClientId,
+        }))
       },
     },
   })

packages/sources/dxfeed/test/integration/__snapshots__/adapter-ws.test.ts.snap

@@ -68,6 +68,44 @@ exports[`websocket quote endpoint should return success 1`] = `
 }
 `;
 
+exports[`websocket quote endpoint should return success for 1st item in multi message 1`] = `
+{
+  "data": {
+    "ask_price": 8,
+    "ask_volume": 9,
+    "bid_price": 5,
+    "bid_volume": 6,
+    "mid_price": 6.5,
+  },
+  "result": null,
+  "statusCode": 200,
+  "timestamps": {
+    "providerDataReceivedUnixMs": 1018,
+    "providerDataStreamEstablishedUnixMs": 1010,
+    "providerIndicatedTimeUnixMs": 7,
+  },
+}
+`;
+
+exports[`websocket quote endpoint should return success for 2nd item in multi message 1`] = `
+{
+  "data": {
+    "ask_price": 17,
+    "ask_volume": 18,
+    "bid_price": 14,
+    "bid_volume": 15,
+    "mid_price": 15.5,
+  },
+  "result": null,
+  "statusCode": 200,
+  "timestamps": {
+    "providerDataReceivedUnixMs": 1018,
+    "providerDataStreamEstablishedUnixMs": 1010,
+    "providerIndicatedTimeUnixMs": 16,
+  },
+}
+`;
+
 exports[`websocket stock endpoint should return success 1`] = `
 {
   "data": {

packages/sources/dxfeed/test/integration/adapter-ws.test.ts

@@ -19,11 +19,18 @@ describe('websocket', () => {
     base: 'TSLA',
     transport: 'ws',
   }
-
   const quoteData = {
     base: 'TSLA',
     endpoint: 'stock_quotes',
   }
+  const quoteMulti1Data = {
+    base: 'MULTI_1',
+    endpoint: 'stock_quotes',
+  }
+  const quoteMulti2Data = {
+    base: 'MULTI_2',
+    endpoint: 'stock_quotes',
+  }
 
   beforeAll(async () => {
     oldEnv = JSON.parse(JSON.stringify(process.env))
@@ -44,8 +51,10 @@ describe('websocket', () => {
 
     // Send initial request to start background execute and wait for cache to be filled with result
     await testAdapter.request(quoteData)
+    await testAdapter.request(quoteMulti1Data)
+    await testAdapter.request(quoteMulti2Data)
     await testAdapter.request(stockData)
-    await testAdapter.waitForCache(4)
+    await testAdapter.waitForCache(6)
   })
 
   afterAll(async () => {
@@ -68,6 +77,16 @@ describe('websocket', () => {
       expect(response.json()).toMatchSnapshot()
     })
 
+    it('should return success for 1st item in multi message', async () => {
+      const response = await testAdapter.request(quoteMulti1Data)
+      expect(response.json()).toMatchSnapshot()
+    })
+
+    it('should return success for 2nd item in multi message', async () => {
+      const response = await testAdapter.request(quoteMulti2Data)
+      expect(response.json()).toMatchSnapshot()
+    })
+
     it('should return bid when ask is 0', async () => {
       const response = await testAdapter.request({
         base: 'NO_ASK',

packages/sources/dxfeed/test/integration/fixtures.ts

@@ -63,6 +63,19 @@ export const mockWebSocketServer = (URL: string): MockWebsocketServer => {
     },
   ]
 
+  const multiQuoteReponse = [
+    {
+      data: [
+        'Quote',
+        [
+          ...['MULTI_1', 1, 2, 3, 4, 'V', 5, 6, 7, 'V', 8, 9],
+          ...['MULTI_2', 10, 11, 12, 13, 'V', 14, 15, 16, 'V', 17, 18],
+        ],
+      ],
+      channel: '/service/data',
+    },
+  ]
+
   const noBidQuoteReponse = [
     {
       data: [
@@ -125,6 +138,7 @@ export const mockWebSocketServer = (URL: string): MockWebsocketServer => {
     )
     socket.on('message', () => {
       socket.send(JSON.stringify(quoteReponse))
+      socket.send(JSON.stringify(multiQuoteReponse))
       socket.send(JSON.stringify(noBidQuoteReponse))
       socket.send(JSON.stringify(noAskQuoteReponse))
       socket.send(JSON.stringify(invalidQuoteReponse))