import { Duplex } from 'stream' ; import { getLogs } from '../log' ; import Dockerode from 'dockerode' ; import { ContainerWorkNode } from '../planner/work-node' ; import { WorkNodeConsoleLogLevel } from '../planner/work-node-status' ; export async function awaitStream ( node , docker , stream ) { return new Promise < void > ( ( resolve , reject ) => { function writeLog ( leve ) { return ( buffer ) => { for ( const log of getLogs ( buffer ) ) { node . status . console . write ( 'process' , leve , log . endsWith ( '\n' ) ? log . substr ( 0 , log . length - 1 ) : log ) ; } } ; } stream . on ( 'error' , reject ) ; stream . on ( 'end' , resolve ) ; stream . on ( 'close' , resolve ) ; demuxStream ( stream , writeLog ( 'info' ) , writeLog ( 'error' ) ) ; } ) ; } function demuxStream ( stream , stdoutFn , stderrFn ) { let nextDataType = null ; let nextDataLength = null ; let buffer = Buffer . from ( '' ) ; function processData ( data ? ) { if ( data ) { buffer = Buffer . concat ( [ buffer , data ] ) ; } if ( nextDataType && nextDataLength ) { if ( buffer . length >= nextDataLength ) { const content = bufferSlice ( nextDataLength ) ; if ( nextDataType === 1 ) { stdoutFn ( content ) ; } else { stderrFn ( content ) ; } nextDataType = null ; processData ( ) ; } } else { if ( buffer . length >= 8 ) { const header = bufferSlice ( 8 ) ; nextDataType = header . readUInt8 ( 0 ) ; nextDataLength = header . readUInt32BE ( 4 ) ; processData ( ) ; } } } function bufferSlice ( end ) { const out = buffer . slice ( 0 , end ) ; buffer = Buffer . from ( buffer . slice ( end , buffer . length ) ) ; return out ; } stream . on ( 'data' , processData ) ; }