import { onFnCall , proxyFn , uuid , CreateTask } from '../../../common/utils' ; import { EventEmitter } from 'events' ; import { join } from 'path' ; import { Worker as NodeWorker } from 'worker_threads' ; type EventFn = { [ index ] : ( ... args : any [ ] ) => void ; } ; type RequestEventFn = { [ index ] : ( ... args : any [ ] ) => Promise < any > ; } ; type OnContextHandler = ( fn : ( func : string , args : any [ ] ) => any ) => void ; interface WorkerOptions { workerData ? : any ; sharedMemorySize ? : number ; } const Worker = ( name , opts = { } as WorkerOptions ) => { const modulePath = join ( __dirname , '..' , `${ name }.js` ) ; const worker = new NodeWorker ( modulePath , { workerData : opts . workerData , } ) ; const ee = new EventEmitter ( ) ; const pendingRequests = new Map ( ) ; const sharedBuffer = new SharedArrayBuffer ( opts . sharedMemorySize || 4 ) ; const sharedArray = new Int32Array ( sharedBuffer ) ; const wakeThread = ( ) => Atomics . notify ( sharedArray , 0 ) ; const writeSharedArray = ( id , data ) => { const jsonString = JSON . stringify ( data ) || 'null' ; const buffer = Buffer . from ( jsonString ) ; const length = buffer . byteLength ; for ( let ix = 0 ; ix < length ; ix ++ ) Atomics . store ( sharedArray , ix + 2 , buffer [ ix ] ) ; Atomics . store ( sharedArray , 1 , length ) ; Atomics . store ( sharedArray , 0 , id ) ; wakeThread ( ) ; } ; const call = onFnCall ( ( event , args ) => worker . postMessage ( [ event , args ] ) ) ; const on = proxyFn ( ( event , cb ) => ee . on ( event , cb ) ) ; const request = onFnCall ( ( event , args ) => { const task = CreateTask ( ) ; const id = uuid ( ) ; pendingRequests . set ( id , task . done ) ; worker . postMessage ( [ event , args , id ] ) ; return task . promise ; } ) ; const onContextHandler = ( fn ) => ee . on ( 'context-handler' , fn ) ; worker . on ( 'message' , async ( [ e , data , id , requestSync , func ] ) => { if ( e === '@@request-sync-context' ) { const listener = ee . listeners ( 'context-handler' ) [ 0 ] ; if ( ! listener ) throw new Error ( 'no "onContextHandler" function registered for synchronous RPC requests. you should register a function handler with "onContextHandler"' ) ; try { const result = await listener ( func , data ) ; return writeSharedArray ( id , result ) ; } catch ( err ) { console . error ( 'worker request-sync-context response listener failed:' , err ) ; return writeSharedArray ( id , undefined ) ; } } if ( requestSync ) { const listener = ee . listeners ( e ) [ 0 ] ; if ( ! listener ) return wakeThread ( ) ; const result = await listener ( ... data ) ; if ( ! result ) return wakeThread ( ) ; return writeSharedArray ( id , result ) ; } if ( ! id ) return ee . emit ( e , ... data ) ; if ( pendingRequests . has ( id ) ) { pendingRequests . get ( id ) ( data ) ; pendingRequests . delete ( id ) ; return ; } const listener = ee . listeners ( e ) [ 0 ] ; if ( ! listener ) return ; try { const result = await listener ( ... data ) ; worker . postMessage ( [ e , result , id ] ) ; } catch ( err ) { console . error ( 'worker request response listener failed:' , err ) ; } } ) ; const terminate = ( ) => worker . terminate ( ) ; worker . postMessage ( [ '@@sab' , [ sharedBuffer ] ] ) ; return { on , call , request , onContextHandler , terminate } ; } ; export default Worker ; export type Worker = ReturnType < typeof Worker > ;