import { Readable , Transform , Writable } from 'stream' ; export default class StreamFactory { param fn public static consumer ( fn = ( ) => { } ) { return new Writable ( { objectMode : true , write ( msg , _encoding , callback ) { fn ( msg ) ; callback ( ) ; } , } ) ; } public static peeker ( fn ) { if ( typeof fn !== 'function' ) { throw new Error ( 'you must supply a function' ) ; } return new Transform ( { objectMode : true , transform ( msg , _encoding , callback ) { fn ( msg ) ; callback ( undefined , msg ) ; } , } ) ; } param arr public static producer ( messages ? ) { let counter = 0 ; return new Readable ( { objectMode : true , read ( ) { if ( messages !== undefined ) { if ( counter < messages . length ) { this . push ( messages [ counter ++ ] ) ; } else { this . push ( null ) ; } } } , } ) ; } public static recorder ( type , fileStream ) { return new Transform ( { objectMode : true , transform ( msg , encoding , callback ) { const timestamp = Date . now ( ) ; const message = Object . assign ( { } , msg , { data : msg . data . toString ( 'base64' ) , } ) ; fileStream . write ( JSON . stringify ( { type , timestamp , message } , null , 2 ) ) ; fileStream . write ( ',\n' ) ; callback ( undefined , msg ) ; } , } ) ; } return { Generator } public static replayer ( packets ) { let packetCounter = 0 ; let lastTimestamp = packets [ 0 ] . timestamp ; return new Readable ( { objectMode : true , read ( ) { const packet = packets [ packetCounter ++ ] ; if ( packet ) { const { type , timestamp , message } = packet ; const delay = timestamp - lastTimestamp ; lastTimestamp = timestamp ; if ( message ) { const data = message . data ? Buffer . from ( message . data , 'base64' ) : Buffer . alloc ( 0 ) ; const msg = Object . assign ( { } , message , { data } ) ; this . push ( { type , delay , msg } ) ; } else { this . push ( { type , delay , msg : null } ) ; } } else { this . push ( null ) ; } } , } ) ; } }