Stop sending undefine ws message + pass through raw endpoint name

Author: mxiao-cll

src/transports/websocket.ts

@@ -533,10 +533,14 @@ export class WebSocketTransport<
         await this.sendMessages(
           context,
           subscribeMessage
-            ? subscriptions.new.map((sub) => subscribeMessage(sub, context))
+            ? subscriptions.new
+                .map((sub) => subscribeMessage(sub, context))
+                .filter((m) => m !== undefined)
             : subscriptions.new,
           unsubscribeMessage
-            ? subscriptions.stale.map((sub) => unsubscribeMessage(sub, context))
+            ? subscriptions.stale
+                .map((sub) => unsubscribeMessage(sub, context))
+                .filter((m) => m !== undefined)
             : subscriptions.stale,
         )
         recordWsMessageSentMetrics(context, subscriptions.new, subscriptions.stale)

src/util/types.ts

@@ -32,6 +32,9 @@ export type AdapterRequestContext<T> = {
   /** Name of the endpoint this payload should be directed to */
   endpointName: string
 
+  /** Name of the endpoint from input before aliasing */
+  rawEndpointName: string
+
   /** Name of the transport this payload should be directed to */
   transportName: string
 

src/validation/index.ts

@@ -70,6 +70,7 @@ export const validatorMiddleware: AdapterMiddlewareBuilder =
       cacheKey: '',
       data: validatedData,
       endpointName: endpoint.name,
+      rawEndpointName: endpointParam,
     } as AdapterRequestContext<TypeFromDefinition<InputParametersDefinition>>
 
     // We do it afterwards so the custom routers can have a request with a requestContext fulfilled (sans transportName, ofc)
@@ -150,6 +151,7 @@ export const errorCatchingMiddleware = (err: Error, req: FastifyRequest, res: Fa
       cacheKey: errorLabel,
       data: undefined,
       endpointName: errorLabel,
+      rawEndpointName: errorLabel,
       transportName: errorLabel,
       meta: { error: err },
     }

test/transports/websocket.test.ts

@@ -10,6 +10,7 @@ import {
   WebSocketClassProvider,
   WebsocketReverseMappingTransport,
   WebSocketTransport,
+  type WebSocketTransportConfig,
 } from '../../src/transports'
 import { SingleNumberResultResponse, sleep } from '../../src/util'
 import { mockWebSocketProvider, runAllUntilTime, TestAdapter } from '../../src/util/testing-utils'
@@ -65,6 +66,7 @@ const createAdapter = (
     context: EndpointContext<WebSocketTypes>,
   ) => Promise<void> | void,
   pongHandler?: (connection: WebSocket, data: Buffer) => void,
