chore: refactor worker & app factory

Author: cameri

src/@types/adapters.ts

@@ -1,13 +1,13 @@
 import { EventEmitter } from 'node:stream'
 import { SubscriptionFilter } from './subscription'
 
-export interface IWebSocketServerAdapter extends EventEmitter {
+export interface IWebSocketServerAdapter extends EventEmitter, IWebServerAdapter {
   getConnectedClients(): number
-  terminate(): Promise<void>
+  close(callback: () => void): void
 }
 
 export interface IWebServerAdapter extends EventEmitter {
-  listen(port: number)
+  listen(port: number): void
 }
 
 

src/adapters/web-server-adapter.ts

@@ -8,7 +8,7 @@ import { IWebServerAdapter } from '../@types/adapters'
 export class WebServerAdapter extends EventEmitter implements IWebServerAdapter {
 
   public constructor(
-    private readonly webServer: Server,
+    protected readonly webServer: Server,
     private readonly settings: () => ISettings,
   ) {
     super()

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

@@ -43,14 +43,10 @@ export class WebSocketServerAdapter extends WebServerAdapter implements IWebSock
     this.heartbeatInterval = setInterval(this.onHeartbeat.bind(this), WSS_CLIENT_HEALTH_PROBE_INTERVAL)
   }
 
-  public async terminate(): Promise<void> {
-    return void Promise.all(
-      [
-        ...Array.from(this.webSocketServer.clients).map((webSocket: WebSocket) =>
-          webSocket.terminate()
-        ),
-      ],
-    )
+  public close(callback: () => void): void {
+    this.webSocketServer.close(() => {
+      this.webServer.close(callback)
+    })
   }
 
   private onBroadcast(event: Event) {

src/app/app.ts

@@ -1,6 +1,5 @@
-import cluster, { Worker } from 'cluster'
+import { Cluster, Worker } from 'cluster'
 import { cpus } from 'os'
-import process from 'process'
 
 import { IRunnable } from '../@types/base'
 import { ISettings } from '../@types/settings'
@@ -9,31 +8,32 @@ import { Serializable } from 'child_process'
 
 export class App implements IRunnable {
   public constructor(
+    private readonly process: NodeJS.Process,
+    private readonly cluster: Cluster,
     private readonly settingsFactory: () => ISettings,
   ) {
-
-    cluster
+    this.cluster
       .on('message', this.onClusterMessage.bind(this))
       .on('exit', this.onClusterExit.bind(this))
 
-    process
+    this.process
       .on('SIGTERM', this.onExit.bind(this))
   }
 
   public run(): void {
     console.log(`${packageJson.name}@${packageJson.version}`)
     console.log(`supported NIPs: ${packageJson.supportedNips}`)
-    console.log(`primary ${process.pid} - running`)
+    console.log(`primary ${this.process.pid} - running`)
 
     const workerCount = this.settingsFactory().workers?.count || cpus().length
 
     for (let i = 0; i < workerCount; i++) {
-      cluster.fork()
+      this.cluster.fork()
     }
   }
 
   private onClusterMessage(source: Worker, message: Serializable) {
-    for (const worker of Object.values(cluster.workers)) {
+    for (const worker of Object.values(this.cluster.workers)) {
       if (source.id === worker.id) {
         continue
       }
@@ -47,17 +47,12 @@ export class App implements IRunnable {
       if (code === 0 || signal === 'SIGINT') {
         return
       }
-      const worker = cluster.fork()
-
-      const newPID = worker.process.pid
-      const oldPID = deadWorker.process.pid
 
-      console.log('worker ' + oldPID + ' died.')
-      console.log('worker ' + newPID + ' born.')
+      this.cluster.fork()
   }
 
   private onExit() {
     console.log('exiting...')
-    process.exit(0)
+    this.process.exit(0)
   }
 }

src/app/worker.ts

@@ -1,34 +1,13 @@
-import http from 'http'
-import process from 'process'
-import { WebSocketServer } from 'ws'
-
-import { createSettings } from '../factories/settings-factory'
-import { IEventRepository } from '../@types/repositories'
 import { IRunnable } from '../@types/base'
-import { webSocketAdapterFactory } from '../factories/websocket-adapter-factory'
-import { WebSocketServerAdapter } from '../adapters/web-socket-server-adapter'
+import { IWebSocketServerAdapter } from '../@types/adapters'
 
 export class AppWorker implements IRunnable {
-  private webServer: http.Server
-  private wss: WebSocketServer
-  private adapter: WebSocketServerAdapter
-
   public constructor(
-    private readonly eventRepository: IEventRepository
+    private readonly process: NodeJS.Process,
+    private readonly adapter: IWebSocketServerAdapter
   ) {
-    this.webServer = http.createServer()
-    this.wss = new WebSocketServer({ server: this.webServer, maxPayload: 1024 * 1024 })
-    this.adapter = new WebSocketServerAdapter(
-      this.webServer,
-      this.wss,
-      webSocketAdapterFactory(this.eventRepository),
-      createSettings,
-    )
-
     process
-      .on('message', (message: { eventName: string, event: unknown }) => {
-        this.adapter.emit(message.eventName, message.event)
-      })
+      .on('message', this.onMessage.bind(this))
       .on('SIGINT', this.onExit.bind(this))
       .on('SIGHUP', this.onExit.bind(this))
       .on('SIGTERM', this.onExit.bind(this))
@@ -44,20 +23,22 @@ export class AppWorker implements IRunnable {
     console.log(`worker ${process.pid} - listening on port ${port}`)
   }
 
+  private onMessage(message: { eventName: string, event: unknown }): void {
+    this.adapter.emit(message.eventName, message.event)
+  }
+
   private onError(error: Error) {
     console.error(`worker ${process.pid} - error`, error)
     throw error
   }
 
   private onExit() {
     console.log(`worker ${process.pid} - exiting`)
-    this.wss.close(() => {
-      this.webServer.close(() => {
-        // dbClient.destroy(() => {
-        //   process.exit(0)
-        // })
-        process.exit(0)
-      })
+    this.adapter.close(() => {
+      // dbClient.destroy(() => {
+      //   process.exit(0)
+      // })
+      process.exit(0)
     })
   }
 }

src/factories/app-factory.ts

@@ -1,6 +1,9 @@
+import cluster from 'cluster'
+import process from 'process'
+
 import { App } from '../app/app'
 import { SettingsStatic } from '../utils/settings'
 
 export const appFactory = () => {
-  return new App(SettingsStatic.createSettings)
+  return new App(process, cluster, SettingsStatic.createSettings)
 }

src/factories/worker-factory.ts

@@ -1,12 +1,29 @@
+import http from 'http'
+import process from 'process'
+import { WebSocketServer } from 'ws'
+
 import { AppWorker } from '../app/worker'
+import { createSettings } from '../factories/settings-factory'
 import { EventRepository } from '../repositories/event-repository'
 import { getDbClient } from '../database/client'
+import { webSocketAdapterFactory } from './websocket-adapter-factory'
+import { WebSocketServerAdapter } from '../adapters/web-socket-server-adapter'
 
 export const workerFactory = () => {
   const dbClient = getDbClient()
   const eventRepository = new EventRepository(dbClient)
 
-  return new AppWorker(
-    eventRepository
+  const server = http.createServer()
+  const webSocketServer = new WebSocketServer({
+    server,
+    maxPayload: 1024 * 1024,
+  })
+  const adapter = new WebSocketServerAdapter(
+    server,
+    webSocketServer,
+    webSocketAdapterFactory(eventRepository),
+    createSettings,
   )
+
+  return new AppWorker(process, adapter)
 }

src/index.ts

@@ -3,12 +3,12 @@ import cluster from 'cluster'
 import { appFactory } from './factories/app-factory'
 import { workerFactory } from './factories/worker-factory'
 
-export const run = (isPrimary: boolean) => {
+export const getRunner = (isPrimary: boolean) => {
   return (isPrimary)
-    ? appFactory
-    : workerFactory
+    ? appFactory()
+    : workerFactory()
 }
 
 if (require.main === module) {
-  run(cluster.isPrimary)()
+  getRunner(cluster.isPrimary).run()
 }