import { log } from "./util/logging" ; import { EventEmitter } from "events" ; import { ITicks , INeonNotification } from "./util/progress" ; function _onUncaughtException ( error ) { log ( `[BAD] UncaughtException: ${ error . message }` ) ; } function _onUnhandledRejection ( reason , promise ) { if ( reason instanceof Error ) { log ( `[BAD] UnhandledRejection: ${ reason . message }` ) ; } else { log ( `[BAD] UnhandledRejection happened. No reason as error was provided.` ) ; } } process . on ( "uncaughtException" , _onUncaughtException ) ; process . on ( "unhandledRejection" , _onUnhandledRejection ) ; export enum ChannelType { IndexingChannel , DltIndexingChannel } export interface IChannel { poll : ( callback : ( err : string , e : { [ x ] : any ; event_tag : string ; } ) => void ) => void ; shutdown : ( ) => void ; } export class NativeComputationManager < T > { public static EVENTS = { GotItem : "GotItem" , Progress : "Progress" , Stopped : "Stopped" , Finished : "Finished" , Notification : "Notification" , Error : "error" , } ; shutdownRequested : boolean ; isShutdown : boolean ; eventEmitter : EventEmitter ; constructor ( channel ) { this . eventEmitter = new EventEmitter ( ) ; log ( "init of NativeComputationManager" ) ; this . shutdownRequested = false ; this . isShutdown = false ; const loop = ( ) => { if ( this . shutdownRequested ) { log ( "shutdown had been requested, now accepted!" ) ; this . shutdownRequested = false ; channel . shutdown ( ) ; } if ( this . isShutdown ) { log ( "shutting down loop" ) ; return ; } channel . poll ( ( err , eventData ) => { if ( err ) { log ( "Problem polling the channel: " + err ) ; this . eventEmitter . emit ( "error" , err ) ; } else if ( eventData ) { const { event , ... data } = eventData ; this . eventEmitter . emit ( event , data ) ; if ( event == NativeComputationManager . EVENTS . Finished ) { this . isShutdown = true ; log ( "shutdownAcknowledged" ) ; return ; } } setImmediate ( loop ) ; } ) ; } ; setImmediate ( loop ) ; } /**
     * onItem
     */ public onItem ( callback ) { this . eventEmitter . on ( NativeComputationManager . EVENTS . GotItem , callback ) ; } /**
     * onProgress
     */ public onProgress ( callback ) { this . eventEmitter . on ( NativeComputationManager . EVENTS . Progress , callback ) ; return this ; } /**
     * onStopped
     */ public onStopped ( callback ) { this . eventEmitter . on ( NativeComputationManager . EVENTS . Stopped , callback ) ; } /**
     * onNotification
     */ public onNotification ( callback ) { this . eventEmitter . on ( NativeComputationManager . EVENTS . Notification , callback ) ; } /**
     * onError
     */ public onError ( callback ) { this . eventEmitter . on ( NativeComputationManager . EVENTS . Error , callback ) ; } /**
     * onFinished
     */ public onFinished ( callback ) { this . eventEmitter . on ( NativeComputationManager . EVENTS . Finished , callback ) ; } /**
     * requestShutdown
     *
     * Mark the channel for shutdown
     */ public requestShutdown ( ) { this . shutdownRequested = true ; return this ; } }