26382712021

Committed 25 May 2026 04:09AM UTC coverage: 62.715% (-2.4%) from 65.119%

Build # 26382712021

Build Type

Pull #627

github

Committed by

web-flow

Commit Message

Merge 96e29866f into 54139ed49

Pull Request Pull Request #627: test(integration): verify response content-types across HTTP routes

Coverage Stats

1757 of 3168 branches covered (55.46%)

Branch coverage included in aggregate %.

4036 of 6069 relevant lines covered (66.5%)

7.99 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

91.04

/src/adapters/web-socket-server-adapter.ts

import { IncomingMessage, Server } from 'http'
import WebSocket, { OPEN, WebSocketServer } from 'ws'
import { propEq } from 'ramda'

import { IWebSocketAdapter, IWebSocketServerAdapter } from '../@types/adapters'
import { WebSocketAdapterEvent, WebSocketServerAdapterEvent } from '../constants/adapter'
import { createLogger } from '../factories/logger-factory'
import { Event } from '../@types/event'
import { Factory } from '../@types/base'
import { getRemoteAddress } from '../utils/http'
import { isRateLimited } from '../handlers/request-handlers/rate-limiter-middleware'
import { Settings } from '../@types/settings'
import { WebServerAdapter } from './web-server-adapter'

const logger = createLogger('web-socket-server-adapter')

const WSS_CLIENT_HEALTH_PROBE_INTERVAL = 120000

export class WebSocketServerAdapter extends WebServerAdapter implements IWebSocketServerAdapter {
  private webSocketsAdapters: WeakMap<WebSocket, IWebSocketAdapter>

  private heartbeatInterval: NodeJS.Timeout

  public constructor(
    webServer: Server,
    private readonly webSocketServer: WebSocketServer,
    private readonly createWebSocketAdapter: Factory<
      IWebSocketAdapter,
      [WebSocket, IncomingMessage, IWebSocketServerAdapter]
    >,
    private readonly settings: () => Settings,
  ) {
    logger('created')
    super(webServer)

    this.webSocketsAdapters = new WeakMap()

    this.on(WebSocketServerAdapterEvent.Broadcast, this.onBroadcast.bind(this))

    this.webSocketServer
      .on(WebSocketServerAdapterEvent.Connection, this.onConnection.bind(this))
      .on('error', (error) => {
        logger('error: %o', error)
      })
    this.heartbeatInterval = setInterval(this.onHeartbeat.bind(this), WSS_CLIENT_HEALTH_PROBE_INTERVAL)
  }

  public close(callback?: () => void): void {
    super.close(() => {
      logger('closing')
      clearInterval(this.heartbeatInterval)
      this.webSocketServer.clients.forEach((webSocket: WebSocket) => {
        const webSocketAdapter = this.webSocketsAdapters.get(webSocket)
        if (webSocketAdapter) {
          logger('terminating client %s: %s', webSocketAdapter.getClientId(), webSocketAdapter.getClientAddress())
        }
        webSocket.terminate()
      })
      logger('closing web socket server')
      this.webSocketServer.close(() => {
        this.webSocketServer.removeAllListeners()
        if (typeof callback !== 'undefined') {
          callback()
        }
        logger('closed')
      })
    })
    this.removeAllListeners()
  }

  private onBroadcast(event: Event) {
    this.webSocketServer.clients.forEach((webSocket: WebSocket) => {
      if (!propEq('readyState', OPEN)(webSocket)) {
        return
      }
      const webSocketAdapter = this.webSocketsAdapters.get(webSocket) as IWebSocketAdapter
      if (!webSocketAdapter) {
        return
      }
      webSocketAdapter.emit(WebSocketAdapterEvent.Event, event)
    })
  }

  public getConnectedClients(): number {
    return Array.from(this.webSocketServer.clients).filter(propEq('readyState', OPEN)).length
  }

  private async onConnection(client: WebSocket, req: IncomingMessage) {
    const currentSettings = this.settings()
    const remoteAddress = getRemoteAddress(req, currentSettings)

    logger('client %s connected: %o', remoteAddress, req.headers)

    if (await isRateLimited(remoteAddress, currentSettings)) {
      logger('client %s terminated: rate-limited', remoteAddress)
      client.terminate()
      return
    }

    this.webSocketsAdapters.set(client, this.createWebSocketAdapter([client, req, this]))
  }

  private onHeartbeat() {
    this.webSocketServer.clients.forEach((webSocket) => {
      const webSocketAdapter = this.webSocketsAdapters.get(webSocket) as IWebSocketAdapter
      if (webSocketAdapter) {
        webSocketAdapter.emit(WebSocketAdapterEvent.Heartbeat)
      }
    })
  }
}

