import { Source } from '../component' ; import { Readable , Writable } from 'stream' ; import { connect , Socket } from 'net' ; import { MessageType } from '../message' ; export class TcpSource extends Source { param host constructor ( host ? ) { let socket ; /**
         * Set up an incoming stream and attach it to the socket.
         */ const incoming = new Readable ( { objectMode : true , read : function ( ) { } , } ) ; /**
         * Set up outgoing stream and attach it to the socket.
         */ const outgoing = new Writable ( { objectMode : true , write : function ( msg , encoding , callback ) { const b = msg . data ; if ( ! socket ) { let url ; if ( host === undefined ) { const firstSpace = b . indexOf ( ' ' ) ; const secondSpace = b . indexOf ( ' ' , firstSpace + 1 ) ; url = b . slice ( firstSpace , secondSpace ) . toString ( 'ascii' ) ; } else { url = `rtsp://${ host }` ; } const { hostname , port } = new URL ( url ) ; socket = connect ( Number ( port ) || 554 , hostname === null ? undefined : hostname ) ; socket . on ( 'error' , ( e ) => { console . error ( 'TCP socket error:' , e ) ; socket . destroy ( ) ; incoming . push ( null ) ; } ) ; socket . on ( 'data' , ( buffer ) => { if ( ! incoming . push ( { data : buffer , type : MessageType . RAW } ) ) { console . warn ( 'TCP Component internal error: not allowed to push more data' ) ; } } ) ; socket . on ( 'end' , ( ) => { console . warn ( 'socket ended' ) ; incoming . push ( null ) ; } ) ; } try { socket . write ( msg . data , encoding , callback ) ; } catch ( e ) { console . warn ( 'message lost during send:' , msg ) ; } } , } ) ; incoming . on ( 'error' , ( e ) => { console . log ( 'closing TCP socket due to incoming error' , e ) ; socket && socket . end ( ) ; } ) ; incoming . on ( 'finish' , ( ) => { socket && socket . end ( ) ; } ) ; outgoing . on ( 'error' , ( e ) => { console . warn ( 'error during TCP send, ignoring:' , e ) ; } ) ; outgoing . on ( 'finish' , ( ) => { socket && socket . end ( ) ; } ) ; /**
         * initialize the component.
         */ super ( incoming , outgoing ) ; } }