From b9d2918a436e1ed92ce1345ded9bca40e2ce8c94 Mon Sep 17 00:00:00 2001
From: Aaron Roney <twitchax@gmail.com>
Date: Wed, 13 Mar 2019 23:40:29 -0700
Subject: [PATCH] Add reactive support.

---
 README.md                                     |  2 +-
 src/Core/Helpers.cs                           | 28 +++++--
 src/Core/Implementations/CommandEvent.cs      | 35 +++++++++
 src/Core/Implementations/CommandResult.cs     |  2 +-
 .../Implementations/Executables/Executable.cs |  8 ++
 .../Implementations/ObservableCommandEvent.cs | 75 +++++++++++++++++++
 src/Core/Implementations/Shells/Shell.cs      | 35 ++++++++-
 src/Core/Models/ICommandEvent.cs              | 37 +++++++++
 src/Core/Models/IExecutable.cs                | 14 ++++
 src/Core/Models/IShell.cs                     | 14 ++++
 src/Core/Sheller.cs                           |  1 +
 src/Core/Sheller.csproj                       |  2 +-
 src/Test/BasicTests.cs                        | 44 +++++++++++
 src/Test/Sheller.Tests.csproj                 |  1 +
 14 files changed, 285 insertions(+), 13 deletions(-)
 create mode 100644 src/Core/Implementations/CommandEvent.cs
 create mode 100644 src/Core/Implementations/ObservableCommandEvent.cs
 create mode 100644 src/Core/Models/ICommandEvent.cs

diff --git a/README.md b/README.md
index ff88aed..cd85960 100644
--- a/README.md
+++ b/README.md
@@ -34,7 +34,7 @@ Latest .NET Standard 2.0.
 
 ### Examples
 
-This library is extendable, but you can run it a few ways depending on how you have extended it.
+This library is extendable, but you can run it a few ways depending on how you have extended it.  For more examples, check out the [tests](src/Test/BasicTests.cs).
 
 With no extensions, you would run a command like this.
 
diff --git a/src/Core/Helpers.cs b/src/Core/Helpers.cs
index 1cc7bcd..d6ddf13 100644
--- a/src/Core/Helpers.cs
+++ b/src/Core/Helpers.cs
@@ -7,6 +7,7 @@
 using System.Threading;
 using System.Threading.Tasks;
 using Sheller.Implementations;
+using Sheller.Implementations.Shells;
 using Sheller.Models;
 
 namespace Sheller
@@ -78,6 +79,12 @@ internal static void CopyToStringDictionary(this IEnumerable<KeyValuePair<string
             }
         }
 
