package Ġfr . mat k . service . process Ċ Ċ import Ġfr . mat k . utils . Logger Delegate Ċ import Ġio . reactivex . Observable Ċ import Ġio . reactivex . functions . Consumer Ċ import Ġio . reactivex . functions . Function Ċ import Ġorg . slf 4 j . Logger Ċ import Ġjava . io . File Ċ import Ġjava . io . IOException Ċ import Ġjava . util . concurrent . Callable Ċ Ċ object ĠPro cesses Ġ{ ĊĊĠĠĠ Ġprivate Ġval Ġlogger : ĠLogger Ġby ĠLogger Delegate () ĊĊĠĠĠ Ġfun Ġexecute ( ĊĠĠĠĠĠĠĠ Ġvar arg Ġcommand And Args : ĠString , ĊĠĠĠĠĠĠĠ Ġenv : ĠMap < String , ĠString > Ġ= Ġempty Map (), ĊĠĠĠĠĠĠĠ Ġdirectory : ĠFile ? Ġ= Ġnull , ĊĠĠĠĠĠĠĠ Ġverbose : ĠBoolean Ġ= Ġfalse , ĊĠĠĠĠĠĠĠ Ġvalid Exit Codes : ĠList < Int > Ġ= ĠlistOf ( 0 ) ĊĠĠĠ Ġ): ĠObservable < String > Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġexecute ( command And Args . asList (), Ġenv , Ġdirectory , Ġverbose , Ġvalid Exit Codes ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġexecute ( ĊĠĠĠĠĠĠĠ Ġcommand And Args : ĠList < String >, ĊĠĠĠĠĠĠĠ Ġenv : ĠMap < String , ĠString > Ġ= Ġempty Map (), ĊĠĠĠĠĠĠĠ Ġdirectory : ĠFile ? Ġ= Ġnull , ĊĠĠĠĠĠĠĠ Ġverbose : ĠBoolean Ġ= Ġfalse , ĊĠĠĠĠĠĠĠ Ġvalid Exit Codes : ĠList < Int > Ġ= ĠlistOf ( 0 ) ĊĠĠĠ Ġ): ĠObservable < String > Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġresource Factory Ġ= ĠCallable < Process > Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . trace (" Execute Ġprocess Ġ{}", Ġcommand And Args ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġb Ġ= ĠProcess Builder ( command And Args ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġb . environment (). put All ( env ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( directory Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġb . directory ( directory ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġb . redirect Error ( b . redirect Input ()) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġb . start () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠIOException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠStart Process Exception (" Unable Ġto Ġstart Ġprocess ", Ġe ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġval Ġfactory Ġ= ĠFunction < Process , ĠObservable < String >> Ġ{ Ġprocess Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġoutput Ġ= ĠObservable . create < String > Ġ{ Ġemitter Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġprocess . inputStream . buffered Reader (). use Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġit . forEach Line Ġ{ Ġline Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( emitter . is Disposed ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠInter rupted Process Exception () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( verbose ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . trace ("[ {}] Ġ{}", Ġprocess . pid (), Ġline ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġemitter . onNext ( line ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġemitter . onComplete () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( e Ġ! is ĠInter rupted Process Exception ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġemitter . onError ( e ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcompletion Ġ= ĠObservable . create < String > Ġ{ Ġemitter Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġexit Code Ġ= Ġprocess . waitFor () ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( verbose ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . trace ("[ {}] ĠExit Ġ{}", Ġprocess . pid (), Ġexit Code ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( valid Exit Codes . contains ( exit Code )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġemitter . onComplete () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġemitter . onError ( Exec Process Exception ( exit Code , Ġnull )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġoutput . concat With ( completion ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġval Ġdispose Action Ġ= ĠConsumer < Process > Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġit . destroy () ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġreturn ĠObservable . using < String , ĠProcess >( resource Factory , Ġfactory , Ġdispose Action ) ĊĠĠĠ Ġ} Ċ Ċ } Ċ