import { Presence , PresenceClient , PresenceMessage } from '@syncot/presence' ; import { assert , EmitterInterface , noop , SyncOTEmitter } from '@syncot/util' ; import { Duplex } from 'readable-stream' ; /**
 * The events emitted by `PresenceSync`.
 */ export interface PresenceSyncEvents { change : string ; error : Error ; } event event event export interface PresenceSync extends EmitterInterface < SyncOTEmitter < PresenceSyncEvents > > { /**
     * A map of Presence objects indexed by sessionId.
     * It is synchronized with a server based on a filter.
     */ readonly presence : ReadonlyMap < string , Presence > ; } /**
 * Synchronizes `Presence` objects by the specified `sessionId`.
 */ export function syncPresenceBySessionId ( presenceClient , sessionId ) { assertPresenceClient ( presenceClient ) ; assertString ( sessionId , 'sessionId' ) ; return new Sync ( presenceClient , sessionId , undefined , undefined ) ; } /**
 * Synchronizes `Presence` objects by the specified `userId`.
 */ export function syncPresenceByUserId ( presenceClient , userId ) { assertPresenceClient ( presenceClient ) ; assertString ( userId , 'userId' ) ; return new Sync ( presenceClient , undefined , userId , undefined ) ; } /**
 * Synchronizes `Presence` objects by the specified `locationId`.
 */ export function syncPresenceByLocationId ( presenceClient , locationId ) { assertPresenceClient ( presenceClient ) ; assertString ( locationId , 'locationId' ) ; return new Sync ( presenceClient , undefined , undefined , locationId ) ; } /**
 * Synchronizes `Presence` objects by the current `locationId`.
 */ export function syncPresenceByCurrentLocationId ( presenceClient ) { assertPresenceClient ( presenceClient ) ; return new Sync ( presenceClient , undefined , undefined , undefined ) ; } function assertPresenceClient ( presenceClient ) { assert ( presenceClient && typeof presenceClient === 'object' , 'Argument "presenceClient" must be an object.' ) ; } function assertString ( value , name ) { assert ( typeof value === 'string' , `Argument "${ name }" must be a string.` ) ; } class Sync extends SyncOTEmitter < PresenceSyncEvents > implements PresenceSync { public readonly presence : Map < string , Presence > = new Map ( ) ; private stream : Duplex | undefined = undefined ; private streamPromise : Promise < Duplex > | undefined = undefined ; private currentLocationId : string | undefined = undefined ; public constructor ( private readonly presenceClient , private readonly sessionId , private readonly userId , private readonly locationId ) { super ( ) ; this . presenceClient . auth . on ( 'active' , this . onOnline ) ; this . presenceClient . on ( 'presence' , this . onLocalPresence ) ; this . initStream ( ) ; } public destroy ( ) { if ( this . destroyed ) return ; this . presenceClient . auth . off ( 'active' , this . onOnline ) ; this . presenceClient . off ( 'presence' , this . onLocalPresence ) ; this . streamPromise = undefined ; if ( this . stream ) this . stream . destroy ( ) ; super . destroy ( ) ; } private onOnline = ( ) : void => { this . initStream ( ) ; } ; private onError = ( error ) : void => { this . emitAsync ( 'error' , error ) ; } ; private onData = ( message ) : void => { if ( message [ 0 ] ) { for ( let i = 1 , l = message . length ; i < l ; ++ i ) { const presence = message [ i ] as Presence ; this . presence . set ( presence . sessionId , presence ) ; this . emitAsync ( 'change' , presence . sessionId ) ; } } else { for ( let i = 1 , l = message . length ; i < l ; ++ i ) { const sessionId = message [ i ] as string ; this . presence . delete ( sessionId ) ; this . emitAsync ( 'change' , sessionId ) ; } } } ; private onClose = ( ) : void => { this . stream ! . off ( 'error' , this . onError ) ; this . stream ! . off ( 'data' , this . onData ) ; this . stream ! . off ( 'close' , this . onClose ) ; this . stream ! . on ( 'error' , noop ) ; this . stream = undefined ; this . presence . forEach ( ( presence ) => { this . emitAsync ( 'change' , presence . sessionId ) ; } ) ; this . presence . clear ( ) ; this . initStream ( ) ; } ; private onLocalPresence = ( ) : void => { if ( this . stream && this . sessionId === undefined && this . userId === undefined && this . locationId === undefined && this . currentLocationId !== this . presenceClient . locationId ) { this . stream . destroy ( ) ; } } ; private initStream ( ) { if ( this . stream ) { this . stream . destroy ( ) ; return ; } if ( ! this . presenceClient . auth . active ) { return ; } this . currentLocationId = this . presenceClient . locationId ; const streamPromise = this . sessionId !== undefined ? this . presenceClient . streamPresenceBySessionId ( this . sessionId ) : this . userId !== undefined ? this . presenceClient . streamPresenceByUserId ( this . userId ) : this . locationId !== undefined ? this . presenceClient . streamPresenceByLocationId ( this . locationId ) : this . currentLocationId !== undefined ? this . presenceClient . streamPresenceByLocationId ( this . currentLocationId ) : undefined ; this . streamPromise = streamPromise ; if ( ! streamPromise ) { return ; } streamPromise . then ( ( stream ) => { if ( this . streamPromise === streamPromise ) { this . stream = stream ; this . stream . on ( 'error' , this . onError ) ; this . stream . on ( 'data' , this . onData ) ; this . stream . on ( 'close' , this . onClose ) ; } else { stream . on ( 'error' , noop ) ; stream . destroy ( ) ; } } , ( error ) => { if ( this . streamPromise === streamPromise ) { this . onError ( error ) ; } } ) ; } }