3763964114

Build Type

github

Committed by GitHub

Commit Message

Merge b2fcbdf3d into ab86a6565

Pull Request Pull Request #96: feat: merkle tree and proofs

Run Details

310 of 393 branches covered (78.88%)

Branch coverage included in aggregate %.

6 of 14 new or added lines in 2 files covered. (42.86%)

76 existing lines in 8 files now uncovered.

749 of 940 relevant lines covered (79.68%)

7.76 hits per line

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

59.52

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

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

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 { ISettings } from '../@types/settings'
import { propEq } from 'ramda'
import { WebServerAdapter } from './web-server-adapter'

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

const WSS_CLIENT_HEALTH_PROBE_INTERVAL = 30000

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

  private heartbeatInterval: NodeJS.Timer

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

    this.webSocketsAdapters = new WeakMap()

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

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

  public close(callback: () => void): void {
    this.onClose()
    this.webSocketServer.close(() => {
      this.webServer.close(callback)
    })
  }

  private onBroadcast(event: Event) {
    this.webSocketServer.clients.forEach((webSocket: WebSocket) => {
      if (!propEq('readyState', OPEN)(webSocket)) {
        return
      }
      const webSocketAdapter = this.webSocketsAdapters.get(webSocket)
      debug('broadcasting event to client %s: %o', webSocketAdapter.getClientId(), event)
      webSocketAdapter.emit(WebSocketAdapterEvent.Event, event)
    })
  }

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

  private onConnection(client: WebSocket, req: IncomingMessage) {
    debug('client connected: %o', req.headers)
    this.webSocketsAdapters.set(client, this.createWebSocketAdapter([client, req, this]))
  }

  private onHeartbeat() {
    this.webSocketServer.clients.forEach((webSocket) =>
      this.webSocketsAdapters.get(webSocket).emit(WebSocketAdapterEvent.Heartbeat)
    )
  }

  protected onClose() {
    debug('closing')
    clearInterval(this.heartbeatInterval)
    this.webSocketServer.clients.forEach((webSocket: WebSocket) => {
      debug('terminating client')
      webSocket.terminate()
    })
    this.removeAllListeners()
    this.webSocketServer.removeAllListeners()
    super.onClose()
    debug('closed')
  }
}

1	import { IncomingMessage, Server } from 'http'
2	import WebSocket, { OPEN, WebSocketServer } from 'ws'	1✔
3
4	import { IWebSocketAdapter, IWebSocketServerAdapter } from '../@types/adapters'
5	import { WebSocketAdapterEvent, WebSocketServerAdapterEvent } from '../constants/adapter'	1✔
6	import { createLogger } from '../factories/logger-factory'	1✔
7	import { Event } from '../@types/event'
8	import { Factory } from '../@types/base'
9	import { ISettings } from '../@types/settings'
10	import { propEq } from 'ramda'	1✔
11	import { WebServerAdapter } from './web-server-adapter'	1✔
12
13	const debug = createLogger('web-socket-server-adapter')	1✔
14
15	const WSS_CLIENT_HEALTH_PROBE_INTERVAL = 30000	1✔
16
17	export class WebSocketServerAdapter extends WebServerAdapter implements IWebSocketServerAdapter {	1✔
18	private webSocketsAdapters: WeakMap<WebSocket, IWebSocketAdapter>
19
20	private heartbeatInterval: NodeJS.Timer
21
22	public constructor(
23	webServer: Server,
24	private readonly webSocketServer: WebSocketServer,	1✔
25	private readonly createWebSocketAdapter: Factory<	1✔
26	IWebSocketAdapter,
27	[WebSocket, IncomingMessage, IWebSocketServerAdapter]
28	>,
29	settings: () => ISettings,
30	) {
31	super(webServer, settings)	1✔
32
33	this.webSocketsAdapters = new WeakMap()	1✔
34
35	this	1✔
36	.on(WebSocketServerAdapterEvent.Broadcast, this.onBroadcast.bind(this))
37
38	this.webSocketServer	1✔
39	.on(WebSocketServerAdapterEvent.Close, this.onClose.bind(this))
40	.on(WebSocketServerAdapterEvent.Connection, this.onConnection.bind(this))
41	.on('error', (error) => {
42	debug('error: %o', error)	×
43	throw error	×
44	})
45	this.heartbeatInterval = setInterval(this.onHeartbeat.bind(this), WSS_CLIENT_HEALTH_PROBE_INTERVAL)	1✔
46	}
47
48	public close(callback: () => void): void {
49	this.onClose()	1✔
50	this.webSocketServer.close(() => {	1✔
51	this.webServer.close(callback)	1✔
52	})
53	}
54
55	private onBroadcast(event: Event) {
UNCOV 56	this.webSocketServer.clients.forEach((webSocket: WebSocket) => {	×
UNCOV 57	if (!propEq('readyState', OPEN)(webSocket)) {	×
58	return	×
59	}
UNCOV 60	const webSocketAdapter = this.webSocketsAdapters.get(webSocket)	×
UNCOV 61	debug('broadcasting event to client %s: %o', webSocketAdapter.getClientId(), event)	×
UNCOV 62	webSocketAdapter.emit(WebSocketAdapterEvent.Event, event)	×
63	})
64	}
65
66	public getConnectedClients(): number {
67	return Array.from(this.webSocketServer.clients).filter(propEq('readyState', OPEN)).length	×
68	}
69
70	private onConnection(client: WebSocket, req: IncomingMessage) {
UNCOV 71	debug('client connected: %o', req.headers)	×
UNCOV 72	this.webSocketsAdapters.set(client, this.createWebSocketAdapter([client, req, this]))	×
73	}
74
75	private onHeartbeat() {
76	this.webSocketServer.clients.forEach((webSocket) =>	×
77	this.webSocketsAdapters.get(webSocket).emit(WebSocketAdapterEvent.Heartbeat)	×
78	)
79	}
80
81	protected onClose() {
82	debug('closing')	1✔
83	clearInterval(this.heartbeatInterval)	1✔
84	this.webSocketServer.clients.forEach((webSocket: WebSocket) => {	1✔
85	debug('terminating client')	×
86	webSocket.terminate()	×
87	})
88	this.removeAllListeners()	1✔
89	this.webSocketServer.removeAllListeners()	1✔
90	super.onClose()	1✔
91	debug('closed')	1✔
92	}
93	}

Cameri / nostr-ts-relay / 3763964114

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