1	import { IncomingMessage, Server } from 'http'
2	import WebSocket, { OPEN, WebSocketServer } from 'ws'	1✔
3	import { propEq } from 'ramda'	1✔
4
5	import { IWebSocketAdapter, IWebSocketServerAdapter } from '../@types/adapters'
6	import { WebSocketAdapterEvent, WebSocketServerAdapterEvent } from '../constants/adapter'	1✔
7	import { createLogger } from '../factories/logger-factory'	1✔
8	import { Event } from '../@types/event'
9	import { Factory } from '../@types/base'
10	import { getRemoteAddress } from '../utils/http'	1✔
11	import { isRateLimited } from '../handlers/request-handlers/rate-limiter-middleware'	1✔
12	import { Settings } from '../@types/settings'
13	import { WebServerAdapter } from './web-server-adapter'	1✔
14
15	const logger = createLogger('web-socket-server-adapter')	1✔
16
17	const WSS_CLIENT_HEALTH_PROBE_INTERVAL = 120000	1✔
18
19	export class WebSocketServerAdapter extends WebServerAdapter implements IWebSocketServerAdapter {	1✔
20	private webSocketsAdapters: WeakMap<WebSocket, IWebSocketAdapter>
21
22	private heartbeatInterval: NodeJS.Timeout
23
24	public constructor(
25	webServer: Server,
26	private readonly webSocketServer: WebSocketServer,	16✔
27	private readonly createWebSocketAdapter: Factory<	16✔
28	IWebSocketAdapter,
29	[WebSocket, IncomingMessage, IWebSocketServerAdapter]
30	>,
31	private readonly settings: () => Settings,	16✔
32	) {
33	logger('created')	16✔
34	super(webServer)	16✔
35
36	this.webSocketsAdapters = new WeakMap()	16✔
37
38	this.on(WebSocketServerAdapterEvent.Broadcast, this.onBroadcast.bind(this))	16✔
39
40	this.webSocketServer	16✔
41	.on(WebSocketServerAdapterEvent.Connection, this.onConnection.bind(this))
42	.on('error', (error) => {
43	logger('error: %o', error)	×
44	})
45	this.heartbeatInterval = setInterval(this.onHeartbeat.bind(this), WSS_CLIENT_HEALTH_PROBE_INTERVAL)	16✔
46	}
47
48	public close(callback?: () => void): void {
49	super.close(() => {	21✔
50	logger('closing')	20✔
51	clearInterval(this.heartbeatInterval)	20✔
52	this.webSocketServer.clients.forEach((webSocket: WebSocket) => {	20✔
53	const webSocketAdapter = this.webSocketsAdapters.get(webSocket)	2✔
54	if (webSocketAdapter) {	2!
55	logger('terminating client %s: %s', webSocketAdapter.getClientId(), webSocketAdapter.getClientAddress())	×
56	}
57	webSocket.terminate()	2✔
58	})
59	logger('closing web socket server')	20✔
60	this.webSocketServer.close(() => {	20✔
61	this.webSocketServer.removeAllListeners()	20✔
62	if (typeof callback !== 'undefined') {	20✔
63	callback()	2✔
64	}
65	logger('closed')	20✔
66	})
67	})
68	this.removeAllListeners()	21✔
69	}
70
71	private onBroadcast(event: Event) {
72	this.webSocketServer.clients.forEach((webSocket: WebSocket) => {	2✔
73	if (!propEq('readyState', OPEN)(webSocket)) {	2✔
74	return	1✔
75	}
76	const webSocketAdapter = this.webSocketsAdapters.get(webSocket) as IWebSocketAdapter	1✔
77	if (!webSocketAdapter) {	1!
78	return	×
79	}
80	webSocketAdapter.emit(WebSocketAdapterEvent.Event, event)	1✔
81	})
82	}
83
84	public getConnectedClients(): number {
85	return Array.from(this.webSocketServer.clients).filter(propEq('readyState', OPEN)).length	2✔
86	}
87
88	private async onConnection(client: WebSocket, req: IncomingMessage) {
89	const currentSettings = this.settings()	4✔
90	const remoteAddress = getRemoteAddress(req, currentSettings)	4✔
91
92	logger('client %s connected: %o', remoteAddress, req.headers)	4✔
93
94	if (await isRateLimited(remoteAddress, currentSettings)) {	4✔
95	logger('client %s terminated: rate-limited', remoteAddress)	1✔
96	client.terminate()	1✔
97	return	1✔
98	}
99
100	this.webSocketsAdapters.set(client, this.createWebSocketAdapter([client, req, this]))	3✔
101	}
102
103	private onHeartbeat() {
104	this.webSocketServer.clients.forEach((webSocket) => {	1✔
105	const webSocketAdapter = this.webSocketsAdapters.get(webSocket) as IWebSocketAdapter	1✔
106	if (webSocketAdapter) {	1!
107	webSocketAdapter.emit(WebSocketAdapterEvent.Heartbeat)	1✔
108	}
109	})
110	}
111	}

cameri / nostream / 26382712021

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous