Skip to content

Commit ba8c923

Browse files
mxiao-cllmxiao-cll
committed
Add fallback request
1 parent 984c318 commit ba8c923

9 files changed

Lines changed: 252 additions & 2 deletions

File tree

‎docs/reference-tables/ea-settings.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -61,6 +61,7 @@
6161
| TLS_PASSPHRASE | string | | Password to be used to generate an encryption key | | |
6262
| TLS_PRIVATE_KEY | string | undefined | Base64 Private Key of TSL/SSL certificate | - Value must be a valid base64 string | |
6363
| TLS_PUBLIC_KEY | string | undefined | Base64 Public Key of TSL/SSL certificate | - Value must be a valid base64 string | |
64+
| TRANSPORT_FALLBACK_ENABLED | boolean | false | Flag to enable endpoint fallback transports when configured | | |
6465
| WARMUP_SUBSCRIPTION_TTL | number | 300000 | TTL for batch warmer subscriptions | - Value must be an integer<br> - Value must be above the minimum<br> - Value must be below the maximum | 0 | 3600000 |
6566
| WS_CONNECTION_OPEN_TIMEOUT | number | 10000 | The maximum amount of time in milliseconds to wait for the websocket connection to open (including custom open handler) | - Value must be an integer<br> - Value must be above the minimum<br> - Value must be below the maximum | 500 | 30000 |
6667
| WS_HEARTBEAT_INTERVAL_MS | number | 10000 | The number of ms between each hearbeat message that EA sends to server, only works if heartbeat handler is provided | - Value must be an integer<br> - Value must be above the minimum<br> - Value must be below the maximum | 5000 | 300000 |

‎src/adapter/basic.ts‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@ import {
1515
highestRateLimitTiers,
1616
} from '../rate-limiting'
1717
import { RateLimiterFactory, RateLimitingStrategy } from '../rate-limiting/factory'
18+
import { Transport } from '../transports'
1819
import {
1920
AdapterRequest,
2021
AdapterResponse,
@@ -449,6 +450,14 @@ export class Adapter<
449450
const endpoint = this.endpointsMap[req.requestContext.endpointName]
450451
const transport = endpoint.transportRoutes.get(req.requestContext.transportName)
451452

453+
return this.handleSingleTransportRequest(req, replySent, transport)
454+
}
455+
456+
private async handleSingleTransportRequest(
457+
req: AdapterRequest<EmptyInputParameters>,
458+
replySent: Promise<unknown>,
459+
transport: Transport<EndpointGenerics>,
460+
): Promise<Readonly<AdapterResponse>> {
452461
// First try to find the response in our cache, keep it ready
453462
const cachedResponse = await this.findResponseInCache(req)
454463

‎src/adapter/endpoint.ts‎

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -46,6 +46,7 @@ export class AdapterEndpoint<T extends EndpointGenerics> implements AdapterEndpo
4646
settings: T['Settings'],
4747
) => string
4848
defaultTransport?: string
49+
fallbackTransport?: string
4950

