import { isFunction , isString } from "lodash" ; import { Writable } from "stream" ; import { format } from "util" ; import { Constructor } from "../common_types" ; import { LocalStore , Logger , Message , MessageLogger , MessageStore , MessageType } from "./common" ; export interface MessageStreamServerOptions { parent ? : MessageStreamServer ; outStream ? : Writable ; store ? : MessageStore ; interceptStdio ? : boolean ; } export type WriteCallback = ( error : Error | null | undefined ) => void ; const intercepted = new Map < Writable , Writable [ "write" ] > ( ) ; export class MessageStreamServer implements MessageLogger { readonly from : string ; readonly outStream : Writable ; readonly isMessageLogger : true = true ; protected store : MessageStore ; protected write : Writable [ "write" ] ; protected intercepting = false ; constructor ( id , options = { } ) { const outStream = options . outStream || ( options . parent && options . parent . outStream ) ; if ( ! outStream ) { throw new Error ( `MessageStreamServer: either parent or outStream must be specified` ) ; } this . outStream = outStream ; this . store = options . store || ( options . parent && options . parent . store ) || new LocalStore ( ) ; this . from = options . parent ? `${ options . parent . from }:${ id }` : id ; this . write = options . parent ? options . parent . write : outStream . write . bind ( outStream ) ; if ( options . interceptStdio ) { this . intercept ( process . stdout , MessageType . stdout ) ; this . intercept ( process . stderr , MessageType . stderr ) ; this . intercepting = true ; } } get messages ( ) { return this . store . messages ; } get summary ( ) { return this . store . summary ; } end = ( ) => { this . outStream . end ( ) ; } ; info : Logger = ( arg , ... args ) => { this . log ( MessageType . info , arg , ... args ) ; } ; warning : Logger = ( arg , ... args ) => { this . log ( MessageType . warning , arg , ... args ) ; } ; error : Logger = ( arg , ... args ) => { this . log ( MessageType . error , arg , ... args ) ; } ; log = ( type , arg , ... args ) => { const m = { type , timestamp : Date . now ( ) , from : this . from , content : format ( arg , ... args ) , } ; this . message ( m ) ; } ; append ( toAppend ) { for ( const m of toAppend ) { this . message ( m ) ; } } message = ( msg , cb ? ) : boolean => { this . store . store ( msg ) ; return this . write ( JSON . stringify ( msg ) + "\n" , cb ) ; } ; createChild ( id ) { return new ( this . constructor as Constructor < this > ) ( id , { parent : this , interceptConsole : false , } ) ; } stopIntercept ( ) { if ( ! this . intercepting || intercepted . size === 0 ) return ; intercepted . forEach ( ( origWrite , stream ) => { stream . write = origWrite ; } ) ; intercepted . clear ( ) ; this . intercepting = false ; } private intercept ( toIntercept , type ) { if ( intercepted . has ( toIntercept ) ) { throw new Error ( `Only one stdio interceptor can be used at a time` ) ; } intercepted . set ( toIntercept , toIntercept . write ) ; toIntercept . write = ( chunk , encodingOrCb ? , cb ? ) : boolean => { if ( isFunction ( encodingOrCb ) ) cb = encodingOrCb ; if ( ! isString ( chunk ) ) chunk = chunk . toString ( ) ; return this . message ( { type , timestamp : Date . now ( ) , from : this . from , content : chunk , } , cb ) ; } ; } }