examples/typescript/websocket-server-pool/ws-hybrid/src/websocket-server-worker.ts

   1 import { dirname, extname, join } from 'node:path'
   2 import { fileURLToPath } from 'node:url'
   3 import {
   4   ClusterWorker,
   5   DynamicThreadPool,
   6   availableParallelism
   7 } from 'poolifier'
   8 import { type RawData, WebSocketServer } from 'ws'
   9 import {
  10   type ClusterWorkerData,
  11   type ClusterWorkerResponse,
  12   type DataPayload,
  13   type MessagePayload,
  14   MessageType,
  15   type ThreadWorkerData,
  16   type ThreadWorkerResponse
  17 } from './types.js'
  18
  19 const emptyFunction = (): void => {
  20   /** Intentional */
  21 }
  22
  23 class WebSocketServerWorker extends ClusterWorker<
  24 ClusterWorkerData,
  25 ClusterWorkerResponse
  26 > {
  27   private static readonly startWebSocketServer = (
  28     workerData?: ClusterWorkerData
  29   ): ClusterWorkerResponse => {
  30     const { port } = workerData as ClusterWorkerData
  31     const wss = new WebSocketServer({ port }, () => {
  32       console.info(
  33         `⚡️[ws server]: WebSocket server is started in cluster worker at ws://localhost:${port}/`
  34       )
  35     })
  36
  37     wss.on('connection', ws => {
  38       ws.on('error', console.error)
  39       ws.on('message', (message: RawData) => {
  40         const { type, data } = JSON.parse(
  41           // eslint-disable-next-line @typescript-eslint/no-base-to-string
  42           message.toString()
  43         ) as MessagePayload<DataPayload>
  44         switch (type) {
  45           case MessageType.echo:
  46             this.requestHandlerPool
  47               .execute({ data }, 'echo')
  48               .then(response => {
  49                 ws.send(
  50                   JSON.stringify({
  51                     type: MessageType.echo,
  52                     data: response.data
  53                   })
  54                 )
  55                 return null
  56               })
  57               .catch(emptyFunction)
  58             break
  59           case MessageType.factorial:
  60             this.requestHandlerPool
  61               .execute({ data }, 'factorial')
  62               .then(response => {
  63                 ws.send(
  64                   JSON.stringify({
  65                     type: MessageType.factorial,
  66                     data: response.data
  67                   })
  68                 )
  69                 return null
  70               })
  71               .catch(emptyFunction)
  72             break
  73         }
  74       })
  75     })
  76     return {
  77       status: true,
  78       port: wss.options.port
  79     }
  80   }
  81
  82   private static readonly requestHandlerWorkerFile = join(
  83     dirname(fileURLToPath(import.meta.url)),
  84     `request-handler-worker${extname(fileURLToPath(import.meta.url))}`
  85   )
  86
  87   private static readonly requestHandlerPool = new DynamicThreadPool<
  88   ThreadWorkerData<DataPayload>,
  89   ThreadWorkerResponse<DataPayload>
  90   >(
  91     1,
  92     Math.round(availableParallelism() / 2),
  93     WebSocketServerWorker.requestHandlerWorkerFile,
  94     {
  95       enableTasksQueue: true,
  96       tasksQueueOptions: {
  97         concurrency: 8
  98       },
  99       errorHandler: (e: Error) => {
 100         console.error('Thread worker error:', e)
 101       }
 102     }
 103   )
 104
 105   public constructor () {
 106     super(WebSocketServerWorker.startWebSocketServer)
 107   }
 108 }
 109
 110 export const webSocketServerWorker = new WebSocketServerWorker()