Skip to content

Commit 65a702d

Browse files
cl-efornaciaricl-efornaciari
committed
DF-25178: Optimizes blocksize-capital EA WS connection for subs/unsubs
1 parent 66af2ae commit 65a702d

4 files changed

Lines changed: 108 additions & 46 deletions

File tree

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

Lines changed: 31 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -1,11 +1,13 @@
1-
import { BaseEndpointTypes } from '../endpoint/crypto-lwba'
1+
import { EndpointContext } from '@chainlink/external-adapter-framework/adapter'
2+
import { SubscriptionDeltas } from '@chainlink/external-adapter-framework/transports/abstract/streaming'
23
import { WebsocketReverseMappingTransport } from '@chainlink/external-adapter-framework/transports/websocket'
34
import { makeLogger, ProviderResult } from '@chainlink/external-adapter-framework/util'
5+
import { BaseEndpointTypes } from '../endpoint/crypto-lwba'
46
import {
57
BaseMessage,
6-
blocksizeDefaultUnsubscribeMessageBuilder,
78
blocksizeDefaultWebsocketOpenHandler,
89
buildBlocksizeWebsocketTickersMessage,
10+
buildTicker,
911
} from './utils'
1012

1113
const logger = makeLogger('BlocksizeCapitalLwbaWebsocketEndpoint')
@@ -77,12 +79,33 @@ export const transport: WebsocketReverseMappingTransport<WsTransportTypes, strin
7779
},
7880
},
7981
builders: {
80-
subscribeMessage: (params) => {
81-
const pair = `${params.base}${params.quote}`.toUpperCase()
82-
transport.setReverseMapping(pair, params)
83-
return buildBlocksizeWebsocketTickersMessage('bidask_subscribe', pair)
82+
customSubscriptionMessages: (
83+
_context: EndpointContext<WsTransportTypes>,
84+
subscriptions: SubscriptionDeltas<{ quote: string; base: string }>,
85+
) => {
86+
const messages = []
87+
if (subscriptions.new.length > 0) {
88+
const pairsWithParams = subscriptions.new.map((param) => ({
89+
ticker: buildTicker(param),
90+
param: param,
91+
}))
92+
pairsWithParams.forEach(({ ticker, param }) => transport.setReverseMapping(ticker, param))
93+
messages.push(
94+
buildBlocksizeWebsocketTickersMessage(
95+
'bidask_subscribe',
96+
subscriptions.new.map(buildTicker),
97+
),
98+
)
99+
}
100+
if (subscriptions.stale.length > 0) {
101+
messages.push(
102+
buildBlocksizeWebsocketTickersMessage(
103+
'bidask_unsubscribe',
104+
subscriptions.stale.map(buildTicker),
105+
),
106+
)
107+
}
108+
return messages
84109
},
85-
unsubscribeMessage: (params) =>
86-
blocksizeDefaultUnsubscribeMessageBuilder(params.base, params.quote, 'bidask_unsubscribe'),
87110
},
88111
})

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

Lines changed: 32 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -1,11 +1,13 @@
1-
import { BaseEndpointTypes } from '../endpoint/price'
2-
import { makeLogger } from '@chainlink/external-adapter-framework/util'
1+
import { EndpointContext } from '@chainlink/external-adapter-framework/adapter'
2+
import { SubscriptionDeltas } from '@chainlink/external-adapter-framework/transports/abstract/streaming'
33
import { WebsocketReverseMappingTransport } from '@chainlink/external-adapter-framework/transports/websocket'
4+
import { makeLogger } from '@chainlink/external-adapter-framework/util'
5+
import { BaseEndpointTypes } from '../endpoint/price'
46
import {
57
BaseMessage,
6-
blocksizeDefaultUnsubscribeMessageBuilder,
78
blocksizeDefaultWebsocketOpenHandler,
89
buildBlocksizeWebsocketTickersMessage,
10+
buildTicker,
911
handlePriceUpdates,
1012
VwapUpdate,
1113
} from './utils'
@@ -52,12 +54,33 @@ export const transport: WebsocketReverseMappingTransport<WsTransportTypes, strin
5254
},
5355
},
5456
builders: {
55-
subscribeMessage: (params) => {
56-
const pair = `${params.base}${params.quote}`.toUpperCase()
57-
transport.setReverseMapping(pair, params)
58-
return buildBlocksizeWebsocketTickersMessage('vwap_subscribe', pair)
57+
customSubscriptionMessages: (
58+
_context: EndpointContext<WsTransportTypes>,
59+
subscriptions: SubscriptionDeltas<{ quote: string; base: string }>,
60+
) => {
61+
const messages = []
62+
if (subscriptions.new.length > 0) {
63+
const pairsWithParams = subscriptions.new.map((param) => ({
64+
ticker: buildTicker(param),
65+
param: param,
66+
}))
67+
pairsWithParams.forEach(({ ticker, param }) => transport.setReverseMapping(ticker, param))
68+
messages.push(
69+
buildBlocksizeWebsocketTickersMessage(
70+
'vwap_subscribe',
71+
subscriptions.new.map(buildTicker),
72+
),
73+
)
74+
}
75+
if (subscriptions.stale.length > 0) {
76+
messages.push(
77+
buildBlocksizeWebsocketTickersMessage(
78+
'vwap_unsubscribe',
79+
subscriptions.stale.map(buildTicker),
80+
),
81+
)
82+
}
83+
return messages
5984
},
60-
unsubscribeMessage: (params) =>
61-
blocksizeDefaultUnsubscribeMessageBuilder(params.base, params.quote, 'vwap_unsubscribe'),
6285
},
6386
})

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

