import { Server } from 'socket.io' ; import { io , Socket as ClientSocket } from 'socket.io-client' ; import { Message } from '@statechannels/client-api-schema' ; import delay from 'delay' ; import { AbortController } from 'abort-controller' ; import { MessageHandler , MessageServiceFactory , MessageServiceInterface } from './types' ; import { LatencyOptions } from './test-message-service' ; export class SocketIOMessageService implements MessageServiceInterface { private server : Server ; private peers : ClientSocket [ ] = [ ] ; private _abortController = new AbortController ( ) ; private _latencyOptions : LatencyOptions = { meanDelay : undefined , dropRate : 0 } ; constructor ( private handler , hostName , port ) { this . server = new Server ( ) ; this . server . listen ( port ) ; } public async registerPeer ( url ) { const socket = io ( url ) ; socket . on ( 'message' , async ( messages ) => { for ( const m of messages ) { await this . handler ( m ) ; } } ) ; this . peers . push ( socket ) ; } public setLatencyOptions ( options ) { this . _latencyOptions = options ; } public async send ( messages ) { const shouldDrop = Math . random ( ) > 1 - this . _latencyOptions . dropRate ; if ( ! shouldDrop ) { const { meanDelay } = this . _latencyOptions ; if ( meanDelay && meanDelay > 0 ) { const delayAmount = meanDelay / 2 + Math . random ( ) * meanDelay ; await delay ( delayAmount , { signal : this . _abortController . signal } ) ; } this . server . emit ( 'message' , messages ) ; } } public async destroy ( ) { this . _abortController . abort ( ) ; this . server . close ( ) ; for ( const p of this . peers ) { p . disconnect ( ) ; } } public static async createFactory ( hostname , port ) { return ( handler ) => new SocketIOMessageService ( handler , hostname , port ) ; } }