/**
 * Copyright 2018 Twitter, Inc.
 * Licensed under the Apache License, Version 2.0
 * http://www.apache.org/licenses/LICENSE-2.0
 */ import Semaphore from "sqrl/lib/jslib/Semaphore" ; import * as split2 from "split2" ; import { CliActionOutput } from "./CliOutput" ; import { Context } from "sqrl/lib/api/ctx" ; import { promiseFinally , SqrlObject } from "sqrl-common" ; import { FeatureMap , Executable , Execution } from "sqrl" ; import { CliManipulator } from "sqrl-cli-functions" ; export class CliRun { constructor ( private executable , private output ) { } triggerRecompile ( compileCallback ) { this . output . sourceRecompiling ( ) ; return compileCallback ( ) . then ( ( rv ) => { this . executable = rv ; this . output . sourceUpdated ( ) ; } ) . catch ( ( err ) => { this . output . sourceRecompileError ( err ) ; } ) ; } async action ( trc , inputs , features ) { const manipulator = new CliManipulator ( ) ; const execution = await this . executable . execute ( trc , { featureTimeoutMs : 10000 , inputs , manipulator , } ) ; const loggedFeatures = { } ; await Promise . all ( features . map ( async ( featureName ) => { const value = await execution . fetchFeature ( featureName ) ; loggedFeatures [ featureName ] = value ; } ) ) ; await execution . fetchFeature ( "SqrlExecutionComplete" ) ; await manipulator . mutate ( trc ) ; this . output . action ( manipulator , execution , loggedFeatures ) ; } async stream ( options ) { const { concurrency } = options ; const stream = process . stdin . pipe ( split2 ( ) ) ; const busy = new Semaphore ( ) ; stream . on ( "data" , ( line ) => { const lineValues = Object . assign ( { } , options . inputs ) ; try { lineValues [ options . streamFeature ] = JSON . parse ( line . toString ( "utf-8" ) ) ; } catch ( err ) { console . error ( "Error: Invalid JSON value: %s" , JSON . stringify ( line ) + err . toString ( ) ) ; return ; } promiseFinally ( busy . wrap ( this . action ( options . ctx , lineValues , options . features ) . catch ( ( err ) => { console . error ( "Error: " + err . toString ( ) ) ; } ) ) , ( ) => { stream . resume ( ) ; } ) ; if ( concurrency && busy . getCount ( ) === concurrency ) { stream . pause ( ) ; } } ) ; await new Promise ( ( resolve , reject ) => { stream . on ( "end" , ( ) => resolve ( ) ) ; stream . on ( "error" , ( err ) => reject ( err ) ) ; } ) ; await busy . waitForZero ( ) ; } close ( ) { this . output . close ( ) ; } }