import { v4 } from 'uuid' ; import { memoize , throttle } from 'lodash' ; import { promiseSleep } from './promiseSleep' ; import { CommsNode , CommsNodeCallbacks , CommsNodeOptions , SendToOptions } from './commsNode' ; import { closeMessage } from './peerMessageHandler' ; export enum McastMessageType { connect = 'connect' , otherPeers = 'otherPeers' , data = 'data' , close = 'close' } interface McastMessage { peerId : string ; userId : string ; type : McastMessageType ; recipientIds ? : string [ ] ; payload ? : any ; } interface McastConnection { timestamp : number ; userId : string ; } /**
 * A node in a multicast network.  Uses gtove-relay-server to do the multicasting.
 */ export class McastNode extends CommsNode { static MCAST_URL = 'https://radiant-thicket-18054.herokuapp.com/mcast/' ; private signalChannelId : string ; private readonly onEvents : CommsNodeCallbacks ; private connectedPeers : { [ key ] : McastConnection ; } ; private ignoredPeers : { [ key ] : boolean ; } ; private readonly memoizedThrottle : ( key : string , func : ( ... args : any [ ] ) => any ) => ( ... args : any [ ] ) => any ; private sequenceId : number | null = null ; private requestOffersInterval : number ; param signalChannelId param userId param commsNodeOptions constructor ( signalChannelId , userId , commsNodeOptions ) { super ( ) ; this . signalChannelId = signalChannelId ; this . options = commsNodeOptions ; this . onEvents = commsNodeOptions . onEvents || { } ; this . connectedPeers = { } ; this . peerId = v4 ( ) ; this . ignoredPeers = { [ this . peerId ] : true } ; this . userId = userId ; const throttleWait = commsNodeOptions . throttleWait || 250 ; console . log ( `Created multi-cast node for user ${ this . userId } with id ${ this . peerId }` ) ; this . memoizedThrottle = memoize ( ( throttleKey , func ) => ( throttle ( func , throttleWait ) ) ) ; this . sendToRaw = this . sendToRaw . bind ( this ) ; } async init ( ) { const listenPromise = this . listen ( ) ; await this . sendConnectMessage ( ) ; this . startHeartbeat ( ) ; return listenPromise ; } startHeartbeat ( ) { if ( ! this . requestOffersInterval ) { this . requestOffersInterval = window . setInterval ( this . heartbeat . bind ( this ) , CommsNode . HEARTBEAT_INTERVAL_MS ) ; } } async heartbeat ( ) { await this . sendConnectMessage ( ) ; const timeout = Date . now ( ) - 2 * CommsNode . HEARTBEAT_INTERVAL_MS ; for ( let peerId of Object . keys ( this . connectedPeers ) ) { if ( this . connectedPeers [ peerId ] . timestamp < timeout ) { console . warn ( `Peer ${ peerId } sent last message ${ ( Date . now ( ) - this . connectedPeers [ peerId ] . timestamp ) / 1000 } seconds ago - destroying it.` ) ; this . destroyPeer ( peerId ) ; } } } async getFromMcastServer ( ) { const response = await fetch ( `${ McastNode . MCAST_URL }${ this . signalChannelId }${ this . sequenceId !== null ? `?sequenceId=${ this . sequenceId }` : '' }` , { cache : 'no-store' } ) ; if ( response . ok ) { this . sequenceId = Number ( response . headers . get ( 'x-relay-sequenceId' ) ) ; return response . json ( ) ; } else { throw new Error ( 'invalid response on GET from mcast server: ' + response . statusText ) ; } } async postToMcastServer ( body ) { const response = await fetch ( `${ McastNode . MCAST_URL }${ this . signalChannelId }` , { method : 'POST' , body : JSON . stringify ( body ) } ) ; if ( ! response . ok ) { throw new Error ( 'invalid response on POST to mcast server: ' + response . statusText ) ; } } async sendConnectMessage ( ) { await this . postToMcastServer ( { type : McastMessageType . connect , peerId : this . peerId , userId : this . userId } ) ; } return { Promise } async listen ( ) { while ( ! this . shutdown ) { try { const message = await this . getFromMcastServer ( ) ; if ( this . shutdown || ! message . type || this . ignoredPeers [ message . peerId ] ) { continue ; } if ( message . recipientIds === undefined || message . recipientIds ! . indexOf ( this . peerId ) >= 0 ) { await this . onEvent ( message . type , message . peerId , message . userId , message . payload ) ; } if ( ! this . ignoredPeers [ message . peerId ] && ! this . connectedPeers [ message . peerId ] ) { await this . sendConnectMessage ( ) ; } if ( this . connectedPeers [ message . peerId ] ) { this . connectedPeers [ message . peerId ] . timestamp = Date . now ( ) ; } } catch ( err ) { console . error ( err ) ; await promiseSleep ( 5000 ) ; } } } handleNewConnection ( peerId , userId ) { if ( ! this . ignoredPeers [ peerId ] && ! this . connectedPeers [ peerId ] ) { if ( this . onEvents . shouldConnect && ! this . onEvents . shouldConnect ( this , peerId , userId ) ) { this . ignoredPeers [ peerId ] = true ; } else { console . log ( 'Established connection with' , peerId ) ; this . connectedPeers [ peerId ] = { timestamp : Date . now ( ) , userId } ; return true ; } } return false ; } async onEvent ( type , senderId , userId , payload ) { switch ( type ) { case McastMessageType . connect : if ( this . handleNewConnection ( senderId , userId ) ) { await this . postToMcastServer ( { type : McastMessageType . otherPeers , recipientIds : [ senderId ] , peerId : this . peerId , userId : this . userId , payload : Object . keys ( this . connectedPeers ) . map ( ( peerId ) => ( { peerId , userId : this . connectedPeers [ peerId ] . userId } ) ) . concat ( { peerId : this . peerId , userId : this . userId } ) } ) ; } else { return ; } break ; case McastMessageType . otherPeers : const connectedPeers = payload as { peerId : string ; userId : string ; } [ ] ; for ( let peer of connectedPeers ) { if ( this . handleNewConnection ( peer . peerId , peer . userId ) ) { await this . doCustomEvents ( McastMessageType . connect , peer . peerId , null ) ; } } break ; case McastMessageType . close : this . onClose ( senderId ) ; break ; default : break ; } await this . doCustomEvents ( type , senderId , payload ) ; } async doCustomEvents ( type , senderId , payload ) { if ( this . onEvents [ type ] ) { await this . onEvents [ type ] ( this , senderId , payload ) ; } } onClose ( peerId ) { if ( this . connectedPeers [ peerId ] ) { console . log ( 'Lost connection with' , peerId ) ; delete ( this . connectedPeers [ peerId ] ) ; } delete ( this . ignoredPeers [ peerId ] ) ; } async destroyPeer ( peerId ) { this . onClose ( peerId ) ; await this . doCustomEvents ( McastMessageType . close , peerId , null ) ; } private async sendToRaw ( message , recipientIds , onSentMessage ? ) { const payload = ( typeof ( message ) === 'object' ) ? JSON . stringify ( message , ( k , v ) => ( v === undefined ? null : v ) ) : message ; await this . postToMcastServer ( { type : McastMessageType . data , recipientIds , peerId : this . peerId , userId : this . userId , payload } ) ; onSentMessage && onSentMessage ( recipientIds , message ) ; } param message param only param except param throttleKey param onSentMessage async sendTo ( message , { only , except , throttleKey , onSentMessage } = { } ) { const recipients = ( only || Object . keys ( this . connectedPeers ) ) . filter ( ( peerId ) => ( ! except || except . indexOf ( peerId ) < 0 ) ) ; if ( recipients && recipients . length === 0 ) { onSentMessage && onSentMessage ( recipients , message ) ; return ; } if ( throttleKey ) { await this . memoizedThrottle ( throttleKey , this . sendToRaw ) ( message , recipients , onSentMessage ) ; } else { await this . sendToRaw ( message , recipients , onSentMessage ) ; } } async disconnectAll ( ) { await this . postToMcastServer ( { type : McastMessageType . close , peerId : this . peerId , userId : this . userId } ) ; for ( let peerId of Object . keys ( this . connectedPeers ) ) { await this . doCustomEvents ( McastMessageType . close , peerId , null ) ; } this . connectedPeers = { } ; } async destroy ( ) { console . log ( 'Shutting down multicast node' , this . peerId ) ; this . shutdown = true ; await this . disconnectAll ( ) ; } async close ( peerId , reason ? ) { if ( this . connectedPeers [ peerId ] ) { if ( reason ) { await this . sendToRaw ( closeMessage ( reason ) , [ peerId ] ) ; } await this . destroyPeer ( peerId ) ; } } }