/**
 * Copyright (c) 2018-present, tarant
 *
 * This source code is licensed under the MIT license found in the
 * LICENSE file in the root directory of this source tree.
 */ import Message from '../mailbox/message' ; import ISubscriber from '../mailbox/subscriber' ; import ActorMessage from './actor-message' ; import ActorSystem from './actor-system' ; import IMaterializer from './materializer/materializer' ; import { Topic } from '..' ; import uuid from '../helper/uuid' ; import ActorProxy from './actor-proxy' ; import IActorSupervisor , { SupervisionStrategies } from './supervision/actor-supervisor' ; type Timer = any ; type Cancellable = string ; export interface IActor extends ISubscriber < ActorMessage > , IActorSupervisor { id : string ; } /**
 * Class that must be extended by all actors. All defined public methods in actors should be
 * asynchronous (return a Promise<T>) or return void.
 */ export default abstract class Actor implements IActor { public readonly id : string ; public readonly partitions : string [ ] ; protected readonly self : this = this ; protected readonly system ? : ActorSystem ; private readonly materializers : IMaterializer [ ] = [ ] ; private readonly supervisor ? : IActorSupervisor ; private readonly scheduled : Map < Cancellable , Timer > = new Map ( ) ; private readonly topicSubscriptions : Map < string , string > = new Map ( ) ; private busy : boolean = false ; protected constructor ( id ? ) { this . id = id || uuid ( ) ; this . partitions = [ this . id ] ; } param message public async onReceiveMessage ( message ) { if ( this . busy ) { return false ; } this . busy = true ; const actorMessage = message . content ; try { this . materializers . forEach ( ( materializer ) => materializer . onBeforeMessage ( this , actorMessage ) ) ; const result = await this . dispatchAndPromisify ( actorMessage ) ; actorMessage . resolve ( result ) ; } catch ( ex ) { this . materializers . forEach ( ( materializer ) => materializer . onError ( this , actorMessage , ex ) ) ; const strategy = await this . supervisor ! . supervise ( this . self , ex , actorMessage ) ; if ( strategy === 'drop-message' ) { actorMessage . reject ( ex ) ; return true ; } else if ( strategy === 'retry-message' ) { return false ; } else { actorMessage . reject ( ex ) ; return true ; } } finally { this . busy = false ; this . materializers . forEach ( ( materializer ) => materializer . onAfterMessage ( this , actorMessage ) ) ; } return true ; } param actor param exception param message public supervise ( actor , exception , message ) { return this . supervisor ! . supervise ( actor , exception , message ) ; } param interval param fn param values see Actor protected schedule ( interval , fn , values ) { const id = uuid ( ) ; setTimeout ( ( ) => { const sysAny = this . system as any ; this . scheduled . set ( id , setInterval ( ( ) => ActorProxy . sendAndReturn ( sysAny . mailbox , this . id , fn . name , values ) , interval ) ) ; } , 0 ) ; return id ; } param timeout param fn param values see Actor protected scheduleOnce ( timeout , fn , values ) { const id = uuid ( ) ; setTimeout ( ( ) => { const sysAny = this . system as any ; this . scheduled . set ( id , setTimeout ( ( ) => { ActorProxy . sendAndReturn ( sysAny . mailbox , this . id , fn . name , values ) ; this . scheduled . delete ( id ) ; } , timeout ) ) ; } , 0 ) ; return id ; } param cancellable see Actor see Actor protected cancel ( cancellable ) { setTimeout ( ( ) => { const id = this . scheduled . get ( cancellable ) as Timer ; clearTimeout ( id ) ; clearInterval ( id ) ; this . scheduled . delete ( cancellable ) ; } , 0 ) ; } param classFn param values protected actorOf < T extends IActor > ( classFn , values ) { const actor = this . system ! . actorOf ( classFn , values ) as any ; actor . ref . supervisor = this ; return actor ; } protected subscribeToTopic ( topic ) { setTimeout ( async ( ) => { const topicSubscription = await topic . subscribe ( this ) ; this . topicSubscriptions . set ( topic . id , topicSubscription ) ; } , 0 ) ; } protected unsubscribeFromTopic ( topic ) { const id = this . topicSubscriptions . get ( topic . id ) ; if ( id === undefined ) { return ; } topic . unsubscribe ( id ) ; this . topicSubscriptions . delete ( topic . id ) ; } private dispatchAndPromisify ( actorMessage ) { try { const r = this . constructor . prototype [ actorMessage . methodName ] . apply ( this , actorMessage . arguments ) ; if ( r && r . then && r . catch ) { return r ; } else { return Promise . resolve ( r ) ; } } catch ( ex ) { return Promise . reject ( ex ) ; } } private initialized ( ) { this . materializers . forEach ( ( materializer ) => materializer . onInitialize ( this ) ) ; } }