examples/typescript/http-server-pool/express-hybrid/src/express-worker.ts

   1 import type { Server } from 'node:http'
   2 import type { AddressInfo } from 'node:net'
   3
   4 import express, { type Express, type Request, type Response } from 'express'
   5 import {
   6   availableParallelism,
   7   ClusterWorker,
   8   DynamicThreadPool
   9 } from 'poolifier'
  10
  11 import {
  12   type ClusterWorkerData,
  13   type ClusterWorkerResponse,
  14   type DataPayload,
  15   type ThreadWorkerData,
  16   type ThreadWorkerResponse
  17 } from './types.js'
  18
  19 const emptyFunction = (): void => {
  20   /* Intentional */
  21 }
  22
  23 class ExpressWorker extends ClusterWorker<
  24 ClusterWorkerData,
  25 ClusterWorkerResponse
  26 > {
  27   private static server: Server
  28   private static requestHandlerPool: DynamicThreadPool<
  29   ThreadWorkerData<DataPayload>,
  30   ThreadWorkerResponse<DataPayload>
  31   >
  32
  33   private static readonly startExpress = (
  34     workerData?: ClusterWorkerData
  35   ): ClusterWorkerResponse => {
  36     const { port, workerFile, minWorkers, maxWorkers, ...poolOptions } =
  37       workerData!
  38
  39     ExpressWorker.requestHandlerPool = new DynamicThreadPool<
  40     ThreadWorkerData<DataPayload>,
  41     ThreadWorkerResponse<DataPayload>
  42     >(
  43       minWorkers ?? 1,
  44       maxWorkers ?? availableParallelism(),
  45       workerFile,
  46       poolOptions
  47     )
  48
  49     const application: Express = express()
  50
  51     // Parse only JSON requests body
  52     application.use(express.json())
  53
  54     application.all('/api/echo', (req: Request, res: Response) => {
  55       ExpressWorker.requestHandlerPool
  56         .execute({ data: req.body }, 'echo')
  57         .then(response => {
  58           return res.send(response.data).end()
  59         })
  60         .catch(emptyFunction)
  61     })
  62
  63     application.get('/api/factorial/:number', (req: Request, res: Response) => {
  64       const { number } = req.params
  65       ExpressWorker.requestHandlerPool
  66         .execute({ data: { number: parseInt(number) } }, 'factorial')
  67         .then(response => {
  68           return res.send(response.data).end()
  69         })
  70         .catch(emptyFunction)
  71     })
  72
  73     let listenerPort: number | undefined
  74     ExpressWorker.server = application.listen(port, () => {
  75       listenerPort = (ExpressWorker.server.address() as AddressInfo).port
  76       console.info(
  77         `⚡️[express server]: Express server is started in cluster worker at http://localhost:${listenerPort}/`
  78       )
  79     })
  80     return {
  81       status: true,
  82       port: listenerPort ?? port
  83     }
  84   }
  85
  86   public constructor () {
  87     super(ExpressWorker.startExpress, {
  88       killHandler: async () => {
  89         await ExpressWorker.requestHandlerPool.destroy()
  90         ExpressWorker.server.close()
  91       }
  92     })
  93   }
  94 }
  95
  96 export const expressWorker = new ExpressWorker()