5051
constructor(params: AdapterEndpointParams<T>) {
5152
this.name = params.name
@@ -55,6 +56,7 @@ export class AdapterEndpoint<T extends EndpointGenerics> implements AdapterEndpo
5556
this.transportRoutes = params.transportRoutes
5657
this.customRouter = params.customRouter
5758
this.defaultTransport = params.defaultTransport
59+
this.fallbackTransport = params.fallbackTransport?.toLowerCase()
5860
} else {
5961
this.transportRoutes = new TransportRoutes<T>().register(
6062
DEFAULT_TRANSPORT_NAME,
@@ -186,6 +188,34 @@ export class AdapterEndpoint<T extends EndpointGenerics> implements AdapterEndpo
186188
return transportName
187189
}
188190

191+
getFallbackTransportNameForRequest(primaryTransportName: string, settings: T['Settings']) {
192+
if (!settings.TRANSPORT_FALLBACK_ENABLED || !this.fallbackTransport) {
193+
logger.trace('TRANSPORT_FALLBACK_ENABLED is false or fallbackTransport is not set')
194+
return
195+
}
196+
197+
const fallbackTransportName = this.fallbackTransport.toLowerCase()
198+
199+
if (!this.transportRoutes.get(fallbackTransportName)) {
200+
throw new AdapterError({
201+
statusCode: 400,
202+
message: `No fallback transport found for key "${fallbackTransportName}", must be one of ${JSON.stringify(
203+
this.transportRoutes.routeNames(),
204+
)}`,
205+
})
206+
}
207+
208+
if (fallbackTransportName === primaryTransportName.toLowerCase()) {
209+
throw new AdapterError({
210+
statusCode: 400,
211+
message: `Fallback transport "${fallbackTransportName}" cannot be the same as primary transport.`,
212+
})
213+
}
214+
215+
logger.debug(`Request can fall back to transport "${fallbackTransportName}"`)
216+
return fallbackTransportName
217+
}
218+
189219
/**
190220
* Default routing strategy. Will try and use the transport override if present
191221
* or transport input parameter in the request body.

‎src/adapter/types.ts‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -183,6 +183,9 @@ type MultiTransportAdapterEndpointParams<T extends EndpointGenerics> = {
183183

184184
/** If no value is returned from the custom router or the default (transport param), which transport to use */
185185
defaultTransport?: string
186+
187+
/** Backup transport to use when the primary transport is unable to return data */
188+
fallbackTransport?: string
186189
}
187190

188191
/**

‎src/config/index.ts‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -276,6 +276,11 @@ export const BaseSettingsDefinition = {
276276
default: 200,
277277
validate: validator.integer({ min: 10, max: 1000 }),
278278
},
279+
TRANSPORT_FALLBACK_ENABLED: {
280+
description: 'Flag to enable endpoint fallback transports when configured',
281+
type: 'boolean',
282+
default: false,
283+
},
279284
DEFAULT_CACHE_KEY: {
280285
description: 'Default key to be used when one cannot be determined from request parameters',
281286
type: 'string',

‎src/util/types.ts‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,12 @@ export type AdapterRequestContext<T> = {
3838
/** Precalculated cache key used to get and set corresponding values from the cache and subscription sets */
3939
cacheKey: string
4040

41+
/** Fallback transport context to use if the primary transport is unable to return data */
42+
fallback?: {
43+
transportName: string
44+
cacheKey: string
45+
}
46+
4147
/** Normalized and validated data coming from the request body */
4248
data: T
4349

‎src/validation/index.ts‎

Lines changed: 24 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -77,6 +77,10 @@ export const validatorMiddleware: AdapterMiddlewareBuilder =
7777
req,
7878
adapter.config.settings,
7979
)
80+
const fallbackTransportName = endpoint.getFallbackTransportNameForRequest(
81+
req.requestContext.transportName,
82+
adapter.config.settings,
83+
)
8084

8185
// Custom input validation defined in the EA
8286
const error =
@@ -108,6 +112,12 @@ export const validatorMiddleware: AdapterMiddlewareBuilder =
108112

