import { Callback , RPCResponse , RPCMessage , toRPC } from './common' ; import { IFuture , future } from 'fp-future' ; export interface IWebSocket { /**
     * Closes the WebSocket connection, optionally using code as the the WebSocket connection close code and reason as the the WebSocket connection close reason.
     */ close ( code ? , reason ? ) : void ; /**
     * Transmits data using the WebSocket connection. data can be a string, a Blob, an ArrayBuffer, or an ArrayBufferView.
     */ send ( data ) : void ; onclose : ( ( this : this , ev : any ) => any ) | null ; onerror : ( ( this : this , ev : any ) => any ) | null ; onmessage : ( ( this : this , ev : any ) => any ) | null ; onopen : ( ( this : this , ev : any ) => any ) | null ; } export type WebSocketProviderOptions = { /**
     * WebSocketConstructor, used in Node.js where WebSocket is not globally available
     */ WebSocketConstructor ? : any ; timeout ? : number ; protocol ? : string ; } ; public export class WebSocketProvider < T extends IWebSocket > { isDisposed = false ; responseCallbacks = new Map < number , IFuture < any > > ( ) ; notificationCallbacks = new Set < Callback > ( ) ; connection ! : IFuture < T > ; debug = false ; private lastChunk : string = '' ; private lastChunkTimeout : any ; constructor ( public url , public options = { } ) { this . connect ( ) ; } dispose ( ) { this . isDisposed = true ; const connection = this . connection ; this . timeout ( new Error ( 'Provider disposed.' ) ) ; connection . then ( ( $ ) => $ . close ( ) ) ; } send ( ) { throw new Error ( 'Sync requests are deprecated' ) ; } sendAsync ( payload , callback ) { const toSend = [ ] ; let didFinish ; if ( payload instanceof Array ) { didFinish = Promise . all ( payload . map ( ( $ ) => { const defer = future < any > ( ) ; try { const message = toRPC ( $ ) ; toSend . push ( message ) ; this . responseCallbacks . set ( message . id , defer ) ; } catch ( e ) { defer . reject ( e ) ; } return defer ; } ) ) ; } else { const defer = future < any > ( ) ; try { const message = toRPC ( payload ) ; toSend . push ( message ) ; this . responseCallbacks . set ( message . id , defer ) ; } catch ( e ) { defer . reject ( e ) ; } didFinish = defer ; } didFinish . then ( ( $ ) => callback ( null , $ ) , ( err ) => callback ( err ) ) ; this . connection . then ( ( ws ) => { toSend . forEach ( ( $ ) => { const s = JSON . stringify ( $ ) ; if ( this . debug ) console . log ( 'SEND >> ' + s ) ; ws . send ( s ) ; } ) ; } , ( err ) => { callback ( err ) ; } ) ; } /**
     * Will parse the response and make an array out of it.
     */ private parseResponse ( data ) { let returnValues = [ ] ; let dechunkedData = data . replace ( /\}[\n\r]?\{/g , '}|--|{' ) . replace ( /\}\][\n\r]?\[\{/g , '}]|--|[{' ) . replace ( /\}[\n\r]?\[\{/g , '}|--|[{' ) . replace ( /\}\][\n\r]?\{/g , '}]|--|{' ) . split ( '|--|' ) ; dechunkedData . forEach ( ( chunk ) => { let data = chunk ; if ( this . lastChunk ) { data = this . lastChunk + data ; } let result = null ; try { result = JSON . parse ( data ) ; } catch ( e ) { this . lastChunk = data ; clearTimeout ( this . lastChunkTimeout ) ; this . lastChunkTimeout = setTimeout ( ( ) => { this . timeout ( ) ; } , 1000 * 15 ) ; return ; } clearTimeout ( this . lastChunkTimeout ) ; this . lastChunk = '' ; if ( result ) returnValues . push ( result ) ; } ) ; return returnValues ; } private processMessage ( message ) { if ( 'id' in message ) { const id = message . id ; const defer = this . responseCallbacks . get ( id ) ; if ( ! defer ) { console . error ( 'Error: Received a response for an unknown request' , message ) ; return ; } this . responseCallbacks . delete ( id ) ; if ( 'error' in message ) { defer . reject ( Object . assign ( new Error ( message . error . message || message . error ) , message . error ) ) ; } else if ( 'result' in message ) { defer . resolve ( message ) ; } } else { this . notificationCallbacks . forEach ( ( $ ) => $ ( null , message ) ) ; } } /**
     * Timeout all requests when the end/error event is fired
     */ private timeout ( error ? ) { if ( ! this . connection || ! this . connection . isPending ) { this . connection = future < T > ( ) ; } const timeoutError = error || new Error ( 'Connection timeout' ) ; this . responseCallbacks . forEach ( ( $ ) => $ . reject ( timeoutError ) ) ; this . responseCallbacks . clear ( ) ; if ( ! this . isDisposed ) { this . connect ( ) ; } } private connect ( ) { if ( this . connection && ! this . connection . isPending ) { this . connection . then ( ( $ ) => $ . close ( ) ) ; } if ( ! this . connection || ! this . connection . isPending ) { this . connection = future < T > ( ) ; } this . lastChunk = '' ; let ctor = this . options . WebSocketConstructor || ( typeof WebSocket !== 'undefined' ? WebSocket : void 0 ) ; if ( ! ctor ) { throw new Error ( 'Please provide a WebSocketConstructor' ) ; } const connection = new ctor ( this . url , this . options . protocol ) ; connection . onopen = ( ) => { this . connection . resolve ( connection ) ; } ; connection . onerror = ( error ) => { const theError = new Error ( 'Error in web socket' ) ; ( theError as any ) . data = error ; this . timeout ( theError ) ; } ; connection . onclose = ( event ) => { this . timeout ( new Error ( `Connection closed (${ ( event && event . reason ) || 'Unknown reason' })` ) ) ; } ; connection . onmessage = ( e ) => { let data = typeof e . data === 'string' ? e . data : '' ; if ( this . debug ) console . log ( 'RECV << ' + e . data ) ; this . parseResponse ( data ) . forEach ( ( result ) => { if ( result instanceof Array ) { result . forEach ( ( $ ) => this . processMessage ( $ ) ) ; } else { this . processMessage ( result ) ; } } ) ; } ; } } export default WebSocketProvider ;