import { IncomingMessage } from 'http' ; import { Observable , Observer , fromEventPattern } from 'rxjs' ; import { Socket , createServer , Server , NetConnectOpts , connect as netConnect , } from 'net' ; import WebSocket from 'ws' ; import { takeUntil , last } from 'rxjs/operators' ; param request param protocol export function requestUrl ( request , protocol ) { const path = request . url ; const host = request . headers . host ; return new URL ( path , `${ protocol }://${ host }` ) ; } param options export function connect ( options ) { return Observable . create ( ( observer ) => { const tcp = netConnect ( options , ( ) => { observer . next ( tcp ) ; } ) ; tcp . on ( 'close' , ( ) => { observer . complete ( ) ; } ) ; tcp . on ( 'error' , err => { observer . error ( err ) ; } ) ; return ( ) => { tcp . destroy ( ) ; } ; } ) ; } param server param port export function listen ( server , port ? ) { return Observable . create ( ( observer ) => { server . on ( 'connection' , socket => { observer . next ( socket ) ; } ) ; server . on ( 'error' , err => { server . close ( ) ; observer . error ( err ) ; } ) ; server . listen ( port ) ; return ( ) => { server . close ( ) ; } ; } ) ; } param connection export function cleanup ( socket ) { return Observable . create ( ( observer ) => { socket . on ( 'close' , ( ) => { observer . complete ( ) ; } ) ; socket . on ( 'error' , err => { socket . destroy ( ) ; observer . error ( err ) ; } ) ; observer . next ( socket ) ; return ( ) => { socket . destroy ( ) ; } ; } ) ; } param address param options export function websocket ( address , options ? ) { return Observable . create ( ( observer ) => { const socket = new WebSocket ( address , options ) ; socket . on ( 'open' , ( ) => { observer . next ( socket ) ; } ) ; socket . on ( 'close' , ( ) => { observer . complete ( ) ; } ) ; socket . on ( 'error' , err => { observer . error ( err ) ; } ) ; return ( ) => { socket . close ( ) ; } ; } ) ; } export function handleUpgrade ( server , req , socket , head ) { return Observable . create ( ( observer ) => { let websocket ; let closed = false ; server . handleUpgrade ( req , socket , head , ws => { websocket = ws ; if ( closed ) { ws . close ( ) ; return ; } observer . next ( ws ) ; ws . on ( 'close' , ( ) => { observer . complete ( ) ; } ) ; ws . on ( 'error' , err => { observer . error ( err ) ; } ) ; } ) ; return ( ) => { closed = true ; if ( websocket ) { websocket . close ( ) ; } } ; } ) ; } param websocket export function messages ( websocket ) { return fromEventPattern ( h => websocket . on ( 'message' , h ) , h => websocket . off ( 'message' , h ) ) ; } export function completeWith < T > ( observable ) { return takeUntil < T > ( observable . pipe ( last ( ) ) ) ; }