import { BehaviorSubject , merge , Observable , Subject , Subscription } from "rxjs" ; import { concatMap , filter } from "rxjs/operators" ; import { Uuid } from ".." ; import { Observation , Sensor , Thing } from "./objects" ; import { SensorObserverController } from "./sensor-observer-controller" ; import { ThingObserverController } from "./thing-observer-controller" ; /**
 * Represents change information when sensors are registered or deregistered.
 */ export interface RegisteredSensorsChangeInfo { /**
     * New sensors that have been added.
     */ added : Array < Readonly < Sensor > > ; /**
     * Previously added sensors that have been removed.
     */ removed : Array < Readonly < Sensor > > ; /**
     * Existing sensors that have been readvertised (some properties might have changed).
     */ changed : Array < Readonly < Sensor > > ; /**
     * The current sensors after changes have been applied.
     */ total : Array < Readonly < Sensor > > ; } /**
 * A convenience controller that observes things and associated sensor and
 * observation objects, combining the functionality of `ThingObserverController`
 * and `SensorObserverController`.
 *
 * This controller is designed to be used by a client that wants to easily
 * handle sensor-related events as well as sensor observations.
 */ export class ThingSensorObservationObserverController extends ThingObserverController { private _sensorObserverController : SensorObserverController ; private _thingOnlineSubscription : Subscription ; private _thingOfflineSubscription : Subscription ; private _sensorSubscriptions = new Map < string , Subscription [ ] > ( ) ; private _observationSubscriptions = new Map < Uuid , Subscription > ( ) ; private _registeredSensors : Map < Uuid , Sensor > ; private _registeredSensorsChangeInfo$ : BehaviorSubject < RegisteredSensorsChangeInfo > ; private _sensorObservation$ = new Subject < { obs : Observation ; sensor : Sensor ; } > ( ) ; private _thingFilter : ( thing : Thing ) => boolean ; private _sensorFilter : ( sensor : Sensor , thing : Thing ) => boolean ; onInit ( ) { super . onInit ( ) ; this . _registeredSensors = new Map < Uuid , Sensor > ( ) ; this . _registeredSensorsChangeInfo$ = new BehaviorSubject < RegisteredSensorsChangeInfo > ( { added : [ ] , removed : [ ] , changed : [ ] , total : [ ] , } ) ; this . _sensorObserverController = this . container . registerController ( "SensorObserverController_" + this . runtime . newUuid ( ) , SensorObserverController ) ; if ( ! this . _sensorObserverController ) { throw new Error ( `SensorObserverController could not be registered.` ) ; } } onCommunicationManagerStarting ( ) { super . onCommunicationManagerStarting ( ) ; this . _observeThings ( ) ; } onCommunicationManagerStopping ( ) { super . onCommunicationManagerStopping ( ) ; this . _sensorSubscriptions . forEach ( subs => subs . forEach ( sub => sub . unsubscribe ( ) ) ) ; this . _sensorSubscriptions . clear ( ) ; this . _observationSubscriptions . forEach ( sub => sub . unsubscribe ( ) ) ; this . _observationSubscriptions . clear ( ) ; this . _thingOnlineSubscription ?. unsubscribe ( ) ; this . _thingOfflineSubscription ?. unsubscribe ( ) ; this . _registeredSensorsChangeInfo$ . next ( { added : [ ] , removed : Array . from ( this . _registeredSensors . values ( ) ) , changed : [ ] , total : [ ] , } ) ; this . _registeredSensors . clear ( ) ; } /**
     * Gets an Observable emitting information about changes in the currently
     * registered sensors.
     *
     * Registered sensor objects are augmented by a property `thing` which
     * references the associated `Thing` object.
     *
     * Emitted sensor objects are read-only. If you need to modify one, e.g. to
     * delete the `thing` property, clone the object first (using `clone()`
     * function in `@coaty/core`).
     */ get registeredSensorsChangeInfo$ ( ) { return this . _registeredSensorsChangeInfo$ ; } /**
     * Gets an Observable emitting incoming observations on registered sensors.
     */ get sensorObservation$ ( ) { return this . _sensorObservation$ ; } param thingFilter set thingFilter ( thingFilter ) { this . _thingFilter = thingFilter ; } param sensorFilter set sensorFilter ( sensorFilter ) { this . _sensorFilter = sensorFilter ; } private _observeThings ( ) { this . _thingOnlineSubscription = merge ( this . discoverThings ( ) , this . observeAdvertisedThings ( ) ) . subscribe ( thing => this . _onThingReceived ( thing ) ) ; this . _thingOfflineSubscription = this . communicationManager . observeDeadvertise ( ) . subscribe ( event => event . data . objectIds . forEach ( id => this . _onDeadvertised ( id ) ) ) ; } private _onThingReceived ( thing ) { if ( this . _thingFilter && ! this . _thingFilter ( thing ) ) { return ; } const advertiseObservable = this . _sensorObserverController . observeAdvertisedSensors ( ) . pipe ( filter ( sensor => sensor . parentObjectId === thing . objectId ) ) ; const subscription = merge ( this . _sensorObserverController . querySensorsOfThing ( thing . objectId ) . pipe ( concatMap ( sensors => { return new Observable < Sensor > ( observer => { sensors . forEach ( sensor => { observer . next ( sensor ) ; } ) ; observer . complete ( ) ; } ) ; } ) ) , advertiseObservable ) . subscribe ( sensor => this . _onSensorReceived ( sensor , thing ) ) ; if ( this . _sensorSubscriptions . has ( thing . objectId ) ) { this . _sensorSubscriptions . get ( thing . objectId ) . push ( subscription ) ; } else { this . _sensorSubscriptions . set ( thing . objectId , [ subscription ] ) ; } } private _onSensorReceived ( sensor , thing ) { if ( this . _sensorFilter && ! this . _sensorFilter ( sensor , thing ) ) { return ; } sensor [ "thing" ] = thing ; const isRegistered = this . _registeredSensors . has ( sensor . objectId ) ; this . _registeredSensors . set ( sensor . objectId , sensor ) ; this . _registeredSensorsChangeInfo$ . next ( { added : isRegistered ? [ ] : [ sensor ] , removed : [ ] , changed : isRegistered ? [ sensor ] : [ ] , total : Array . from ( this . _registeredSensors . values ( ) ) , } ) ; if ( this . _observationSubscriptions . has ( sensor . objectId ) ) { this . _observationSubscriptions . get ( sensor . objectId ) . unsubscribe ( ) ; } this . _observationSubscriptions . set ( sensor . objectId , this . _observeObservations ( sensor ) ) ; } private _observeObservations ( sensor ) { return this . _sensorObserverController . observeChanneledObservations ( sensor . objectId ) . subscribe ( observation => { this . _sensorObservation$ . next ( { obs : observation , sensor } ) ; } ) ; } private _onDeadvertised ( objectId ) { const removedThingIds = new Set < Uuid > ( ) ; const removedSensors = [ ] ; this . _registeredSensors . forEach ( ( s , id ) => { if ( s [ "thing" ] . parentObjectId === objectId ) { removedThingIds . add ( s [ "thing" ] . objectId ) ; this . _observationSubscriptions . get ( id ) . unsubscribe ( ) ; this . _observationSubscriptions . delete ( id ) ; removedSensors . push ( s ) ; this . _registeredSensors . delete ( id ) ; } } ) ; removedThingIds . forEach ( id => { this . _sensorSubscriptions . get ( id ) . forEach ( sub => sub . unsubscribe ( ) ) ; this . _sensorSubscriptions . delete ( id ) ; } ) ; if ( removedSensors . length === 0 ) { return ; } this . _registeredSensorsChangeInfo$ . next ( { added : [ ] , removed : removedSensors , changed : [ ] , total : Array . from ( this . _registeredSensors . values ( ) ) , } ) ; } }