import { Observable , Observer } from 'rxjs' ; export const streamToObservable = < T = Buffer > ( getStream ) : Observable < T > => new Observable ( ( observer ) => { const stream = getStream ( ) ; let done = false ; const cleanup = ( ) => { if ( ! done ) { done = true ; stream . removeListener ( 'end' , onEnd ) ; stream . removeListener ( 'error' , onError ) ; stream . removeListener ( 'data' , onData ) ; } } ; const onEnd = ( ) => { cleanup ( ) ; observer . complete ( ) ; } ; const onError = ( error ) => { cleanup ( ) ; observer . error ( error ) ; } ; const onData = ( data ) => observer . next ( data ) ; stream . once ( 'error' , onError ) ; stream . once ( 'end' , onEnd ) ; stream . on ( 'data' , onData ) ; return cleanup ; } ) ;