Lines changed: 14 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -19,31 +19,28 @@ export type VwapUpdate = {
1919
ts: number
2020
}
2121

22-
export type ProviderParams = {
23-
tickers?: string[]
24-
api_key?: string
22+
export const buildBlocksizeWebsocketAuthMessage = (apiKey: string) => {
23+
return {
24+
jsonrpc: '2.0',
25+
method: 'authentication_logon',
26+
params: {
27+
api_key: apiKey,
28+
},
29+
}
2530
}
2631

27-
const buildBlocksizeWebsocketMessage = (method: string, params: ProviderParams): unknown => {
32+
export const buildBlocksizeWebsocketTickersMessage = (method: string, pairs: string[]) => {
2833
return {
2934
jsonrpc: '2.0',
3035
method: method,
31-
params: params,
36+
params: {
37+
tickers: pairs,
38+
},
3239
}
3340
}
3441

35-
export const buildBlocksizeWebsocketAuthMessage = (apiKey: string) =>
36-
buildBlocksizeWebsocketMessage('authentication_logon', { api_key: apiKey })
37-
export const buildBlocksizeWebsocketTickersMessage = (method: string, pair: string) =>
38-
buildBlocksizeWebsocketMessage(method, { tickers: [pair] })
39-
40-
export const blocksizeDefaultUnsubscribeMessageBuilder = (
41-
base: string,
42-
quote: string,
43-
method: string,
44-
): unknown => {
45-
const pair = `${base}${quote}`.toUpperCase()
46-
return buildBlocksizeWebsocketTickersMessage(method, pair)
42+
export const buildTicker = (pair: { base: string; quote: string }) => {
43+
return `${pair.base}${pair.quote}`.toUpperCase()
4744
}
4845

4946
// use as open handler for standard WS connections

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

Lines changed: 31 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -1,12 +1,14 @@
1-
import { BaseEndpointTypes } from '../endpoint/vwap'
1+
import { EndpointContext } from '@chainlink/external-adapter-framework/adapter'
2+
import { SubscriptionDeltas } from '@chainlink/external-adapter-framework/transports/abstract/streaming'
23
import { WebsocketReverseMappingTransport } from '@chainlink/external-adapter-framework/transports/websocket'
34
import { makeLogger } from '@chainlink/external-adapter-framework/util'
5+
import { BaseEndpointTypes } from '../endpoint/vwap'
46
import {
57
BaseMessage,
68
VwapUpdate,
7-
blocksizeDefaultUnsubscribeMessageBuilder,
89
blocksizeDefaultWebsocketOpenHandler,
910
buildBlocksizeWebsocketTickersMessage,
11+
buildTicker,
1012
handlePriceUpdates,
1113
} from './utils'
1214

@@ -60,16 +62,33 @@ export const transport: WebsocketReverseMappingTransport<WsTransportTypes, strin
6062
},
6163
},
6264
builders: {
63-
subscribeMessage: (params) => {
64-
const pair = `${params.base}${params.quote}`.toUpperCase()
65-
transport.setReverseMapping(pair, params)
66-
return buildBlocksizeWebsocketTickersMessage('fixedvwap_subscribe', pair)
65+
customSubscriptionMessages: (
66+
_context: EndpointContext<WsTransportTypes>,
67+
subscriptions: SubscriptionDeltas<{ quote: string; base: string }>,
68+
) => {
69+
const messages = []
70+
if (subscriptions.new.length > 0) {
71+
const pairsWithParams = subscriptions.new.map((param) => ({
72+
ticker: buildTicker(param),
73+
param: param,
74+
}))
75+
pairsWithParams.forEach(({ ticker, param }) => transport.setReverseMapping(ticker, param))
76+
messages.push(
77+
buildBlocksizeWebsocketTickersMessage(
78+
'fixedvwap_subscribe',
79+
subscriptions.new.map(buildTicker),
80+
),
81+
)
82+
}
83+
if (subscriptions.stale.length > 0) {
84+
messages.push(
85+
buildBlocksizeWebsocketTickersMessage(
86+
'fixedvwap_unsubscribe',
87+
subscriptions.stale.map(buildTicker),
88+
),
89+
)
90+
}
91+
return messages
6792
},
68-
unsubscribeMessage: (params) =>
69-
blocksizeDefaultUnsubscribeMessageBuilder(
70-
params.base,
71-
params.quote,
72-
'fixedvwap_unsubscribe',
73-
),
7493
},
7594
})

0 commit comments

Comments
 (0)