+        internal static void ForEach<T>(this IEnumerable<T> list, Action<T> functor)
+        {
+            foreach(var item in list)
+                functor(item);
+        }
+
         internal static string EscapeQuotes(this string s) => s.Replace("\"", "\\\"");
 
         internal static Task<ICommandResult> RunCommand(
@@ -86,7 +93,8 @@ internal static Task<ICommandResult> RunCommand(
             IEnumerable<string> standardInputs = null, 
             IEnumerable<Action<string>> standardOutputHandlers = null,
             IEnumerable<Action<string>> standardErrorHandlers = null,
-            Func<string, string, Task<string>> inputRequestHandler = null)
+            Func<string, string, Task<string>> inputRequestHandler = null,
+            ObservableCommandEvent observableCommandEvent = null)
         {
             var t = new Task<ICommandResult>(() => 
             {
@@ -104,22 +112,28 @@ internal static Task<ICommandResult> RunCommand(
                 
                 process.OutputDataReceived += (s, e) => 
                 {
-                    if(e.Data == null) return;
+                    var data = e.Data;
+                    if(data == null) return;
 
-                    standardOutput.AppendLine(e.Data);
+                    standardOutput.AppendLine(data);
                     if(standardOutputHandlers != null)
                         foreach(var handler in standardOutputHandlers)
-                            handler(e.Data);
+                            handler(data);
+
+                    observableCommandEvent.FireEvent(new CommandEvent(CommandEventType.StandardOutput, data));
                 };
                 
                 process.ErrorDataReceived += (s, e) => 
                 {
-                    if(e.Data == null) return;
+                    var data = e.Data;
+                    if(data == null) return;
 
-                    standardError.AppendLine(e.Data);
+                    standardError.AppendLine(data);
                     if(standardErrorHandlers != null)
                         foreach(var handler in standardErrorHandlers)
-                            handler(e.Data);
+                            handler(data);
+
+                    observableCommandEvent.FireEvent(new CommandEvent(CommandEventType.StandardError, data));
                 };
 
                 if(inputRequestHandler != null)
diff --git a/src/Core/Implementations/CommandEvent.cs b/src/Core/Implementations/CommandEvent.cs
new file mode 100644
index 0000000..f43ca2d
--- /dev/null
+++ b/src/Core/Implementations/CommandEvent.cs
@@ -0,0 +1,35 @@
+
+using System;
+using Sheller.Models;
+
+namespace Sheller.Implementations
+{
+    /// <summary>
+    /// The default result type for executables.
+    /// </summary>
+    public class CommandEvent : ICommandEvent
+    {
+        /// <summary>
+        /// StreamType property.
+        /// </summary>
+        /// <value>The stream type of the event.</value>
+        public CommandEventType Type { get; private set; }
+
+        /// <summary>
+        /// Data property.
+        /// </summary>
+        /// <value>The string data of the event.</value>
+        public string Data { get; private set; }
+
+        /// <summary>
+        /// The CommandEvent constructor.
+        /// </summary>
+        /// <param name="type"></param>
+        /// <param name="data"></param>
+        public CommandEvent(CommandEventType type, string data)
+        {
+            Type = type;
+            Data = data;
+        }
+    }
+}
\ No newline at end of file
diff --git a/src/Core/Implementations/CommandResult.cs b/src/Core/Implementations/CommandResult.cs
index da1f89f..b8e31f6 100644
--- a/src/Core/Implementations/CommandResult.cs
+++ b/src/Core/Implementations/CommandResult.cs
@@ -13,7 +13,7 @@ public class CommandResult : ICommandResult
         /// Succeeded property.
         /// </summary>
         /// <value>The succeeded status of an executable.</value>
-        public bool Succeeded { get; }
+        public bool Succeeded { get; private set; }
 
         /// <summary>
         /// ExitCode property.
diff --git a/src/Core/Implementations/Executables/Executable.cs b/src/Core/Implementations/Executables/Executable.cs
index 9f93d93..2008388 100644
--- a/src/Core/Implementations/Executables/Executable.cs
+++ b/src/Core/Implementations/Executables/Executable.cs
@@ -251,6 +251,14 @@ async Task<TResult> executionTask()
         public TExecutable UseInputRequestHandler(Func<string, string, Task<string>> inputRequestHandler) => CreateFrom(this, shell: _shell.UseInputRequestHandler(inputRequestHandler));
         IExecutable IExecutable.UseInputRequestHandler(Func<string, string, Task<string>> inputRequestHandler) => UseInputRequestHandler(inputRequestHandler);
 
+        /// <summary>
+        /// Provides an <see cref="IObservable{T}"/> to which a subscription can be placed.
+        /// The observable never completes, since executions can be run many times.
+        /// </summary>
+        /// <returns>A `new` instance of type <typeparamref name="TExecutable"/> with the subscribers attached to the observable.</returns>
+        public TExecutable WithSubscribe(Action<IObservable<ICommandEvent>> subscriber) => CreateFrom(this, shell: _shell.WithSubscribe(subscriber));
+        IExecutable IExecutable.WithSubscribe(Action<IObservable<ICommandEvent>> subscriber) => WithSubscribe(subscriber);
+
         /// <summary>
         /// Adds a wait <see cref="Func{T}"/> (of which there may be many) to the execution context and returns a `new` context instance.
         /// </summary>
diff --git a/src/Core/Implementations/ObservableCommandEvent.cs b/src/Core/Implementations/ObservableCommandEvent.cs
new file mode 100644
index 0000000..cafe465
--- /dev/null
+++ b/src/Core/Implementations/ObservableCommandEvent.cs
@@ -0,0 +1,75 @@
+using System;
+using System.Collections.Generic;
+using Sheller.Models;
+
+namespace Sheller.Implementations
+{
+    /// <summary>
+    /// Observable for command events.
+    /// </summary>
+    public class ObservableCommandEvent : IObservable<ICommandEvent>
+    {
+        internal List<IObserver<ICommandEvent>> Observers { get; private set; } = new List<IObserver<ICommandEvent>>();
+
+        /// <summary>
+        /// The Subscribe method.
+        /// </summary>
+        /// <param name="observer">The observer.</param>
+        /// <returns>An <see cref="IDisposable"/>.</returns>
+        public IDisposable Subscribe(IObserver<ICommandEvent> observer) 
+        {
+            if(!Observers.Contains(observer)) 
+                Observers.Add(observer);
+            return new Unsubscriber(Observers, observer);
+        }
+
+        internal void FireEvent(ICommandEvent commandEvent)
+        {
+            foreach(var observer in Observers)
+                observer.OnNext(commandEvent);
+        }
+
+        internal void FireError(Exception e)
+        {
+            foreach(var observer in Observers)
+                observer.OnError(e);
+        }
+
+        internal void FireCompleted()
+        {
+            foreach (var observer in Observers.ToArray())
+                if (Observers.Contains(observer))
+                    observer.OnCompleted();
+
+            Observers.Clear();
+        }
+
+        internal static ObservableCommandEvent Merge(ObservableCommandEvent ob1, ObservableCommandEvent ob2)
+        {
+            var newObservable = new ObservableCommandEvent();
+            
+            ob1.Observers.ForEach(o => newObservable.Subscribe(o));
+            ob2.Observers.ForEach(o => newObservable.Subscribe(o));
+
+            return newObservable;
+        }
+
+        private class Unsubscriber : IDisposable
+        {
+            private List<IObserver<ICommandEvent>> _observers;
+            private IObserver<ICommandEvent> _observer;
+
+            public Unsubscriber(List<IObserver<ICommandEvent>> observers, IObserver<ICommandEvent> observer)
+            {
+                this._observers = observers;
+                this._observer = observer;
+            }
+
+            public void Dispose()
+            {
+                if (_observer != null && _observers.Contains(_observer))
+                    _observers.Remove(_observer);
+            }
+        }
+    }
+}
\ No newline at end of file
diff --git a/src/Core/Implementations/Shells/Shell.cs b/src/Core/Implementations/Shells/Shell.cs
index 6ed479a..15ff306 100644
--- a/src/Core/Implementations/Shells/Shell.cs
+++ b/src/Core/Implementations/Shells/Shell.cs
@@ -35,6 +35,8 @@ namespace Sheller.Implementations.Shells
         private IEnumerable<Action<string>> _standardErrorHandlers;
         private Func<string, string, Task<String>> _inputRequestHandler;
 
+        private ObservableCommandEvent _observableCommandEvent;
+
         private bool _throws;
 
         /// <summary>
@@ -51,7 +53,7 @@ namespace Sheller.Implementations.Shells
         /// Initializes the shell.
         /// </summary>
         /// <param name="shell">The name or path of the shell.</param>
-        public virtual TShell Initialize(string shell) => Initialize(shell, null, null, null, null, null, null);
+        public virtual TShell Initialize(string shell) => Initialize(shell, null, null, null, null, null, null, null);
 
         /// <summary>
         /// Initializes the shell.
@@ -62,6 +64,7 @@ namespace Sheller.Implementations.Shells
         /// <param name="standardOutputHandlers">The standard output handlers for capture from the execution.</param>
         /// <param name="standardErrorHandlers">The standard error handlers for capture from the execution.</param>
         /// <param name="inputRequestHandler">The request handler from the execution.</param>
+        /// <param name="observableCommandEvent">The observable that fires on stdout/stderr.</param>
         /// <param name="throws">Indicates that a non-zero exit code throws.</param>
         protected virtual TShell Initialize(
             string shell, 
@@ -70,6 +73,7 @@ protected virtual TShell Initialize(
             IEnumerable<Action<string>> standardOutputHandlers,
             IEnumerable<Action<string>> standardErrorHandlers,
             Func<string, string, Task<string>> inputRequestHandler,
+            ObservableCommandEvent observableCommandEvent, 
             bool? throws)
         {
             _shell = shell;
@@ -81,6 +85,8 @@ protected virtual TShell Initialize(
             _standardErrorHandlers = standardErrorHandlers ?? new List<Action<string>>();
             _inputRequestHandler = inputRequestHandler;
 
+            _observableCommandEvent = observableCommandEvent ?? new ObservableCommandEvent();
+            
             _throws = throws ?? true;
 
             return this as TShell;
@@ -94,6 +100,7 @@ private static TShell CreateFrom(
             IEnumerable<Action<string>> standardOutputHandlers = null,
             IEnumerable<Action<string>> standardErrorHandlers = null,
             Func<string, string, Task<string>> inputRequestHandler = null,
+            ObservableCommandEvent observableCommandEvent = null, 
             bool? throws = null) =>
                 new TShell().Initialize(
                     shell ?? old._shell,
@@ -102,6 +109,7 @@ private static TShell CreateFrom(
                     standardOutputHandlers ?? old._standardOutputHandlers,
                     standardErrorHandlers ?? old._standardErrorHandlers,
                     inputRequestHandler ?? old._inputRequestHandler,
+                    observableCommandEvent ?? old._observableCommandEvent,
                     throws ?? old._throws
                 );
 
@@ -123,10 +131,18 @@ public virtual async Task<ICommandResult> ExecuteCommandAsync(string executable,
                 _standardInputs,
                 _standardOutputHandlers, 
                 _standardErrorHandlers,
-                _inputRequestHandler);
+                _inputRequestHandler,
+                _observableCommandEvent);
 
             if(_throws && result.ExitCode != 0)
-                throw new ExecutionFailedException($"The execution resulted in a non-zero exit code ({result.ExitCode}).", result);
+            {
+                var error = new ExecutionFailedException($"The execution resulted in a non-zero exit code ({result.ExitCode}).", result);
+                _observableCommandEvent.FireError(error);
+                throw error;
+            }
+
+            // TODO: Add a `UseSubscribeComplete` (or something like it) that instructs the observable to complete here.
+            //_observableCommandEvent.FireCompleted();
 
             return result;
         }
@@ -199,6 +215,19 @@ public virtual async Task<ICommandResult> ExecuteCommandAsync(string executable,
         public TShell UseInputRequestHandler(Func<string, string, Task<string>> inputRequestHandler) => CreateFrom(this, inputRequestHandler: inputRequestHandler);
         IShell IShell.UseInputRequestHandler(Func<string, string, Task<string>> inputRequestHandler) => UseInputRequestHandler(inputRequestHandler);
 
+        /// <summary>
+        /// Provides an <see cref="IObservable{T}"/> to which a subscription can be placed.
+        /// The observable never completes, since executions can be run many times.
+        /// </summary>
+        /// <returns>A `new` instance of type <typeparamref name="TShell"/> with the subscribers attached to the observable.</returns>
+        public TShell WithSubscribe(Action<IObservable<ICommandEvent>> subscriber)
+        {
+            var newObservable = new ObservableCommandEvent();
+            subscriber(newObservable);
+            return CreateFrom(this, observableCommandEvent: ObservableCommandEvent.Merge(_observableCommandEvent, newObservable));
+        }
+        IShell IShell.WithSubscribe(Action<IObservable<ICommandEvent>> subscriber) => WithSubscribe(subscriber);
+
         /// <summary>
         /// Ensures the shell context will not throw on a non-zero exit code and returns a `new` context instance.
         /// </summary>
diff --git a/src/Core/Models/ICommandEvent.cs b/src/Core/Models/ICommandEvent.cs
new file mode 100644
index 0000000..46d0ce5
--- /dev/null
+++ b/src/Core/Models/ICommandEvent.cs
@@ -0,0 +1,37 @@
+using System;
+
+namespace Sheller.Models
+{
+    /// <summary>
+    /// The default result interface for executables.
+    /// </summary>
+    public interface ICommandEvent
+    {
+        /// <summary>
+        /// StreamType property.
+        /// </summary>
+        /// <value>The stream type of the event.</value>
+        CommandEventType Type { get; }
+
+        /// <summary>
+        /// Data property.
+        /// </summary>
+        /// <value>The string data of the event.</value>
+        string Data { get; }
+    }
+
+    /// <summary>
+    /// The type (stdout, stderr) of the event data.
+    /// </summary>
+    public enum CommandEventType
+    {
+        /// <summary>
+        /// The Standard Output type.
+        /// </summary>
+        StandardOutput,
+        /// <summary>
+        /// The Standard Error type.
+        /// </summary>
+        StandardError
+    }
+}
\ No newline at end of file
diff --git a/src/Core/Models/IExecutable.cs b/src/Core/Models/IExecutable.cs
index ac3d193..ede87a5 100644
--- a/src/Core/Models/IExecutable.cs
+++ b/src/Core/Models/IExecutable.cs
@@ -84,6 +84,13 @@ public interface IExecutable
         /// <returns>A `new` instance of <see cref="IShell"/> with the request handler passed to this call.</returns>
         IExecutable UseInputRequestHandler(Func<string, string, Task<string>> inputRequestHandler);
 
+        /// <summary>
+        /// Provides an <see cref="IObservable{T}"/> to which a subscription can be placed.
+        /// The observable never completes, since executions can be run many times.
+        /// </summary>
+        /// <returns>A `new` instance of type <see cref="IExecutable"/> with the subscribers attached to the observable.</returns>
+        IExecutable WithSubscribe(Action<IObservable<ICommandEvent>> subscriber);
+
         /// <summary>
         /// Adds a wait <see cref="Func{T}"/> (of which there may be many) to the execution context and returns a `new` context instance.
         /// </summary>
@@ -166,6 +173,13 @@ public interface IExecutable<out TExecutable> : IExecutable where TExecutable :
         /// <returns>A `new` instance of <typeparamref name="TExecutable"/> with the request handler passed to this call.</returns>
         new TExecutable UseInputRequestHandler(Func<string, string, Task<string>> inputRequestHandler);
 
+        /// <summary>
+        /// Provides an <see cref="IObservable{T}"/> to which a subscription can be placed.
+        /// The observable never completes, since executions can be run many times.
+        /// </summary>
+        /// <returns>A `new` instance of type <typeparamref name="TExecutable"/> with the subscribers attached to the observable.</returns>
+        new TExecutable WithSubscribe(Action<IObservable<ICommandEvent>> subscriber);
+
         /// <summary>
         /// Adds a wait <see cref="Func{T}"/> (of which there may be many) to the execution context and returns a `new` context instance.
         /// </summary>
diff --git a/src/Core/Models/IShell.cs b/src/Core/Models/IShell.cs
index 347f2d6..f3c99ef 100644
--- a/src/Core/Models/IShell.cs
+++ b/src/Core/Models/IShell.cs
@@ -73,6 +73,13 @@ public interface IShell
         /// <returns>A `new` instance of <see cref="IShell"/> with the request handler passed to this call.</returns>
         IShell UseInputRequestHandler(Func<string, string, Task<string>> inputRequestHandler);
 
+        /// <summary>
+        /// Provides an <see cref="IObservable{T}"/> to which a subscription can be placed.
+        /// The observable never completes, since executions can be run many times.
+        /// </summary>
+        /// <returns>A `new` instance of type <see cref="IShell"/> with the subscribers attached to the observable.</returns>
+        IShell WithSubscribe(Action<IObservable<ICommandEvent>> subscriber);
+
         /// <summary>
         /// Ensures the shell context will not throw on a non-zero exit code and returns a `new` context instance.
         /// </summary>
@@ -161,6 +168,13 @@ public interface IShell<out TShell> : IShell where TShell : IShell<TShell>
         /// <returns>A `new` instance of <typeparamref name="TShell"/> with the standard error handler passed to this call.</returns>
         new TShell UseInputRequestHandler(Func<string, string, Task<string>> inputRequestHandler);
 
+        /// <summary>
+        /// Provides an <see cref="IObservable{T}"/> to which a subscription can be placed.
+        /// The observable never completes, since executions can be run many times.
+        /// </summary>
+        /// <returns>A `new` instance of type <typeparamref name="TShell"/> with the subscribers attached to the observable.</returns>
+        new TShell WithSubscribe(Action<IObservable<ICommandEvent>> subscriber);
+
         /// <summary>
         /// Ensures the shell context will not throw on a non-zero exit code and returns a `new` context instance.
         /// </summary>
diff --git a/src/Core/Sheller.cs b/src/Core/Sheller.cs
index 986169d..a7d34ce 100644
--- a/src/Core/Sheller.cs
+++ b/src/Core/Sheller.cs
@@ -5,6 +5,7 @@
 using Sheller.Models;
 
 // TODO:
+//   * stdout/stderr to memorystream?
 
 namespace Sheller
 {
diff --git a/src/Core/Sheller.csproj b/src/Core/Sheller.csproj
index 373e7c6..52a744c 100644
--- a/src/Core/Sheller.csproj
+++ b/src/Core/Sheller.csproj
@@ -5,7 +5,7 @@
     <IsTestProject>false</IsTestProject>
     <TargetFramework>netstandard2.0</TargetFramework>
 
-    <Version>4.0.0</Version>
+    <Version>5.0.0</Version>
     <AssemblyName>Sheller</AssemblyName>
     <PackageId>Sheller</PackageId>
     <PackageProjectUrl>https://github.com/twitchax/sheller</PackageProjectUrl>
diff --git a/src/Test/BasicTests.cs b/src/Test/BasicTests.cs
index 48cb3f7..c55ce7d 100644
--- a/src/Test/BasicTests.cs
+++ b/src/Test/BasicTests.cs
@@ -1,10 +1,13 @@
 using System;
+using System.Collections.Generic;
 using System.Diagnostics;
+using System.Reactive.Linq;
 using System.Text;
 using System.Threading.Tasks;
 using Sheller.Implementations;
 using Sheller.Implementations.Executables;
 using Sheller.Implementations.Shells;
+using Sheller.Models;
 using Xunit;
 
 // NOTE: win tests require WSL...because...lazy.
@@ -356,5 +359,46 @@ await Assert.ThrowsAsync<ExecutionTimeoutException>(() =>
             Assert.True(delta.TotalSeconds > min);
             Assert.True(delta.TotalSeconds < max);
         }
+
+        [Fact]
+        [Trait("os", "nix_win")]
+        public async void CanExecuteAndSubscribe()
+        {
+            var events = new List<string>();
+            var expected = "lol";
+
+            var command1 = Builder
+                .UseShell<Bash>()
+                .UseExecutable("echo")
+                .WithArgument(expected)
+                .WithSubscribe(o =>
+                {
+                    o.Where(ev => ev.Type == CommandEventType.StandardOutput).Select(ev => ev.Data).Do(data =>
+                    {
+                        events.Add(data);
+                    }).Subscribe();
+                });
+
+            var command2 = command1
+                .WithSubscribe(o =>
+                {
+                    o.Where(ev => ev.Type == CommandEventType.StandardOutput).Select(ev => ev.Data).Do(data =>
+                    {
+                        events.Add(data);
+                    }).Subscribe();
+                });
+
+            await command1
+                .ExecuteAsync();
+
+            await command1
+                .ExecuteAsync();
+
+            await command2
+                .ExecuteAsync();
+            
+            Assert.Equal(4, events.Count);
+            Assert.Contains(expected, events);
+        }
     }
 }
diff --git a/src/Test/Sheller.Tests.csproj b/src/Test/Sheller.Tests.csproj
index 71b39ec..b216514 100644
--- a/src/Test/Sheller.Tests.csproj
+++ b/src/Test/Sheller.Tests.csproj
@@ -12,6 +12,7 @@
 
   <ItemGroup>
     <PackageReference Include="Microsoft.NET.Test.Sdk" Version="15.8.0" />
+    <PackageReference Include="System.Reactive" Version="4.1.3" />
     <PackageReference Include="xunit" Version="2.3.1" />
     <PackageReference Include="xunit.runner.visualstudio" Version="2.3.1" />
   </ItemGroup>