109113
// Now that all the transformations have been applied, all that's left is calculating the cache key
110114
if (endpoint.cacheKeyGenerator) {
115+
if (fallbackTransportName) {
116+
errorCatcherLogger.warn(
117+
'Fallback transports ignored for endpoints with custom cache key generators',
118+
)
119+
}
120+
111121
let cacheKey
112122
cacheKey = endpoint.cacheKeyGenerator(req.requestContext.data)
113123
if (cacheKey.length > adapter.config.settings.MAX_COMMON_KEY_SIZE) {
@@ -125,14 +135,26 @@ export const validatorMiddleware: AdapterMiddlewareBuilder =
125135

126136
req.requestContext.cacheKey = `${cachePrefix}${cacheKey}`
127137
} else {
128-
const transportName = req.requestContext.transportName
129138
req.requestContext.cacheKey = calculateCacheKey({
130139
data: req.requestContext.data,
131140
adapterName: adapter.name,
132141
endpointName: endpoint.name,
133-
transportName,
142+
transportName: req.requestContext.transportName,
134143
adapterSettings: adapter.config.settings,
135144
})
145+
146+
if (fallbackTransportName) {
147+
req.requestContext.fallback = {
148+
transportName: fallbackTransportName,
149+
cacheKey: calculateCacheKey({
150+
data: req.requestContext.data,
151+
adapterName: adapter.name,
152+
endpointName: endpoint.name,
153+
transportName: fallbackTransportName,
154+
adapterSettings: adapter.config.settings,
155+
}),
156+
}
157+
}
136158
}
137159

138160
done()

‎test/cache/cache-key.test.ts‎

Lines changed: 105 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@ import untypedTest, { TestFn } from 'ava'
22
import { Adapter, AdapterEndpoint, EndpointGenerics } from '../../src/adapter'
33
import { Cache, calculateCacheKey } from '../../src/cache'
44
import { AdapterConfig, BaseAdapterSettings, BaseSettingsDefinition } from '../../src/config'
5+
import { TransportRoutes } from '../../src/transports'
56
import { AdapterRequest, AdapterResponse } from '../../src/util'
67
import { NopTransport, NopTransportTypes, TestAdapter } from '../../src/util/testing-utils'
78
import { InputParameters } from '../../src/validation'
@@ -172,6 +173,110 @@ test.serial('custom cache key', async (t) => {
172173
t.is(response.json().result, 'test:custom_cache_key')
173174
})
174175

176+
test.serial('builds fallback cache key from fallback transport name', async (t) => {
177+
const config = new AdapterConfig(
178+
{},
179+
{
180+
envDefaultOverrides: {
181+
TRANSPORT_FALLBACK_ENABLED: true,
182+
},
183+
},
184+
)
185+
const adapter = new Adapter({
186+
name: 'TEST',
187+
defaultEndpoint: 'test-fallback-cache-key',
188+
config,
189+
endpoints: [
190+
new AdapterEndpoint<NopTransportTypes>({
191+
name: 'test-fallback-cache-key',
192+
transportRoutes: new TransportRoutes<NopTransportTypes>()
193+
.register(
194+
'primary',
195+
new (class extends NopTransport {
196+
override async foregroundExecute(
197+
req: AdapterRequest<NopTransportTypes['Parameters']>,
198+
) {
199+
return {
200+
data: null,
201+
statusCode: 200,
202+
result: req.requestContext.fallback?.cacheKey as unknown as null,
203+
timestamps: {
204+
providerDataRequestedUnixMs: 0,
205+
providerDataReceivedUnixMs: 0,
206+
providerIndicatedTimeUnixMs: undefined,
207+
},
208+
}
209+
}
210+
})(),
211+
)
212+
.register('fallback', new NopTransport()),
213+
defaultTransport: 'primary',
214+
fallbackTransport: 'fallback',
215+
}),
216+
],
217+
})
218+
const testAdapter = await TestAdapter.start(adapter, t.context)
219+
220+
const response = await testAdapter.request({})
221+
222+
t.is(
223+
response.json().result,
224+
`TEST-test-fallback-cache-key-fallback-${BaseSettingsDefinition.DEFAULT_CACHE_KEY.default}`,
225+
)
226+
})
227+
228+
test.serial('custom cache key generator ignores fallback transport', async (t) => {
229+
const config = new AdapterConfig(
230+
{},
231+
{
232+
envDefaultOverrides: {
233+
TRANSPORT_FALLBACK_ENABLED: true,
234+
},
235+
},
236+
)
237+
const adapter = new Adapter({
238+
name: 'TEST',
239+
defaultEndpoint: 'test-custom-cache-key-fallback',
240+
config,
241+
endpoints: [
242+
new AdapterEndpoint<NopTransportTypes>({
243+
name: 'test-custom-cache-key-fallback',
244+
cacheKeyGenerator: (_) => `test:custom_cache_key`,
245+
transportRoutes: new TransportRoutes<NopTransportTypes>()
246+
.register(
247+
'primary',
248+
new (class extends NopTransport {
249+
override async foregroundExecute(
250+
req: AdapterRequest<NopTransportTypes['Parameters']>,
251+
) {
252+
return {
253+
data: null,
254+
statusCode: 200,
255+
result: (req.requestContext.fallback?.cacheKey ||
256+
req.requestContext.cacheKey) as unknown as null,
257+
timestamps: {
258+
providerDataRequestedUnixMs: 0,
259+
providerDataReceivedUnixMs: 0,
260+
providerIndicatedTimeUnixMs: undefined,
261+
},
262+
}
263+
}
264+
})(),
265+
)
266+
.register('fallback', new NopTransport()),
267+
defaultTransport: 'primary',
268+
fallbackTransport: 'fallback',
269+
}),
270+
],
271+
})
272+
const testAdapter = await TestAdapter.start(adapter, t.context)
273+
274+
const response = await testAdapter.request({})
275+
276+
t.is(response.statusCode, 200)
277+
t.is(response.json().result, 'test:custom_cache_key')
278+
})
279+
175280
test.serial('custom cache key is truncated if over max size', async (t) => {
176281
const response = await t.context.testAdapter.request({
177282
endpoint: 'test-custom-cache-key-long',

‎test/transports/routing.test.ts‎

Lines changed: 69 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -680,6 +680,75 @@ test.serial('transports with same name throws error', async (t) => {
680680
)
681681
})
682682

683+
test.serial('fallback transport is ignored when disabled', (t) => {
684+
const endpoint = new AdapterEndpoint<BaseEndpointTypes>({
685+
inputParameters,
686+
name: 'price',
687+
transportRoutes: transports,
688+
fallbackTransport: 'invalid',
689+
})
690+
const config = new AdapterConfig(settings)
691+
config.initialize()
692+
693+
t.is(endpoint.getFallbackTransportNameForRequest('batch', config.settings), undefined)
694+
})
695+
696+
test.serial('fallback transport is normalized when enabled', (t) => {
697+
const endpoint = new AdapterEndpoint<BaseEndpointTypes>({
698+
inputParameters,
699+
name: 'price',
700+
transportRoutes: transports,
701+
fallbackTransport: 'WEBSOCKET',
702+
})
703+
const config = new AdapterConfig(settings, {
704+
envDefaultOverrides: {
705+
TRANSPORT_FALLBACK_ENABLED: true,
706+
},
707+
})
708+
config.initialize()
709+
710+
t.is(endpoint.getFallbackTransportNameForRequest('batch', config.settings), 'websocket')
711+
})
712+
713+
test.serial('fallback transport must be registered when enabled', (t) => {
714+
const endpoint = new AdapterEndpoint<BaseEndpointTypes>({
715+
inputParameters,
716+
name: 'price',
717+
transportRoutes: transports,
718+
fallbackTransport: 'invalid',
719+
})
720+
const config = new AdapterConfig(settings, {
721+
envDefaultOverrides: {
722+
TRANSPORT_FALLBACK_ENABLED: true,
723+
},
724+
})
725+
config.initialize()
726+
727+
t.throws(() => endpoint.getFallbackTransportNameForRequest('batch', config.settings), {
728+
message:
729+
'No fallback transport found for key "invalid", must be one of ["websocket","batch","sse"]',
730+
})
731+
})
732+
733+
test.serial('fallback transport cannot match primary transport', (t) => {
734+
const endpoint = new AdapterEndpoint<BaseEndpointTypes>({
735+
inputParameters,
736+
name: 'price',
737+
transportRoutes: transports,
738+
fallbackTransport: 'batch',
739+
})
740+
const config = new AdapterConfig(settings, {
741+
envDefaultOverrides: {
742+
TRANSPORT_FALLBACK_ENABLED: true,
743+
},
744+
})
745+
config.initialize()
746+
747+
t.throws(() => endpoint.getFallbackTransportNameForRequest('batch', config.settings), {
748+
message: 'Fallback transport "batch" cannot be the same as primary transport.',
749+
})
750+
})
751+
683752
test.serial('transport override routes to correct Transport', async (t) => {
684753
axiosMock
685754
.onPost(`${restUrl}/price`, {

0 commit comments

Comments
 (0)