DF-25178: Optimizes blocksize-capital EA WS connection for subs/unsubs

.changeset/floppy-spiders-turn.md

@@ -0,0 +1,5 @@
+---
+'@chainlink/blocksize-capital-adapter': minor
+---
+
+Optimize blocksize-capital WS subscribe/unsubscribe via customSubscriptionMessages batching

packages/sources/blocksize-capital/src/transport/crypto-lwba.ts

@@ -1,11 +1,13 @@
-import { BaseEndpointTypes } from '../endpoint/crypto-lwba'
+import { EndpointContext } from '@chainlink/external-adapter-framework/adapter'
+import { SubscriptionDeltas } from '@chainlink/external-adapter-framework/transports/abstract/streaming'
 import { WebsocketReverseMappingTransport } from '@chainlink/external-adapter-framework/transports/websocket'
 import { makeLogger, ProviderResult } from '@chainlink/external-adapter-framework/util'
+import { BaseEndpointTypes } from '../endpoint/crypto-lwba'
 import {
   BaseMessage,
-  blocksizeDefaultUnsubscribeMessageBuilder,
   blocksizeDefaultWebsocketOpenHandler,
   buildBlocksizeWebsocketTickersMessage,
+  buildTicker,
 } from './utils'
 
 const logger = makeLogger('BlocksizeCapitalLwbaWebsocketEndpoint')
@@ -77,12 +79,33 @@ export const transport: WebsocketReverseMappingTransport<WsTransportTypes, strin
       },
     },
     builders: {
-      subscribeMessage: (params) => {
-        const pair = `${params.base}${params.quote}`.toUpperCase()
-        transport.setReverseMapping(pair, params)
-        return buildBlocksizeWebsocketTickersMessage('bidask_subscribe', pair)
+      customSubscriptionMessages: (
+        _context: EndpointContext<WsTransportTypes>,
+        subscriptions: SubscriptionDeltas<{ quote: string; base: string }>,
+      ) => {
+        const messages = []
+        if (subscriptions.new.length > 0) {
+          const pairsWithParams = subscriptions.new.map((param) => ({
+            ticker: buildTicker(param),
+            param: param,
+          }))
+          pairsWithParams.forEach(({ ticker, param }) => transport.setReverseMapping(ticker, param))
+          messages.push(
+            buildBlocksizeWebsocketTickersMessage(
+              'bidask_subscribe',
+              subscriptions.new.map(buildTicker),
+            ),
+          )
+        }
+        if (subscriptions.stale.length > 0) {
+          messages.push(
+            buildBlocksizeWebsocketTickersMessage(
+              'bidask_unsubscribe',
+              subscriptions.stale.map(buildTicker),
+            ),
+          )
+        }
+        return messages
       },
-      unsubscribeMessage: (params) =>
-        blocksizeDefaultUnsubscribeMessageBuilder(params.base, params.quote, 'bidask_unsubscribe'),
     },
   })

packages/sources/blocksize-capital/src/transport/price.ts

@@ -1,11 +1,13 @@
-import { BaseEndpointTypes } from '../endpoint/price'
-import { makeLogger } from '@chainlink/external-adapter-framework/util'
+import { EndpointContext } from '@chainlink/external-adapter-framework/adapter'
+import { SubscriptionDeltas } from '@chainlink/external-adapter-framework/transports/abstract/streaming'
 import { WebsocketReverseMappingTransport } from '@chainlink/external-adapter-framework/transports/websocket'
+import { makeLogger } from '@chainlink/external-adapter-framework/util'
+import { BaseEndpointTypes } from '../endpoint/price'
 import {
   BaseMessage,
-  blocksizeDefaultUnsubscribeMessageBuilder,
   blocksizeDefaultWebsocketOpenHandler,
   buildBlocksizeWebsocketTickersMessage,
+  buildTicker,
   handlePriceUpdates,
   VwapUpdate,
 } from './utils'
@@ -52,12 +54,33 @@ export const transport: WebsocketReverseMappingTransport<WsTransportTypes, strin
       },
     },
     builders: {
-      subscribeMessage: (params) => {
-        const pair = `${params.base}${params.quote}`.toUpperCase()
-        transport.setReverseMapping(pair, params)
-        return buildBlocksizeWebsocketTickersMessage('vwap_subscribe', pair)
+      customSubscriptionMessages: (
+        _context: EndpointContext<WsTransportTypes>,
+        subscriptions: SubscriptionDeltas<{ quote: string; base: string }>,
+      ) => {
+        const messages = []
+        if (subscriptions.new.length > 0) {
+          const pairsWithParams = subscriptions.new.map((param) => ({
+            ticker: buildTicker(param),
+            param: param,
+          }))
+          pairsWithParams.forEach(({ ticker, param }) => transport.setReverseMapping(ticker, param))
+          messages.push(
+            buildBlocksizeWebsocketTickersMessage(
+              'vwap_subscribe',
+              subscriptions.new.map(buildTicker),
+            ),
+          )
+        }
+        if (subscriptions.stale.length > 0) {
+          messages.push(
+            buildBlocksizeWebsocketTickersMessage(
+              'vwap_unsubscribe',
+              subscriptions.stale.map(buildTicker),
+            ),
+          )
+        }
+        return messages
       },
-      unsubscribeMessage: (params) =>
-        blocksizeDefaultUnsubscribeMessageBuilder(params.base, params.quote, 'vwap_unsubscribe'),
     },
   })

