import Peer from 'simple-peer' ; import { v4 } from 'uuid' ; import { memoize , throttle } from 'lodash' ; import { promiseSleep } from './promiseSleep' ; import { CommsNode , CommsNodeCallbacks , CommsNodeOptions , SendToOptions } from './commsNode' ; import { closeMessage } from './peerMessageHandler' ; interface ConnectedPeer { peerId : string ; peer : Peer . Instance ; connected : number ; initiatedByMe : boolean ; errorCount : number ; linkReadUrl ? : string ; linkWriteUrl ? : string ; lastSignal : number ; } export interface SimplePeerOffer { sdp : string ; type : 'offer' | 'answer' ; } interface SignalMessage { peerId : string ; userId : string ; offer ? : SimplePeerOffer ; recipientId ? : string ; } /**
 * A node in a peer-to-peer network.  Uses gtove-relay-server to do signalling and webRTC (via simple-peer) for the actual
 * communication.  Builds a totally connected network topology - each node in the network has a direct peer-to-peer
 * connection to each other node.
 */ export class PeerNode extends CommsNode { static SIGNAL_URL = 'https://radiant-thicket-18054.herokuapp.com/mcast/' ; static RELAY_URL = 'https://radiant-thicket-18054.herokuapp.com/link/' ; private requestOffersInterval : number ; private readonly signalChannelId : string ; private onEvents : CommsNodeCallbacks ; private connectedPeers : { [ key ] : ConnectedPeer ; } ; private ignoredPeers : { [ key ] : boolean ; } ; private readonly memoizedThrottle : ( key : string , func : ( ... args : any [ ] ) => any ) => ( ... args : any [ ] ) => any ; private sequenceId : number | null = null ; param signalChannelId param userId param commsNodeOptions constructor ( signalChannelId , userId , commsNodeOptions ) { super ( ) ; this . signalChannelId = signalChannelId ; this . options = commsNodeOptions ; this . onEvents = commsNodeOptions . onEvents || { } ; this . connectedPeers = { } ; this . ignoredPeers = { } ; this . peerId = v4 ( ) ; this . userId = userId ; const throttleWait = commsNodeOptions . throttleWait || 250 ; console . log ( `Created peer-to-peer 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 ( ) { await this . requestOffers ( ) ; this . startHeartbeat ( ) ; await this . listenForSignal ( ) ; } private onSignalError ( error ) { this . onEvents . signalError && this . onEvents . signalError ( this , error ? 'Signal error' : '' ) ; } async getFromSignalServer ( ) { const response = await fetch ( `${ PeerNode . SIGNAL_URL }${ this . signalChannelId }${ this . sequenceId !== null ? `?sequenceId=${ this . sequenceId }` : '' }` , { cache : 'no-store' } ) ; if ( response . ok ) { this . onSignalError ( false ) ; this . sequenceId = Number ( response . headers . get ( 'x-relay-sequenceId' ) ) ; return response . json ( ) ; } else { this . onSignalError ( true ) ; throw new Error ( 'invalid response from signal server: ' + JSON . stringify ( response ) ) ; } } async postToSignalServer ( body ) { let success = true ; try { const response = await fetch ( `${ PeerNode . SIGNAL_URL }${ this . signalChannelId }` , { method : 'POST' , body : JSON . stringify ( body ) } ) ; success = response . ok ; } catch ( e ) { success = false ; } this . onSignalError ( ! success ) ; if ( ! success ) { await promiseSleep ( 5000 ) ; await this . postToSignalServer ( body ) ; } } param signal async handleSignal ( signal ) { if ( this . shutdown || ! signal . peerId ) { return ; } if ( signal . peerId !== this . peerId && ! this . ignoredPeers [ signal . peerId ] ) { if ( ! this . connectedPeers [ signal . peerId ] ) { if ( ! this . onEvents . shouldConnect || this . onEvents . shouldConnect ( this , signal . peerId , signal . userId ) ) { this . addPeer ( signal . peerId , signal . recipientId !== this . peerId ) ; } else { this . ignoredPeers [ signal . peerId ] = true ; } } } if ( this . connectedPeers [ signal . peerId ] ) { this . connectedPeers [ signal . peerId ] . lastSignal = Date . now ( ) ; } if ( signal . recipientId === this . peerId ) { if ( this . ignoredPeers [ signal . peerId ] ) { console . error ( `Received a message addressed to me from ignored peer ${ signal . userId }/${ signal . peerId }` ) ; } else if ( this . connectedPeers [ signal . peerId ] . connected ) { if ( signal . offer && signal . offer . type === 'offer' ) { if ( this . connectedPeers [ signal . peerId ] . linkReadUrl || Date . now ( ) - this . connectedPeers [ signal . peerId ] . connected < 3000 ) { return ; } console . warn ( `Re-initiating connection with ${ signal . peerId }.` ) ; this . connectedPeers [ signal . peerId ] . peer && this . connectedPeers [ signal . peerId ] . peer . destroy ( ) ; this . addPeer ( signal . peerId , false ) ; this . connectedPeers [ signal . peerId ] . peer . signal ( signal . offer ) ; } else if ( ! signal . offer ) { if ( this . connectedPeers [ signal . peerId ] . linkReadUrl ) { return ; } console . warn ( `Changing already-established connection with ${ signal . peerId } to relay.` ) ; this . connectViaRelay ( signal . peerId , false ) ; } else { if ( Date . now ( ) - this . connectedPeers [ signal . peerId ] . connected < 3000 ) { return ; } console . warn ( `Received unexpected answer to our offer from ${ signal . peerId }` ) ; } } else { if ( ! signal . offer ) { await this . connectViaRelay ( signal . peerId , false ) ; } else { if ( this . connectedPeers [ signal . peerId ] . initiatedByMe && signal . offer . type === 'offer' ) { if ( this . peerId > signal . peerId ) { this . connectedPeers [ signal . peerId ] . peer && this . connectedPeers [ signal . peerId ] . peer . destroy ( ) ; this . addPeer ( signal . peerId , false ) ; } else { return ; } } this . connectedPeers [ signal . peerId ] . peer . signal ( signal . offer ) ; } } } } /**
     * Listens for a message from the signalling server, gtove-relay-server.  Signals are used to establish connections
     * between webRTC peers.
     */ async listenForSignal ( ) { while ( ! this . shutdown ) { try { const signal = await this . getFromSignalServer ( ) ; await this . handleSignal ( signal ) ; } catch ( err ) { console . error ( err ) ; await promiseSleep ( 5000 ) ; } } } async requestOffers ( ) { await this . postToSignalServer ( { peerId : this . peerId , userId : this . userId } ) ; } startHeartbeat ( ) { if ( ! this . requestOffersInterval ) { this . requestOffersInterval = window . setInterval ( this . heartbeat . bind ( this ) , CommsNode . HEARTBEAT_INTERVAL_MS ) ; } } async heartbeat ( ) { await this . requestOffers ( ) ; const timeout = Date . now ( ) - 2 * CommsNode . HEARTBEAT_INTERVAL_MS ; for ( let peerId of Object . keys ( this . connectedPeers ) ) { if ( this . connectedPeers [ peerId ] . lastSignal < timeout ) { console . warn ( `Peer ${ peerId } sent last signal ${ ( Date . now ( ) - this . connectedPeers [ peerId ] . lastSignal ) / 1000 } seconds ago - destroying it.` ) ; this . destroyPeer ( peerId ) ; } } } async sendOffer ( peerId , offer ) { await this . postToSignalServer ( { peerId : this . peerId , userId : this . userId , offer , recipientId : peerId } ) ; await promiseSleep ( 5000 ) ; if ( this . connectedPeers [ peerId ] && this . connectedPeers [ peerId ] . connected ) { return ; } if ( this . connectedPeers [ peerId ] ) { this . connectViaRelay ( peerId , offer . type === 'offer' ) ; } } private async connectViaRelay ( peerId , initiator ) { console . log ( `Failed to connect peer-to-peer with ${ peerId }, falling back to relay.` ) ; this . connectedPeers [ peerId ] = { peerId , peer : this . connectedPeers [ peerId ] && this . connectedPeers [ peerId ] . peer , connected : Date . now ( ) , initiatedByMe : initiator , lastSignal : Date . now ( ) , linkReadUrl : PeerNode . RELAY_URL + `${ this . peerId }-${ peerId }` , linkWriteUrl : PeerNode . RELAY_URL + `${ peerId }-${ this . peerId }` , errorCount : 0 } ; this . listenForRelayTraffic ( peerId ) ; if ( initiator ) { await this . postToSignalServer ( { peerId : this . peerId , userId : this . userId , recipientId : peerId } ) ; } this . connectedPeers [ peerId ] . peer . emit ( 'connect' ) ; } async listenForRelayTraffic ( peerId ) { while ( ! this . shutdown && this . connectedPeers [ peerId ] ) { try { const response = await fetch ( this . connectedPeers [ peerId ] . linkReadUrl ! , { cache : 'no-store' } ) ; if ( response . ok ) { const message = await response . json ( ) ; this . connectedPeers [ peerId ] . peer . emit ( message . type , message . message ) ; } else { this . connectedPeers [ peerId ] . peer . emit ( 'error' , response ) ; } } catch ( e ) { if ( this . connectedPeers [ peerId ] && this . connectedPeers [ peerId ] . peer ) { this . connectedPeers [ peerId ] . peer . emit ( 'error' , e ) ; } } } } addPeer ( peerId , initiator ) { const peer = new Peer ( { initiator , trickle : false } ) ; peer . on ( 'signal' , ( offer ) => ( this . onSignal ( peerId , offer ) ) ) ; peer . on ( 'error' , ( error ) => ( this . onError ( peerId , error ) ) ) ; peer . on ( 'close' , ( ) => ( this . onClose ( peerId ) ) ) ; peer . on ( 'connect' , ( ) => ( this . onConnect ( peerId ) ) ) ; Object . keys ( this . onEvents ) . forEach ( ( event ) => { peer . on ( event , ( ... args ) => this . onEvents [ event ] ( this , peerId , ... args ) ) ; } ) ; this . connectedPeers [ peerId ] = { peerId , peer , connected : 0 , initiatedByMe : initiator , errorCount : 0 , lastSignal : Date . now ( ) } ; } async onSignal ( peerId , offer ) { if ( ! this . connectedPeers [ peerId ] . linkReadUrl ) { await this . sendOffer ( peerId , offer ) ; } } onError ( peerId , error ) { console . error ( 'Error from' , peerId , error ) ; delete ( this . connectedPeers [ peerId ] ) ; } onClose ( peerId ) { console . log ( 'Lost connection with' , peerId ) ; delete ( this . connectedPeers [ peerId ] ) ; } onConnect ( peerId ) { console . log ( 'Established connection with' , peerId ) ; this . connectedPeers [ peerId ] . connected = Date . now ( ) ; } private async sendDataViaP2POrRelay ( peerId , message ) { try { if ( this . connectedPeers [ peerId ] . linkWriteUrl ) { await fetch ( this . connectedPeers [ peerId ] . linkWriteUrl ! , { method : 'POST' , body : JSON . stringify ( { type : 'data' , message } ) } ) ; } else { this . connectedPeers [ peerId ] . peer . send ( message ) ; } this . connectedPeers [ peerId ] . errorCount = 0 ; } catch ( e ) { if ( this . connectedPeers [ peerId ] ) { this . connectedPeers [ peerId ] . errorCount ++ ; const consecutive = this . connectedPeers [ peerId ] . errorCount === 1 ? '' : ` (${ this . connectedPeers [ peerId ] . errorCount } consecutive errors)` ; console . error ( `Error sending to peer ${ peerId }${ consecutive }:` , e ) ; if ( this . connectedPeers [ peerId ] . errorCount > 10 ) { console . error ( `Too many errors, giving up on peer ${ peerId }` ) ; this . destroyPeer ( peerId ) ; } } } } destroyPeer ( peerId ) { console . log ( `Destroying connection with peer ${ peerId }.` ) ; if ( this . connectedPeers [ peerId ] ) { if ( this . connectedPeers [ peerId ] . linkWriteUrl ) { fetch ( this . connectedPeers [ peerId ] . linkWriteUrl ! , { method : 'POST' , body : JSON . stringify ( { type : 'close' } ) } ) ; if ( this . connectedPeers [ peerId ] . peer ) { this . connectedPeers [ peerId ] . peer . emit ( 'close' ) ; } else { this . onEvents . close && this . onEvents . close ( this , peerId ) ; } } else if ( ! this . connectedPeers [ peerId ] . connected ) { if ( this . connectedPeers [ peerId ] . peer ) { this . connectedPeers [ peerId ] . peer . emit ( 'close' ) ; } else { this . onEvents . close && this . onEvents . close ( this , peerId ) ; } } else { if ( this . connectedPeers [ peerId ] . peer ) { this . connectedPeers [ peerId ] . peer . destroy ( ) ; } else { this . onEvents . close && this . onEvents . close ( this , peerId ) ; } } delete ( this . connectedPeers [ peerId ] ) ; } } private async sendToRaw ( message , recipients , onSentMessage ? ) { const stringMessage = ( typeof ( message ) === 'object' ) ? JSON . stringify ( message , ( k , v ) => ( v === undefined ? null : v ) ) : message ; for ( let peerId of recipients ) { if ( this . connectedPeers [ peerId ] && this . connectedPeers [ peerId ] . connected ) { await this . sendDataViaP2POrRelay ( peerId , stringMessage ) ; } else { const intervalId = window . setInterval ( async ( ) => { if ( this . connectedPeers [ peerId ] ) { if ( this . connectedPeers [ peerId ] . connected ) { await this . sendDataViaP2POrRelay ( peerId , stringMessage ) ; } else { return ; } } window . clearInterval ( intervalId ) ; } , 250 ) ; } } onSentMessage && onSentMessage ( recipients , 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 ( throttleKey ) { await this . memoizedThrottle ( throttleKey , this . sendToRaw ) ( message , recipients , onSentMessage ) ; } else { await this . sendToRaw ( message , recipients , onSentMessage ) ; } } async disconnectAll ( ) { Object . keys ( this . connectedPeers ) . forEach ( ( peerId ) => ( this . destroyPeer ( peerId ) ) ) ; this . connectedPeers = { } ; } async destroy ( ) { console . log ( 'Shutting down peer-to-peer node' , this . peerId ) ; this . shutdown = true ; window . clearInterval ( this . requestOffersInterval ) ; await this . disconnectAll ( ) ; } async close ( peerId , reason ? ) { if ( this . connectedPeers [ peerId ] ) { if ( reason ) { await this . sendToRaw ( closeMessage ( reason ) , [ peerId ] ) ; } this . destroyPeer ( peerId ) ; } } }