+  buildersOverride?: WebSocketTransportConfig<WebSocketTypes>['builders'],
 ): Adapter => {
   const websocketTransport = new WebSocketTransport<WebSocketTypes>({
     url: () => ENDPOINT_URL,
@@ -99,7 +101,7 @@ const createAdapter = (
       heartbeat: heartbeatHandler,
       pong: pongHandler,
     },
-    builders: {
+    builders: buildersOverride ?? {
       subscribeMessage: (params) => `S:${params.base}/${params.quote}`,
       unsubscribeMessage: (params) => ({
         request: 'unsubscribe',
@@ -200,6 +202,81 @@ test.serial('connects to websocket, subscribes, gets message, unsubscribes', asy
   await t.context.clock.runToLastAsync()
 })
 
+test.serial('filters out undefined subscribe and unsubscribe messages from builders', async (t) => {
+  mockWebSocketProvider(WebSocketClassProvider)
+  const mockWsServer = new Server(ENDPOINT_URL, { mock: false })
+  const messagesReceived: string[] = []
+
+  mockWsServer.on('connection', (socket) => {
+    socket.on('message', (data) => {
+      messagesReceived.push(typeof data === 'string' ? data : data.toString())
+      socket.send(
+        JSON.stringify({
+          pair: `ETH/DOGE`,
+          value: price,
+        }),
+      )
+    })
+  })
+
+  const adapter = createAdapter(
+    {
+      WS_SUBSCRIPTION_UNRESPONSIVE_TTL: 180_000,
+      WS_SUBSCRIPTION_TTL: 60_000,
+    },
+    undefined,
+    undefined,
+    {
+      subscribeMessage: (params) => {
+        return params.base === 'SKIP' ? undefined : `S:${params.base}/${params.quote}`
+      },
+      unsubscribeMessage: (params) => {
+        return params.quote === 'SKIPUNSUB' ? undefined : `U:${params.base}/${params.quote}`
+      },
+    },
+  )
+
+  const testAdapter = await TestAdapter.startWithMockedCache(adapter, t.context)
+
+  // ETH first so the socket gets a subscribe + provider message before SKIP (no subscribe payload).
+  await testAdapter.request({ base: 'ETH', quote: 'DOGE' })
+  await runAllUntilTime(t.context.clock, BACKGROUND_EXECUTE_MS_WS + 200)
+  await testAdapter.request({ base: 'SKIP', quote: 'DOGE' })
+  await runAllUntilTime(t.context.clock, BACKGROUND_EXECUTE_MS_WS + 200)
+  await testAdapter.request({ base: 'ETH', quote: 'SKIPUNSUB' })
+  await runAllUntilTime(t.context.clock, BACKGROUND_EXECUTE_MS_WS + 200)
+
+  t.false(
+    messagesReceived.includes('S:SKIP/DOGE'),
+    'no subscribe payload for feeds where subscribeMessage returns undefined',
+  )
+  t.false(
+    messagesReceived.some((m) => m === 'undefined'),
+    'undefined must not be serialized and sent',
+  )
+  t.true(messagesReceived.includes('S:ETH/DOGE'))
+  t.true(messagesReceived.includes('S:ETH/SKIPUNSUB'))
+
+  await runAllUntilTime(
+    t.context.clock,
+    adapter.config.settings.WS_SUBSCRIPTION_TTL + BACKGROUND_EXECUTE_MS_WS * 3 + 100,
+  )
+
+  const unsubscribePayloads = messagesReceived.filter((m) => m.startsWith('U:'))
+  t.false(
+    unsubscribePayloads.some((m) => m === 'U:ETH/SKIPUNSUB'),
+    'unsubscribe builder returned undefined for SKIPUNSUB quote — must not send that payload',
+  )
+  t.true(
+    unsubscribePayloads.includes('U:ETH/DOGE'),
+    'defined unsubscribe payloads are still sent for stale subs that map to a message',
+  )
+
+  testAdapter.api.close()
+  mockWsServer.close()
+  await t.context.clock.runToLastAsync()
+})
+
 test.serial('reconnects when url changed', async (t) => {
   // Mock WS
   mockWebSocketProvider(WebSocketClassProvider)

test/validation.test.ts

@@ -837,6 +837,40 @@ test.serial('examples in input parameters', async (t) => {
   ])
 })
 
+test.serial('requestContext has correct endpointName and rawEndpointName', async (t) => {
+  let capturedContext: { endpointName: string; rawEndpointName: string } | undefined
+
+  const adapter = new Adapter({
+    name: 'TEST',
+    endpoints: [
+      new AdapterEndpoint({
+        name: 'test',
+        aliases: ['test-alias'],
+        transport: new (class extends NopTransport {
+          override async foregroundExecute(req: AdapterRequest<EmptyInputParameters>) {
+            capturedContext = {
+              endpointName: req.requestContext.endpointName,
+              rawEndpointName: req.requestContext.rawEndpointName,
+            }
+          }
+        })(),
+      }),
+    ],
+  })
+
+  const testAdapter = await TestAdapter.start(adapter, t.context)
+
+  // When using the canonical name, both fields should match
+  await testAdapter.request({ endpoint: 'test' })
+  t.is(capturedContext?.endpointName, 'test')
+  t.is(capturedContext?.rawEndpointName, 'test')
+
+  // When using an alias, endpointName is the canonical name and rawEndpointName is the alias
+  await testAdapter.request({ endpoint: 'test-alias' })
+  t.is(capturedContext?.endpointName, 'test')
+  t.is(capturedContext?.rawEndpointName, 'test-alias')
+})
+
 test.serial('limit size of input parameters', async (t) => {
   process.env['MAX_PAYLOAD_SIZE_LIMIT'] = '1048576'