import { Readable } from 'stream' ; import { listOnePageLogResources } from '../api/logs' ; import { TwilioServerlessApiClient } from '../client' ; import { Sid } from '../types' ; import { LogsConfig } from '../types/logs' ; export class LogsStream extends Readable { private _pollingFrequency : number ; private _pollingCacheSize : number ; private _interval : NodeJS . Timeout | undefined ; private _viewedSids : Set < Sid > ; private _viewedLogs : Array < { sid : Sid ; dateCreated : Date ; } > ; constructor ( private environmentSid , private serviceSid , private client , private config ) { super ( { objectMode : true } ) ; this . _interval = undefined ; this . _viewedSids = new Set ( ) ; this . _viewedLogs = [ ] ; this . _pollingFrequency = config . pollingFrequency || 1000 ; this . _pollingCacheSize = config . logCacheSize || 1000 ; } set pollingFrequency ( frequency ) { this . _pollingFrequency = frequency ; if ( this . config . tail && this . _interval ) { clearInterval ( this . _interval ) ; this . _interval = setInterval ( ( ) => { this . _poll ( ) ; } , this . _pollingFrequency ) ; } } async _poll ( ) { try { const logs = await listOnePageLogResources ( this . environmentSid , this . serviceSid , this . client , { functionSid : this . config . filterByFunction , pageSize : this . config . limit , } ) ; logs . filter ( log => ! this . _viewedSids . has ( log . sid ) ) . reverse ( ) . forEach ( log => { this . push ( log ) ; } ) ; const viewedLogsSet = new Set ( [ ... this . _viewedLogs . map ( log => `${ log . sid }-${ log . dateCreated . toISOString ( ) }` ) , ... logs . map ( log => `${ log . sid }-${ log . date_created }` ) , ] ) ; this . _viewedLogs = [ ... viewedLogsSet ] . map ( logString => { const [ sid , dateCreated ] = logString . split ( '-' ) ; return { sid , dateCreated : new Date ( dateCreated ) } ; } ) . sort ( ( a , b ) => b . dateCreated . valueOf ( ) - a . dateCreated . valueOf ( ) ) . slice ( 0 , this . _pollingCacheSize ) ; this . _viewedSids = new Set ( this . _viewedLogs . map ( log => log . sid ) ) ; if ( ! this . config . tail ) { this . push ( null ) ; } } catch ( err ) { this . destroy ( err ) ; } } _read ( ) { if ( this . config . tail ) { if ( ! this . _interval ) { this . _interval = setInterval ( ( ) => { this . _poll ( ) ; } , this . _pollingFrequency ) ; } } else { this . _poll ( ) ; } } _destroy ( ) { if ( this . _interval ) { clearInterval ( this . _interval ) ; this . _interval = undefined ; } } }