import { BehaviorSubject , merge , Observable , Subscriber } from "rxjs" ; import { distinctUntilChanged , filter , map } from "rxjs/operators" ; import { CoatyObject , CommunicationOptions , Container , CoreType , CoreTypes , IoActor , IoNode , IoSource , Uuid , } from ".." ; import { IDisposable } from "../runtime/disposable" ; import { AdvertiseEvent , AdvertiseEventData } from "./advertise" ; import { AssociateEvent , AssociateEventData } from "./associate" ; import { CallEvent , CallEventData , ReturnEvent , ReturnEventData } from "./call-return" ; import { ChannelEvent , ChannelEventData } from "./channel" ; import { CommunicationBindingProtocol , CommunicationBindingState , CommunicationEventLike , CommunicationEventType , CommunicationState , } from "./communication-binding" ; import { CommunicationEvent , CommunicationEventData , TopicEvent , TopicEventData } from "./communication-event" ; import { DeadvertiseEvent , DeadvertiseEventData } from "./deadvertise" ; import { DiscoverEvent , DiscoverEventData , ResolveEvent , ResolveEventData } from "./discover-resolve" ; import { IoStateEvent } from "./io-state" ; import { IoValueEvent } from "./io-value" ; import { QueryEvent , QueryEventData , RetrieveEvent , RetrieveEventData } from "./query-retrieve" ; import { RawEvent , RawEventData } from "./raw" ; import { CompleteEvent , CompleteEventData , UpdateEvent , UpdateEventData } from "./update-complete" ; import { MqttBinding } from "./mqtt/mqtt-binding" ; /**
 * Indicates the current operating state of a Communication Manager.
 */ export enum OperatingState { /** Indicates the communication manager is stopped or initialized. */ Stopped , /** Indicates the communication manager is started. */ Started } /**
 * Provides a set of predefined communication events to transfer Coaty objects
 * between distributed Coaty agents based on publish-subscribe messaging.
 */ export class CommunicationManager implements IDisposable { private _configOptions : CommunicationOptions ; private _options : CommunicationOptions ; private _binding : CommunicationBindingProtocol ; private _ioNodes : IoNode [ ] ; private _isDisposed : boolean ; private _communicationStateSubject : BehaviorSubject < CommunicationState > ; private _operatingStateSubject : BehaviorSubject < OperatingState > ; private _observedRequests : Map < string , ObservedRequestItem > ; private _observedResponses : Map < Uuid , ObservedResponseItem > ; private _observedIoStateItems : Map < Uuid , IoStateItem > ; private _ioSourceItems : Map < Uuid , [ string , Uuid [ ] , number ] > ; private _ioActorItems : Map < string , Map < Uuid , Uuid [ ] > > ; private _observedIoValueItems : Map < Uuid , AnyValueItem > ; internal constructor ( private _container , options ) { this . _configOptions = options || { } ; this . _isDisposed = false ; this . _communicationStateSubject = new BehaviorSubject ( CommunicationState . Offline ) ; this . _operatingStateSubject = new BehaviorSubject ( OperatingState . Stopped ) ; this . _initObservedItems ( ) ; this . _initOptions ( ) ; } /**
     * Gets the container's Runtime object.
     */ get runtime ( ) { return this . _container . runtime ; } remarks get options ( ) : Readonly < CommunicationOptions > { return this . _options ; } /**
     * Gets the IO node for the given IO context name, as configured in the
     * configuration common options.
     *
     * Returns undefined if no IO node is configured for the context name.
     */ getIoNodeByContext ( contextName ) { return this . _ioNodes . find ( node => node . name === contextName ) ; } param options returns start ( options ? ) { if ( ! options ) { if ( this . _binding . state === CommunicationBindingState . Initialized ) { this . _joinBinding ( ) ; } return Promise . resolve ( ) ; } return this . _unjoinBinding ( options ) . then ( ( ) => this . _joinBinding ( ) ) ; } deprecated param options returns restart ( options ? ) { return this . start ( options ) ; } returns stop ( ) { return this . _unjoinBinding ( ) ; } returns observeCommunicationState ( ) { return this . _communicationStateSubject . pipe ( distinctUntilChanged ( ) ) ; } returns observeOperatingState ( ) { return this . _operatingStateSubject . asObservable ( ) ; } returns observeDiscover ( ) { return this . _observeRequest ( CommunicationEventType . Discover ) as Observable < DiscoverEvent > ; } returns observeQuery ( ) { return this . _observeRequest ( CommunicationEventType . Query ) as Observable < QueryEvent > ; } param coreType returns observeUpdateWithCoreType ( coreType ) { return this . _observeUpdate ( coreType ) ; } param objectType returns throws observeUpdateWithObjectType ( objectType ) { return this . _observeUpdate ( undefined , objectType ) ; } remarks param operation param context returns throws observeCall ( operation , context ? ) { if ( ! CommunicationEvent . isValidEventFilter ( operation ) ) { throw new TypeError ( `${ operation } is not a valid operation name` ) ; } return ( this . _observeRequest ( CommunicationEventType . Call , operation ) as Observable < CallEvent > ) . pipe ( filter ( event => event . data . matchesFilter ( context ) ) ) ; } param coreType returns observeAdvertiseWithCoreType ( coreType ) { return this . _observeAdvertise ( coreType ) ; } param objectType returns throws observeAdvertiseWithObjectType ( objectType ) { return this . _observeAdvertise ( undefined , objectType ) ; } param eventTarget returns observeDeadvertise ( ) { return this . _observeRequest ( CommunicationEventType . Deadvertise ) as Observable < DeadvertiseEvent > ; } param channelId returns throws observeChannel ( channelId ) { if ( ! CommunicationEvent . isValidEventFilter ( channelId ) ) { throw new TypeError ( `${ channelId } is not a valid channel identifier` ) ; } return this . _observeRequest ( CommunicationEventType . Channel , channelId ) as Observable < ChannelEvent > ; } remarks remarks remarks remarks param topic param options returns observeRaw ( topic , options ? ) { return ( this . _observeRequest ( CommunicationEventType . Raw , topic , options ) as Observable < RawEvent > ) . pipe ( map ( rawEvent => [ rawEvent . topic , rawEvent . data . payload as Uint8Array ] ) ) ; } returns observeIoState ( ioPoint ) { return this . _observeIoState ( ioPoint . objectId ) ; } returns observeIoValue ( ioActor ) { return this . _observeIoValue ( ioActor . objectId ) ; } param event returns publishDiscover ( event ) { return this . _publishEvent ( event ) as Observable < ResolveEvent > ; } param event returns publishQuery ( event ) { return this . _publishEvent ( event ) as Observable < RetrieveEvent > ; } param event returns publishUpdate ( event ) { const coreType = event . data . object . coreType ; const objectType = event . data . object . objectType ; const coreCompletes = this . _publishEvent ( event , coreType ) ; if ( CoreTypes . getObjectTypeFor ( coreType ) !== objectType ) { return merge ( this . _publishEvent ( event , ":" + objectType ) , coreCompletes ) as Observable < CompleteEvent > ; } return coreCompletes as Observable < CompleteEvent > ; } param event returns publishCall ( event ) { return this . _publishEvent ( event , event . operation ) as Observable < ReturnEvent > ; } param event publishAdvertise ( event ) { const coreType = event . data . object . coreType ; const objectType = event . data . object . objectType ; this . _publishEvent ( event , coreType ) ; if ( CoreTypes . getObjectTypeFor ( coreType ) !== objectType ) { this . _publishEvent ( event , ":" + objectType ) ; } } param event publishDeadvertise ( event ) { this . _publishEvent ( event ) ; } param event publishChannel ( event ) { this . _publishEvent ( event , event . channelId ) ; } remarks param event publishRaw ( event ) ; deprecated param topic param value param shouldRetain publishRaw ( topic , value , shouldRetain ? ) ; publishRaw ( topicOrEvent , value ? , shouldRetain ? ) { if ( typeof topicOrEvent === "string" ) { topicOrEvent = RawEvent . withTopicAndPayload ( topicOrEvent , value , { retain : shouldRetain || false } ) ; } this . _binding . publish ( this . _createEventLike ( topicOrEvent , topicOrEvent . topic ) ) ; } param event publishIoValue ( event ) ; deprecated param ioSource param value publishIoValue ( ioSource , value ) ; publishIoValue ( ioSourceOrEvent , value ? ) { if ( ! ( ioSourceOrEvent instanceof IoValueEvent ) ) { ioSourceOrEvent = IoValueEvent . with ( ioSourceOrEvent , value ) ; } const items = this . _ioSourceItems . get ( ioSourceOrEvent . ioSource . objectId ) ; if ( items ) { ioSourceOrEvent . topic = items [ 0 ] ; this . _binding . publish ( this . _createEventLike ( ioSourceOrEvent , ioSourceOrEvent . topic ) ) ; } } param ioSource returns createAssociatingRoute ( ioSource ) { return this . _binding . createIoRoute ( ioSource . objectId ) ; } internal param event publishAssociate ( event ) { if ( ! CommunicationEvent . isValidEventFilter ( event . eventTypeFilter ) ) { console . log ( `[CommunicationManager] [err] ioContextName '${ event . eventTypeFilter }' in Associate event contains invalid characters` ) ; return ; } this . _publishEvent ( event , event . eventTypeFilter ) ; } /**
     * Perform clean-up side effects, such as stopping the communication manager and
     * deallocating resources.
     */ onDispose ( ) { if ( this . _isDisposed ) { return ; } this . _isDisposed = true ; this . _unjoinBinding ( ) ; } private _initObservedItems ( ) { this . _observedRequests = new Map < string , ObservedRequestItem > ( ) ; this . _observedResponses = new Map < Uuid , ObservedResponseItem > ( ) ; this . _observedIoStateItems = new Map < Uuid , IoStateItem > ( ) ; this . _observedIoValueItems = new Map < Uuid , AnyValueItem > ( ) ; this . _ioSourceItems = new Map < Uuid , [ string , Uuid [ ] , number ] > ( ) ; this . _ioActorItems = new Map < string , Map < Uuid , Uuid [ ] > > ( ) ; } private _initOptions ( options ? ) { this . _options = { ... this . _configOptions , ... ( options || { } ) } ; const namespace = this . _options . namespace || this . _options . binding ?. options . namespace ; if ( namespace !== undefined && ! CommunicationEvent . isValidEventFilter ( namespace ) ) { console . log ( `[CommunicationManager] [err] namespace '${ namespace }' contains invalid characters` ) ; } const ioNodesConfig = this . runtime . commonOptions ?. ioContextNodes || { } ; this . _ioNodes = Object . keys ( ioNodesConfig ) . filter ( contextName => { if ( CommunicationEvent . isValidEventFilter ( contextName ) ) { return true ; } console . log ( `[CommunicationManager] [err] ioContextName '${ contextName }' in ioContextNodes contains invalid characters` ) ; return false ; } ) . map ( contextName => { const ioNodeConfig = ioNodesConfig [ contextName ] ; return < IoNode > { name : contextName , coreType : "IoNode" , objectType : CoreTypes . OBJECT_TYPE_IO_NODE , objectId : this . runtime . newUuid ( ) , parentObjectId : this . _container . identity . objectId , ioSources : ioNodeConfig . ioSources || [ ] , ioActors : ioNodeConfig . ioActors || [ ] , characteristics : ioNodeConfig . characteristics , } ; } ) . filter ( node => node . ioSources . length > 0 || node . ioActors . length > 0 ) ; const bindingWithOptions = this . _options . binding || MqttBinding . withOptions ( { namespace : this . _options . namespace , shouldEnableCrossNamespacing : this . _options . shouldEnableCrossNamespacing || false , brokerUrl : this . _options . brokerUrl , tlsOptions : { ... this . _options . mqttClientOptions } , } ) ; this . _binding = new bindingWithOptions . type ( bindingWithOptions . options ) ; const bindingName = `${ this . _binding . apiName }v${ this . _binding . apiVersion }` ; this . _binding . on ( "debug" , debug => console . log ( `[${ bindingName }] [dbg] ${ debug }` ) ) ; this . _binding . on ( "info" , info => console . log ( `[${ bindingName }] [inf] ${ info }` ) ) ; this . _binding . on ( "error" , error => console . log ( `[${ bindingName }] [err] ${ error }` ) ) ; this . _binding . on ( "communicationState" , state => this . _communicationStateSubject . next ( state ) ) ; this . _binding . on ( "inboundEvent" , eventLike => this . _dispatchEvent ( eventLike ) ) ; } private _updateOperatingState ( newState ) { if ( this . _operatingStateSubject . getValue ( ) !== newState ) { this . _operatingStateSubject . next ( newState ) ; } } private _joinBinding ( ) { const identity = this . _container . identity ; const joinEvents = [ ] ; joinEvents . push ( AdvertiseEvent . withObject ( identity ) ) ; this . _ioNodes . forEach ( ioNode => { joinEvents . push ( AdvertiseEvent . withObject ( ioNode ) ) ; } ) ; const unjoinEvent = DeadvertiseEvent . withObjectIds ( ... joinEvents . map ( e => e . data . object . objectId ) ) ; this . _binding . join ( { agentId : identity . objectId , joinEvents : joinEvents . map ( e => this . _createEventLike ( e , e . data . object . coreType ) ) , unjoinEvent : this . _createEventLike ( unjoinEvent ) , } ) ; this . _observeAssociate ( ) ; this . _observeDiscoverIoNodes ( ) ; this . _observeDiscoverIdentity ( ) ; this . _updateOperatingState ( OperatingState . Started ) ; } private _unjoinBinding ( options ? ) { this . _updateOperatingState ( OperatingState . Stopped ) ; this . _unobserveObservedItems ( ) ; const binding = this . _binding ; if ( ! this . _isDisposed ) { this . _initOptions ( options ) ; } return binding . unjoin ( ) ; } private _publishEvent ( event , eventTypeFilter ? , correlationId ? ) { if ( correlationId ) { this . _binding . publish ( this . _createEventLike ( event , eventTypeFilter , correlationId ) ) ; return undefined ; } const responseType = event . responseEventType ; if ( responseType !== undefined ) { const corrId = this . runtime . newUuid ( ) ; return this . _observeResponse ( event , corrId , ( ) => this . _binding . publish ( this . _createEventLike ( event , eventTypeFilter , corrId ) ) ) ; } this . _binding . publish ( this . _createEventLike ( event , eventTypeFilter ) ) ; return undefined ; } remarks private _createEventLike ( event , eventTypeFilter ? , correlationId ? ) { const jsonData = event . data . toJsonObject ( ) ; return { eventType : event . eventType , eventTypeFilter , sourceId : this . _container . identity . objectId , correlationId , isDataRaw : jsonData === undefined , data : jsonData === undefined ? ( event . data as TopicEventData ) . payload : jsonData , options : event instanceof TopicEvent ? event . options : undefined , } ; } private _observeAdvertise ( coreType ? , objectType ? ) { if ( coreType && ! CoreTypes . isCoreType ( coreType ) ) { throw new TypeError ( `${ coreType } is not a CoreType` ) ; } if ( objectType && ! CommunicationEvent . isValidEventFilter ( objectType ) ) { throw new TypeError ( `${ objectType } is not a valid object type` ) ; } if ( objectType ) { const objectCoreType = CoreTypes . getCoreTypeFor ( objectType ) ; if ( objectCoreType ) { return ( this . _observeRequest ( CommunicationEventType . Advertise , objectCoreType ) as Observable < AdvertiseEvent > ) . pipe ( filter ( event => event . data . object . objectType === objectType ) ) ; } } return this . _observeRequest ( CommunicationEventType . Advertise , coreType || ":" + objectType ) as Observable < AdvertiseEvent > ; } private _observeUpdate ( coreType ? , objectType ? ) { if ( coreType && ! CoreTypes . isCoreType ( coreType ) ) { throw new TypeError ( `${ coreType } is not a CoreType` ) ; } if ( objectType && ! CommunicationEvent . isValidEventFilter ( objectType ) ) { throw new TypeError ( `${ objectType } is not a valid object type` ) ; } if ( objectType ) { const objectCoreType = CoreTypes . getCoreTypeFor ( objectType ) ; if ( objectCoreType ) { return ( this . _observeRequest ( CommunicationEventType . Update , objectCoreType ) as Observable < UpdateEvent > ) . pipe ( filter ( event => event . data . object . objectType === objectType ) ) ; } } return this . _observeRequest ( CommunicationEventType . Update , coreType || ":" + objectType ) as Observable < UpdateEvent > ; } private _observeAssociate ( ) { this . _ioNodes . forEach ( group => this . _observeRequest ( CommunicationEventType . Associate , group . name ) . subscribe ( event => this . _handleAssociate ( event as AssociateEvent ) ) ) ; } private _observeDiscoverIoNodes ( ) { if ( this . _ioNodes . length === 0 ) { return ; } this . _observeRequest ( CommunicationEventType . Discover ) . pipe ( filter ( ( event ) => ( event . data . isDiscoveringTypes && event . data . isCoreTypeCompatible ( "IoNode" ) ) ) ) . subscribe ( ( event ) => this . _ioNodes . forEach ( n => event . resolve ( ResolveEvent . withObject ( n ) ) ) ) ; } private _observeDiscoverIdentity ( ) { this . _observeRequest ( CommunicationEventType . Discover ) . pipe ( filter ( ( event ) => ( event . data . isDiscoveringTypes && event . data . isCoreTypeCompatible ( "Identity" ) ) || ( event . data . isDiscoveringObjectId && event . data . objectId === this . _container . identity . objectId ) ) ) . subscribe ( ( event ) => event . resolve ( ResolveEvent . withObject ( this . _container . identity ) ) ) ; } private _observeRequest ( eventType , eventTypeFilter ? , options ? ) { const requestId = this . _getRequestId ( eventType , eventTypeFilter ) ; let item = this . _observedRequests . get ( requestId ) ; if ( ! item ) { const eventLike = { eventType , eventTypeFilter , options , isDataRaw : eventType === CommunicationEventType . Raw , } ; const cleanup = ( ) => { this . _observedRequests . delete ( requestId ) ; this . _binding . unsubscribe ( eventLike ) ; } ; item = new ObservedRequestItem ( eventLike , cleanup ) ; this . _observedRequests . set ( requestId , item ) ; this . _binding . subscribe ( eventLike ) ; } return item . observable ; } private _observeResponse ( request , correlationId , publisher ) { const eventLike = { eventType : request . responseEventType , correlationId } ; const cleanup = ( itm ) => { this . _observedResponses . delete ( itm . correlationId ) ; this . _binding . unsubscribe ( eventLike ) ; } ; const item = new ObservedResponseItem ( request , correlationId , publisher , cleanup ) ; this . _observedResponses . set ( correlationId , item ) ; this . _binding . subscribe ( eventLike ) ; return item . createObservable ( ) ; } private _unobserveObservedItems ( ) { this . _unobserveIoStateAndValue ( ) ; this . _observedRequests . forEach ( item => { item . dispatchComplete ( ) ; } ) ; this . _observedResponses . forEach ( item => { item . dispatchComplete ( ) ; } ) ; this . _initObservedItems ( ) ; } private _getRequestId ( eventType , requestFilter ? ) { let name = CommunicationEventType [ eventType ] ; if ( requestFilter ) { name += requestFilter ; } return name ; } private _dispatchEvent ( eventLike ) { if ( eventLike . eventType === CommunicationEventType . IoValue ) { this . _dispatchIoValue ( eventLike ) ; return ; } if ( eventLike . eventType === CommunicationEventType . Complete || eventLike . eventType === CommunicationEventType . Resolve || eventLike . eventType === CommunicationEventType . Retrieve || eventLike . eventType === CommunicationEventType . Return ) { const item = this . _observedResponses . get ( eventLike . correlationId ) ; if ( item === undefined ) { return ; } if ( item . request . eventType === CommunicationEventType . Update && eventLike . eventType !== CommunicationEventType . Complete ) { console . log ( `[CommunicationManager] [err] expected Complete response event for Update` ) ; return ; } if ( item . request . eventType === CommunicationEventType . Discover && eventLike . eventType !== CommunicationEventType . Resolve ) { console . log ( `[CommunicationManager] [err] expected Resolve response event for Discover` ) ; return ; } if ( item . request . eventType === CommunicationEventType . Query && eventLike . eventType !== CommunicationEventType . Retrieve ) { console . log ( `[CommunicationManager] [err] expected Retrieve response event for Query` ) ; return ; } if ( item . request . eventType === CommunicationEventType . Call && eventLike . eventType !== CommunicationEventType . Return ) { console . log ( `[CommunicationManager] [err] expected Return response event for Call` ) ; return ; } try { const event = this . _createEventInstance ( eventLike , item . request ) ; switch ( event . eventType ) { case CommunicationEventType . Complete : const completeEvent = event as CompleteEvent ; completeEvent . eventRequest . ensureValidResponseParameters ( completeEvent . data ) ; break ; case CommunicationEventType . Resolve : const resolveEvent = event as ResolveEvent ; resolveEvent . eventRequest . ensureValidResponseParameters ( resolveEvent . data ) ; break ; case CommunicationEventType . Retrieve : const retrieveEvent = event as RetrieveEvent ; retrieveEvent . eventRequest . ensureValidResponseParameters ( retrieveEvent . data ) ; break ; case CommunicationEventType . Return : const returnEvent = event as ReturnEvent ; returnEvent . eventRequest . ensureValidResponseParameters ( returnEvent . data ) ; break ; } item . dispatchNext ( event ) ; } catch ( error ) { console . log ( `[CommunicationManager] [err] inbound response event data is not valid: ${ error }` ) ; return ; } } else { const item = this . _observedRequests . get ( this . _getRequestId ( eventLike . eventType , eventLike . eventType === CommunicationEventType . Raw ? eventLike . correlationId : eventLike . eventTypeFilter ) ) ; if ( item === undefined ) { return ; } try { const event = this . _createEventInstance ( eventLike ) ; const correlationId = eventLike . correlationId ; if ( event instanceof DiscoverEvent ) { event . resolve = ( resolveEvent ) => { event . ensureValidResponseParameters ( resolveEvent . data ) ; this . _publishEvent ( resolveEvent , undefined , correlationId ) ; } ; } else if ( event instanceof QueryEvent ) { event . retrieve = ( retrieveEvent ) => { event . ensureValidResponseParameters ( retrieveEvent . data ) ; this . _publishEvent ( retrieveEvent , undefined , correlationId ) ; } ; } else if ( event instanceof UpdateEvent ) { event . complete = ( completeEvent ) => { event . ensureValidResponseParameters ( completeEvent . data ) ; this . _publishEvent ( completeEvent , undefined , correlationId ) ; } ; } else if ( event instanceof CallEvent ) { event . returnEvent = ( returnEvent ) => { event . ensureValidResponseParameters ( returnEvent . data ) ; this . _publishEvent ( returnEvent , undefined , correlationId ) ; } ; } item . dispatchNext ( event ) ; } catch ( error ) { console . log ( `[CommunicationManager] [err] inbound request event data is not valid: ${ error }` ) ; return ; } } } private _dispatchIoValue ( eventLike ) { const items = this . _ioActorItems . get ( eventLike . correlationId ) ; if ( items ) { items . forEach ( ( sourceIds , actorId ) => { const ioValueItem = this . _observedIoValueItems . get ( actorId ) ; if ( ioValueItem ) { ioValueItem . dispatchNext ( eventLike . data ) ; } } ) ; } } remarks param eventLike returns throws private _createEventInstance ( eventLike , requestEvent ? ) { let instance ; switch ( eventLike . eventType ) { case CommunicationEventType . Advertise : instance = new AdvertiseEvent ( AdvertiseEventData . createFrom ( eventLike . data ) ) ; break ; case CommunicationEventType . Associate : instance = new AssociateEvent ( eventLike . eventTypeFilter , AssociateEventData . createFrom ( eventLike . data ) ) ; break ; case CommunicationEventType . Deadvertise : instance = new DeadvertiseEvent ( DeadvertiseEventData . createFrom ( eventLike . data ) ) ; break ; case CommunicationEventType . Channel : instance = new ChannelEvent ( eventLike . eventTypeFilter , ChannelEventData . createFrom ( eventLike . data ) ) ; break ; case CommunicationEventType . Discover : instance = new DiscoverEvent ( DiscoverEventData . createFrom ( eventLike . data ) ) ; break ; case CommunicationEventType . Resolve : instance = new ResolveEvent ( ResolveEventData . createFrom ( eventLike . data ) ) ; break ; case CommunicationEventType . Query : instance = new QueryEvent ( QueryEventData . createFrom ( eventLike . data ) ) ; break ; case CommunicationEventType . Retrieve : instance = new RetrieveEvent ( RetrieveEventData . createFrom ( eventLike . data ) ) ; break ; case CommunicationEventType . Update : instance = new UpdateEvent ( UpdateEventData . createFrom ( eventLike . data ) ) ; break ; case CommunicationEventType . Complete : instance = new CompleteEvent ( CompleteEventData . createFrom ( eventLike . data ) ) ; break ; case CommunicationEventType . Call : instance = new CallEvent ( eventLike . eventTypeFilter , CallEventData . createFrom ( eventLike . data ) ) ; break ; case CommunicationEventType . Return : instance = new ReturnEvent ( ReturnEventData . createFrom ( eventLike . data ) ) ; break ; case CommunicationEventType . Raw : instance = new RawEvent ( eventLike . eventTypeFilter , RawEventData . createFrom ( eventLike . data ) ) ; break ; default : throw new TypeError ( `Couldn't create event instance for event type ${ eventLike . eventType }` ) ; } instance . sourceId = eventLike . sourceId ; instance . eventRequest = requestEvent ; return instance ; } private _findIoPointById ( objectId ) { for ( const ioNode of this . _ioNodes ) { const source = ioNode . ioSources . find ( src => src . objectId === objectId ) ; if ( source ) { return source ; } const actor = ioNode . ioActors . find ( a => a . objectId === objectId ) ; if ( actor ) { return actor ; } } return undefined ; } private _handleAssociate ( event ) { const ioSourceId = event . data . ioSourceId ; const ioActorId = event . data . ioActorId ; const ioActor = this . _findIoPointById ( ioActorId ) as IoActor ; const isIoSourceAssociated = this . _findIoPointById ( ioSourceId ) !== undefined ; const isIoActorAssociated = ioActor !== undefined ; if ( ! isIoSourceAssociated && ! isIoActorAssociated ) { return ; } const ioRoute = event . data . associatingRoute ; if ( isIoSourceAssociated ) { this . _updateIoSourceItems ( ioSourceId , ioActorId , ioRoute , event . data . updateRate ) ; } if ( isIoActorAssociated ) { if ( ioRoute !== undefined ) { this . _associateIoActorItems ( ioSourceId , ioActor , ioRoute , event . data . isExternalRoute ) ; } else { this . _disassociateIoActorItems ( ioSourceId , ioActorId ) ; } } if ( isIoSourceAssociated ) { const item = this . _observedIoStateItems . get ( ioSourceId ) ; if ( item ) { const items = this . _ioSourceItems . get ( ioSourceId ) ; item . dispatchNext ( IoStateEvent . with ( items !== undefined && items [ 1 ] !== undefined , items ? items [ 2 ] : undefined ) ) ; } } if ( isIoActorAssociated ) { const item = this . _observedIoStateItems . get ( ioActorId ) ; if ( item ) { let actorIds ; if ( ioRoute !== undefined ) { actorIds = this . _ioActorItems . get ( ioRoute ) ; } item . dispatchNext ( IoStateEvent . with ( actorIds !== undefined && actorIds . has ( ioActorId ) && actorIds . get ( ioActorId ) . length > 0 ) ) ; } } } private _updateIoSourceItems ( ioSourceId , ioActorId , ioRoute , updateRate ) { let items = this . _ioSourceItems . get ( ioSourceId ) ; if ( ioRoute !== undefined ) { if ( ! items ) { items = [ ioRoute , [ ioActorId ] , updateRate ] ; this . _ioSourceItems . set ( ioSourceId , items ) ; } else { if ( items [ 0 ] === ioRoute ) { if ( items [ 1 ] . indexOf ( ioActorId ) === - 1 ) { items [ 1 ] . push ( ioActorId ) ; } } else { const previousRoute = items [ 0 ] ; items [ 0 ] = ioRoute ; items [ 1 ] . forEach ( actorId => this . _disassociateIoActorItems ( ioSourceId , actorId , previousRoute ) ) ; items [ 1 ] = [ ioActorId ] ; } items [ 2 ] = updateRate ; } } else { if ( items ) { const i = items [ 1 ] . indexOf ( ioActorId ) ; if ( i !== - 1 ) { items [ 1 ] . splice ( i , 1 ) ; } items [ 2 ] = updateRate ; if ( items [ 1 ] . length === 0 ) { this . _ioSourceItems . delete ( ioSourceId ) ; } } } } private _associateIoActorItems ( ioSourceId , ioActor , ioRoute , isExternalRoute ) { const ioActorId = ioActor . objectId ; this . _disassociateIoActorItems ( ioSourceId , ioActorId , undefined , ioRoute ) ; let items = this . _ioActorItems . get ( ioRoute ) ; if ( ! items ) { items = new Map < Uuid , Uuid [ ] > ( ) ; items . set ( ioActorId , [ ioSourceId ] ) ; this . _ioActorItems . set ( ioRoute , items ) ; this . _binding . subscribe ( { eventType : CommunicationEventType . IoValue , eventTypeFilter : ioRoute , isDataRaw : ioActor . useRawIoValues , isExternalIoRoute : isExternalRoute , } ) ; } else { const sourceIds = items . get ( ioActorId ) ; if ( ! sourceIds ) { items . set ( ioActorId , [ ioSourceId ] ) ; } else { if ( sourceIds . indexOf ( ioSourceId ) === - 1 ) { sourceIds . push ( ioSourceId ) ; } } } } private _disassociateIoActorItems ( ioSourceId , ioActorId , currentIoRoute ? , newIoRoute ? ) { const ioRoutesToUnsubscribe = [ ] ; const handler = ( items , route ) => { if ( newIoRoute === route ) { return ; } const sourceIds = items . get ( ioActorId ) ; if ( sourceIds ) { const i = sourceIds . indexOf ( ioSourceId ) ; if ( i !== - 1 ) { sourceIds . splice ( i , 1 ) ; } if ( sourceIds . length === 0 ) { items . delete ( ioActorId ) ; } if ( items . size === 0 ) { ioRoutesToUnsubscribe . push ( route ) ; } } } ; if ( currentIoRoute ) { const items = this . _ioActorItems . get ( currentIoRoute ) ; if ( items ) { handler ( items , currentIoRoute ) ; } } else { this . _ioActorItems . forEach ( handler ) ; } ioRoutesToUnsubscribe . forEach ( route => { this . _ioActorItems . delete ( route ) ; this . _binding . unsubscribe ( { eventType : CommunicationEventType . IoValue , eventTypeFilter : route } ) ; } ) ; } private _unobserveIoStateAndValue ( ) { this . _observedIoStateItems . forEach ( ( item , pointId ) => { item . dispatchNext ( IoStateEvent . with ( false , undefined ) ) ; item . dispatchComplete ( ) ; } ) ; this . _ioActorItems . forEach ( ( actorIds , ioRoute ) => { this . _binding . unsubscribe ( { eventType : CommunicationEventType . IoValue , eventTypeFilter : ioRoute } ) ; } ) ; this . _observedIoValueItems . forEach ( item => item . dispatchComplete ( ) ) ; } private _observeIoState ( ioPointId ) { let item = this . _observedIoStateItems . get ( ioPointId ) ; if ( ! item ) { let hasAssociations = false ; let updateRate ; const srcItems = this . _ioSourceItems . get ( ioPointId ) ; if ( srcItems ) { hasAssociations = true ; updateRate = srcItems [ 2 ] ; } else { this . _ioActorItems . forEach ( sourceItems => { const sourceIds = sourceItems . get ( ioPointId ) ; if ( sourceIds ) { hasAssociations = true ; updateRate = undefined ; } } ) ; } item = new IoStateItem ( IoStateEvent . with ( hasAssociations , updateRate ) ) ; this . _observedIoStateItems . set ( ioPointId , item ) ; } return item . subject ; } private _observeIoValue ( ioActorId ) { let item = this . _observedIoValueItems . get ( ioActorId ) ; if ( ! item ) { item = new AnyValueItem ( ) ; this . _observedIoValueItems . set ( ioActorId , item ) ; } return item . observable ; } } abstract class SubscriberItem < T > { private _subscribers : Array < Subscriber < T > > ; constructor ( ) { this . _subscribers = [ ] ; } dispatchNext ( message ) { for ( let i = this . _subscribers . length - 1 ; i >= 0 ; i -- ) { this . _subscribers [ i ] . next ( message ) ; } } dispatchComplete ( ) { for ( let i = this . _subscribers . length - 1 ; i >= 0 ; i -- ) { this . _subscribers [ i ] . complete ( ) ; } } dispatchError ( error ) { for ( let i = this . _subscribers . length - 1 ; i >= 0 ; i -- ) { this . _subscribers [ i ] . error ( error ) ; } } get subscriberCount ( ) { return this . _subscribers . length ; } protected add ( subscriber ) { this . _subscribers . push ( subscriber ) ; } protected remove ( subscriber ) { const subscriberIndex = this . _subscribers . indexOf ( subscriber ) ; if ( subscriberIndex !== - 1 ) { this . _subscribers . splice ( subscriberIndex , 1 ) ; } } } class ObservedRequestItem extends SubscriberItem < CommunicationEvent < CommunicationEventData > > { observable : Observable < CommunicationEvent < CommunicationEventData > > ; constructor ( public eventLike , private cleanup ) { super ( ) ; this . observable = this . _createObservable ( ) ; } private _createObservable ( ) { return Observable . create ( ( subscriber ) => { this . add ( subscriber ) ; return ( ) => { this . remove ( subscriber ) ; if ( this . subscriberCount === 0 ) { this . cleanup && this . cleanup ( ) ; this . cleanup = undefined ; } } ; } ) ; } } class ObservedResponseItem extends SubscriberItem < CommunicationEvent < CommunicationEventData > > { private _cleanup : ( item : ObservedResponseItem ) => void ; private _hasLastObserverUnsubscribed : boolean ; constructor ( public request , public correlationId , public publisher , cleanup ) { super ( ) ; this . _hasLastObserverUnsubscribed = false ; this . _cleanup = ( item ) => { cleanup ( item ) ; this . _cleanup = undefined ; this . publisher = undefined ; this . request = undefined ; } ; } createObservable ( ) { return Observable . create ( ( subscriber ) => { if ( this . _hasLastObserverUnsubscribed ) { throw new Error ( "in CommunicationManager: resubscribing to a response event is not supported" ) ; } this . add ( subscriber ) ; if ( this . subscriberCount === 1 ) { this . publisher && this . publisher ( ) ; } return ( ) => { this . remove ( subscriber ) ; if ( this . subscriberCount === 0 ) { this . _hasLastObserverUnsubscribed = true ; this . _cleanup && this . _cleanup ( this ) ; } } ; } ) ; } } class IoStateItem extends SubscriberItem < IoStateEvent > { subject : BehaviorSubject < IoStateEvent > ; constructor ( initialValue ) { super ( ) ; this . subject = new BehaviorSubject < IoStateEvent > ( initialValue ) ; } dispatchNext ( message ) { this . subject . next ( message ) ; } dispatchComplete ( ) { this . subject . complete ( ) ; } dispatchError ( error ) { this . subject . error ( error ) ; } } class AnyValueItem extends SubscriberItem < any > { observable : Observable < any > ; constructor ( ) { super ( ) ; this . observable = this . _createObservable ( ) ; } private _createObservable ( ) { return Observable . create ( ( subscriber ) => { this . add ( subscriber ) ; return ( ) => this . remove ( subscriber ) ; } ) ; } }