import debug from 'debug' ; import { Source } from '../component' ; import { Readable , Writable } from 'stream' ; import { MessageType } from '../message' ; import { openWebSocket , WSConfig } from './openwebsocket' ; const CLOSE_GOING_AWAY = 1001 ; export class WSSource extends Source { public onServerClose ? : ( ) => void ; param socket constructor ( socket ) { if ( socket === undefined ) { throw new Error ( 'socket argument missing' ) ; } type { Readable } const incoming = new Readable ( { objectMode : true , read : function ( ) { } , } ) ; socket . onmessage = ( msg ) => { const buffer = Buffer . from ( msg . data ) ; if ( ! incoming . push ( { data : buffer , type : MessageType . RAW } ) ) { if ( socket . readyState === WebSocket . OPEN ) { debug ( 'msl:websocket:incoming' ) ( 'downstream frozen' ) ; socket . close ( ) ; } } } ; incoming . on ( 'error' , ( e ) => { console . warn ( 'closing socket due to incoming error' , e ) ; socket . close ( ) ; } ) ; type { Writable } const outgoing = new Writable ( { objectMode : true , write : function ( msg , encoding , callback ) { try { socket . send ( msg . data ) ; } catch ( e ) { console . warn ( 'message lost during send:' , msg ) ; } callback ( ) ; } , } ) ; outgoing . on ( 'error' , ( e ) => { console . warn ( 'error during websocket send, ignoring:' , e ) ; } ) ; outgoing . on ( 'finish' , ( ) => { debug ( 'msl:websocket:outgoing' ) ( 'finish' ) ; if ( socket . readyState !== WebSocket . CLOSED ) { socket . close ( ) ; } } ) ; param e param e . code socket . onclose = ( e ) => { debug ( 'msl:websocket:close' ) ( `${ e . code }` ) ; if ( e . code === CLOSE_GOING_AWAY ) { this . onServerClose && this . onServerClose ( ) ; } incoming . push ( null ) ; outgoing . end ( ) ; } ; /**
         * initialize the component.
         */ super ( incoming , outgoing ) ; } /**
     * Expose websocket opener as a class method that returns a promise which
     * resolves with a new WebSocketComponent.
     */ static async open ( config ? ) { return await openWebSocket ( config ) . then ( ( socket ) => new WSSource ( socket ) ) ; } }