using System.Diagnostics; using System.Reactive.Concurrency; using System.Reactive.Linq; using System.Reactive.Subjects; using InnovEnergy.Lib.SysTools.Utils; using static System.ConsoleColor; namespace InnovEnergy.Lib.SysTools.Process; using Env = Dictionary; [Obsolete("Use CliWrap instead")] public class AsyncProcess { private readonly Subject _StandardIn; private readonly Subject _StandardOut; private readonly Subject _StandardErr; private readonly ReplaySubject _ExitCode; public SysCommand Command { get; } public Env Environment { get; } public IObserver StandardIn => _StandardIn; public IObservable StandardOut => _StandardOut; public IObservable StandardErr => _StandardErr; public IObservable ExitCode => _ExitCode; public Boolean HasStarted => Process != null; public Boolean HasFinished => HasStarted && Process.HasExited; private System.Diagnostics.Process Process { get; set; } public AsyncProcess(SysCommand command, Env environment = null, Boolean logToConsole = false) { Command = command; Environment = environment; _StandardIn = new Subject(); _StandardOut = new Subject(); _StandardErr = new Subject(); _ExitCode = new ReplaySubject(); if (logToConsole) LogToConsole(); } private void LogToConsole() { _StandardOut.Subscribe(d => d.WriteLine()); _StandardErr.Subscribe(d => d.WriteLine(Red)); _ExitCode.Subscribe(ec => { "ExitCode: ".Write(Cyan); ec.WriteLine(ec == 0 ? Green : Red); }); } public Int32 WaitForExit() { EnsureStarted(); return ExitCode.Wait(); //Process.WaitForExit(); // TODO //return Process.ExitCode; } private void EnsureNotStarted() { if (HasStarted) throw new Exception("Process has already started"); } private void EnsureStarted() { if (!HasStarted) throw new Exception("Process has not yet started"); } public AsyncProcess Run() { EnsureNotStarted(); Process = InitProcess(); return this; } private System.Diagnostics.Process InitProcess() { var startInfo = new ProcessStartInfo { RedirectStandardOutput = true, RedirectStandardError = true, RedirectStandardInput = true, UseShellExecute = false, CreateNoWindow = true, Arguments = Command.Args, FileName = Command.Path, }; foreach (var ev in Environment.EmptyIfNull()) startInfo.EnvironmentVariables[ev.Key] = ev.Value; var proc = new System.Diagnostics.Process { StartInfo = startInfo, EnableRaisingEvents = true }; proc.Start(); // must start BEFORE we access Streams Observable.Repeat(proc.StandardOutput, TaskPoolScheduler.Default) .Select(s => s.ReadLine()) .TakeWhile(l => l != null) .Subscribe(_StandardOut); Observable.Repeat(proc.StandardError, TaskPoolScheduler.Default) .Select(s => s.ReadLine()) .TakeWhile(l => l != null) .Subscribe(_StandardErr); _StandardIn.Subscribe(onNext: proc.StandardInput.WriteLine, onError: _ => proc.StandardInput.Close(), onCompleted: proc.StandardInput.Close); Observable.FromEventPattern(e => proc.Exited += e, e => proc.Exited -= e) .Take(1) .Do(_=>Console.WriteLine("Exited")) .OfType() .Concat(StandardOut.IgnoreElements()) // make sure std streams finish first .Concat(StandardErr.IgnoreElements()) .Select(_ => Process.ExitCode) .Subscribe(_ExitCode); _ExitCode.Subscribe(_ => { proc.StandardInput.Close(); proc.StandardOutput.Close(); proc.StandardError.Close(); }); return proc; } public AsyncProcess Terminate() { try { Process.Kill(); } catch { // ignored } return this; } public AsyncProcess Kill() { EnsureStarted(); try { Process.Kill(); } catch { // ignored } return this; } public IDisposable ThrowOnStdErr() { return StandardErr.Subscribe(e => throw new Exception(e)); } public override String ToString() => Command.ToString(); // TODO: env }