import { Observable } from 'rxjs' ; import { Logger } from 'src/core/server' ; import { Stream , PassThrough } from 'stream' ; const delimiter = '\n' ; export const createNDJSONStream = < Response > ( results , logger ) : Stream => { const stream = new PassThrough ( ) ; results . subscribe ( { next : ( message ) => { try { const line = JSON . stringify ( message ) ; stream . write ( `${ line }${ delimiter }` ) ; } catch ( error ) { logger . error ( 'Could not serialize or stream a message.' ) ; logger . error ( error ) ; } } , error : ( error ) => { stream . end ( ) ; logger . error ( error ) ; } , complete : ( ) => stream . end ( ) , } ) ; return stream ; } ;