import { BehaviorSubject , Observable , Subject , Subscription } from "rxjs" ; import { debounceTime , sampleTime } from "rxjs/operators" ; import { clone , Controller , IoSource , IoSourceBackpressureStrategy , IoStateEvent , Uuid } from ".." ; /**
 * A tuple related to an IoSource:
 * - the IO source object
 * - state subscription
 * - subject for updateRate
 * - subject for association state
 * - subject for pushing IO values
 * - subscription for observable that emits update values
 */ type IoSourceItems = [ IoSource , Subscription , BehaviorSubject < number > , BehaviorSubject < boolean > , Subject < any > , Subscription ] ; /**
 * Provides data transfer rate controlled publishing of IO values for
 * IO sources and monitoring of changes in the association state of IO sources.
 *
 * This controller respects the backpressure strategy of an IO source in order to
 * cope with IO values that are more rapidly produced than specified in the
 * recommended update rate.
 */ export class IoSourceController extends Controller { private _sourceItems : Map < Uuid , IoSourceItems > ; onInit ( ) { super . onInit ( ) ; this . _sourceItems = new Map < Uuid , IoSourceItems > ( ) ; } onCommunicationManagerStarting ( ) { super . onCommunicationManagerStarting ( ) ; this . _reregisterAll ( ) ; } onCommunicationManagerStopping ( ) { super . onCommunicationManagerStopping ( ) ; this . _deregisterAll ( ) ; } param source param value publish < T > ( source , value ) { const [ , , updateRate , association , updateSubject ] = this . _ensureRegistered < T > ( source ) ; if ( ! association . value ) { return ; } const rate = updateRate . value ; if ( rate === undefined || rate === 0 || source . updateStrategy === IoSourceBackpressureStrategy . None ) { updateSubject . next ( value ) ; return ; } if ( value instanceof Uint8Array ) { value = Uint8Array . from ( value ) as any ; } else if ( typeof value === "string" || typeof value === "boolean" || typeof value === "number" || typeof value === "object" ) { value = clone ( value ) ; } updateSubject . next ( value ) ; } param source observeUpdateRate < T > ( source ) { const [ , , updateRate ] = this . _ensureRegistered < T > ( source ) ; return updateRate . asObservable ( ) ; } param source observeAssociation < T > ( source ) { const [ , , , association ] = this . _ensureRegistered < T > ( source ) ; return association . asObservable ( ) ; } /**
     * Determines whether the given IO source is currently associated.
     */ isAssociated < T > ( source ) { const [ , , , association ] = this . _ensureRegistered < T > ( source ) ; return association . value ; } private _reregisterAll ( ) { this . _sourceItems . forEach ( item => { const source = item [ 0 ] ; const ioState = this . communicationManager . observeIoState ( source ) ; item [ 1 ] = ioState . subscribe ( event => this . _onIoStateChanged ( source . objectId , event ) ) ; item [ 2 ] . next ( ioState . value . data . updateRate ) ; item [ 3 ] . next ( ioState . value . data . hasAssociations ) ; this . _updateUpdateRateObservable ( item ) ; } ) ; } private _deregisterAll ( ) { this . _sourceItems . forEach ( item => { item [ 1 ] ?. unsubscribe ( ) ; item [ 1 ] = undefined ; item [ 5 ] ?. unsubscribe ( ) ; item [ 5 ] = undefined ; } ) ; } private _ensureRegistered < T > ( source ) { const sourceId = source . objectId ; let item = this . _sourceItems . get ( sourceId ) ; if ( ! item ) { const ioState = this . communicationManager . observeIoState ( source ) ; item = [ source , ioState . subscribe ( event => this . _onIoStateChanged ( sourceId , event ) ) , new BehaviorSubject < number > ( ioState . value . data . updateRate ) , new BehaviorSubject < boolean > ( ioState . value . data . hasAssociations ) , new Subject < T > ( ) , undefined , ] ; this . _sourceItems . set ( sourceId , item ) ; this . _updateUpdateRateObservable ( item ) ; } return item ; } private _onIoStateChanged ( sourceId , event ) { const item = this . _sourceItems . get ( sourceId ) ; if ( ! item ) { return ; } let needsUpdate = false ; if ( item [ 3 ] . value !== event . data . hasAssociations ) { item [ 3 ] . next ( event . data . hasAssociations ) ; needsUpdate = true ; } if ( item [ 2 ] . value !== event . data . updateRate ) { item [ 2 ] . next ( event . data . updateRate ) ; needsUpdate = true ; } needsUpdate && this . _updateUpdateRateObservable ( item ) ; } private _updateUpdateRateObservable ( item ) { const [ source , , updateRate , association , updateSubject , updateSubscription ] = item ; updateSubscription ?. unsubscribe ( ) ; if ( ! association . value ) { item [ 5 ] = undefined ; return ; } const rate = updateRate . value ; let updateObs ; if ( rate === undefined || rate === 0 ) { updateObs = updateSubject ; } else { switch ( source . updateStrategy || IoSourceBackpressureStrategy . Default ) { case IoSourceBackpressureStrategy . Default : case IoSourceBackpressureStrategy . Sample : updateObs = updateSubject . pipe ( sampleTime ( rate ) ) ; break ; case IoSourceBackpressureStrategy . Throttle : updateObs = updateSubject . pipe ( debounceTime ( rate ) ) ; break ; case IoSourceBackpressureStrategy . None : updateObs = updateSubject ; break ; default : updateObs = updateSubject ; break ; } } item [ 5 ] = updateObs . subscribe ( value => this . communicationManager . publishIoValue ( source , value ) ) ; } }