import { Sink } from '../component' ; import { Readable , Writable } from 'stream' ; import { MessageType } from '../message' ; /**
 * The socket used here is a ws socket returned by
 * a ws Server's 'connection' event.
 */ export class WSSink extends Sink { constructor ( socket ) { const outgoing = new Readable ( { objectMode : true , read : ( ) => { } , } ) ; const incoming = new Writable ( { objectMode : true , write : ( msg , encoding , callback ) => { try { socket . send ( msg . data ) ; } catch ( e ) { console . warn ( 'message lost during send:' , msg ) ; } callback ( ) ; } , } ) ; socket . on ( 'message' , function ( data ) { outgoing . push ( { data , type : MessageType . RAW } ) ; } ) ; socket . on ( 'close' , function ( ) { outgoing . push ( null ) ; } ) ; socket . on ( 'error' , ( e ) => { console . error ( 'WebSocket error:' , e ) ; socket . terminate ( ) ; outgoing . push ( null ) ; } ) ; incoming . on ( 'error' , ( e ) => { console . log ( 'closing WebSocket due to incoming error' , e ) ; socket && socket . close && socket . close ( ) ; } ) ; incoming . on ( 'finish' , ( ) => { socket && socket . close && socket . close ( ) ; } ) ; outgoing . on ( 'error' , ( e ) => { console . warn ( 'error during WebSocket send, ignoring:' , e ) ; } ) ; outgoing . on ( 'finish' , ( ) => { socket && socket . close && socket . close ( ) ; } ) ; /**
         * initialize the component.
         */ super ( incoming , outgoing ) ; } }