packages/sources/blocksize-capital/src/transport/utils.ts

@@ -19,31 +19,28 @@ export type VwapUpdate = {
   ts: number
 }
 
-export type ProviderParams = {
-  tickers?: string[]
-  api_key?: string
+export const buildBlocksizeWebsocketAuthMessage = (apiKey: string) => {
+  return {
+    jsonrpc: '2.0',
+    method: 'authentication_logon',
+    params: {
+      api_key: apiKey,
+    },
+  }
 }
 
-const buildBlocksizeWebsocketMessage = (method: string, params: ProviderParams): unknown => {
+export const buildBlocksizeWebsocketTickersMessage = (method: string, pairs: string[]) => {
   return {
     jsonrpc: '2.0',
     method: method,
-    params: params,
+    params: {
+      tickers: pairs,
+    },
   }
 }
 
-export const buildBlocksizeWebsocketAuthMessage = (apiKey: string) =>
-  buildBlocksizeWebsocketMessage('authentication_logon', { api_key: apiKey })
-export const buildBlocksizeWebsocketTickersMessage = (method: string, pair: string) =>
-  buildBlocksizeWebsocketMessage(method, { tickers: [pair] })
-
-export const blocksizeDefaultUnsubscribeMessageBuilder = (
-  base: string,
-  quote: string,
-  method: string,
-): unknown => {
-  const pair = `${base}${quote}`.toUpperCase()
-  return buildBlocksizeWebsocketTickersMessage(method, pair)
+export const buildTicker = (pair: { base: string; quote: string }) => {
+  return `${pair.base}${pair.quote}`.toUpperCase()
 }
 
 // use as open handler for standard WS connections

packages/sources/blocksize-capital/src/transport/vwap.ts

@@ -1,12 +1,14 @@
-import { BaseEndpointTypes } from '../endpoint/vwap'
+import { EndpointContext } from '@chainlink/external-adapter-framework/adapter'
+import { SubscriptionDeltas } from '@chainlink/external-adapter-framework/transports/abstract/streaming'
 import { WebsocketReverseMappingTransport } from '@chainlink/external-adapter-framework/transports/websocket'
 import { makeLogger } from '@chainlink/external-adapter-framework/util'
+import { BaseEndpointTypes } from '../endpoint/vwap'
 import {
   BaseMessage,
   VwapUpdate,
-  blocksizeDefaultUnsubscribeMessageBuilder,
   blocksizeDefaultWebsocketOpenHandler,
   buildBlocksizeWebsocketTickersMessage,
+  buildTicker,
   handlePriceUpdates,
 } from './utils'
 
@@ -60,16 +62,33 @@ export const transport: WebsocketReverseMappingTransport<WsTransportTypes, strin
       },
     },
     builders: {
-      subscribeMessage: (params) => {
-        const pair = `${params.base}${params.quote}`.toUpperCase()
-        transport.setReverseMapping(pair, params)
-        return buildBlocksizeWebsocketTickersMessage('fixedvwap_subscribe', pair)
+      customSubscriptionMessages: (
+        _context: EndpointContext<WsTransportTypes>,
+        subscriptions: SubscriptionDeltas<{ quote: string; base: string }>,
+      ) => {
+        const messages = []
+        if (subscriptions.new.length > 0) {
+          const pairsWithParams = subscriptions.new.map((param) => ({
+            ticker: buildTicker(param),
+            param: param,
+          }))
+          pairsWithParams.forEach(({ ticker, param }) => transport.setReverseMapping(ticker, param))
+          messages.push(
+            buildBlocksizeWebsocketTickersMessage(
+              'fixedvwap_subscribe',
+              subscriptions.new.map(buildTicker),
+            ),
+          )
+        }
+        if (subscriptions.stale.length > 0) {
+          messages.push(
+            buildBlocksizeWebsocketTickersMessage(
+              'fixedvwap_unsubscribe',
+              subscriptions.stale.map(buildTicker),
+            ),
+          )
+        }
+        return messages
       },
-      unsubscribeMessage: (params) =>
-        blocksizeDefaultUnsubscribeMessageBuilder(
-          params.base,
-          params.quote,
-          'fixedvwap_unsubscribe',
-        ),
     },
   })