packages/peertube-runner/shared/ipc/ipc-server.ts

   1 import { ensureDir } from 'fs-extra'
   2 import { Server as NetIPC } from 'net-ipc'
   3 import { pick } from '@shared/core-utils'
   4 import { RunnerServer } from '../../server'
   5 import { ConfigManager } from '../config-manager'
   6 import { logger } from '../logger'
   7 import { IPCReponse, IPCReponseData, IPCRequest } from './shared'
   8
   9 export class IPCServer {
  10   private netIPC: NetIPC
  11   private runnerServer: RunnerServer
  12
  13   async run (runnerServer: RunnerServer) {
  14     this.runnerServer = runnerServer
  15
  16     await ensureDir(ConfigManager.Instance.getSocketDirectory())
  17
  18     const socketPath = ConfigManager.Instance.getSocketPath()
  19     this.netIPC = new NetIPC({ path: socketPath })
  20     await this.netIPC.start()
  21
  22     logger.info(`IPC socket created on ${socketPath}`)
  23
  24     this.netIPC.on('request', async (req: IPCRequest, res) => {
  25       try {
  26         const data = await this.process(req)
  27
  28         this.sendReponse(res, { success: true, data })
  29       } catch (err) {
  30         logger.error('Cannot execute RPC call', err)
  31         this.sendReponse(res, { success: false, error: err.message })
  32       }
  33     })
  34   }
  35
  36   private async process (req: IPCRequest) {
  37     switch (req.type) {
  38       case 'register':
  39         await this.runnerServer.registerRunner(pick(req, [ 'url', 'registrationToken', 'runnerName', 'runnerDescription' ]))
  40         return undefined
  41
  42       case 'unregister':
  43         await this.runnerServer.unregisterRunner(pick(req, [ 'url', 'runnerName' ]))
  44         return undefined
  45
  46       case 'list-registered':
  47         return Promise.resolve(this.runnerServer.listRegistered())
  48
  49       default:
  50         throw new Error('Unknown RPC call ' + (req as any).type)
  51     }
  52   }
  53
  54   private sendReponse <T extends IPCReponseData> (
  55     response: (data: any) => Promise<void>,
  56     body: IPCReponse<T>
  57   ) {
  58     response(body)
  59       .catch(err => logger.error('Cannot send response after IPC request', err))
  60   }
  61 }