Integrate AsyncActionDriver and ActionDriver.

Author: tosh-coding

src/AsyncFiberWorks/Core/AsyncToggleFilter.cs

@@ -9,18 +9,6 @@ internal class ToggleFilter
     {
         private bool _running = true;
 
-        ///<summary>
-        /// Executes a single action.
-        ///</summary>
-        ///<param name="toExecute"></param>
-        public void Execute(Action toExecute)
-        {
-            if (_running)
-            {
-                toExecute();
-            }
-        }
-
         /// <summary>
         /// When disabled, actions will be ignored by filter.
         /// The filter is typically disabled at shutdown

src/AsyncFiberWorks/Core/ToggleFilter.cs

@@ -1,16 +1,20 @@
+using AsyncFiberWorks.Core;
+using AsyncFiberWorks.Fibers;
 using System;
+using System.Collections.Generic;
 
 namespace AsyncFiberWorks.Procedures
 {
     /// <summary>
-    /// Default driver implementation. Invokes all of the subscriber's actions.
+    /// Invokes all of the subscriber's actions.
     /// </summary>
     public class ActionDriver : IActionDriver
     {
-        private readonly ActionList _actions = new ActionList();
+        private object _lock = new object();
+        private LinkedList<Action<FiberExecutionEventArgs>> _actions = new LinkedList<Action<FiberExecutionEventArgs>>();
 
         /// <summary>
-        /// Create a driver.
+        /// Create an action driver.
         /// </summary>
         public ActionDriver()
         {
@@ -23,20 +27,77 @@ public ActionDriver()
         /// <returns>Unsubscriber.</returns>
         public IDisposable Subscribe(Action action)
         {
-            return _actions.AddHandler(action);
+            return AddHandler((e) => action());
         }
 
         /// <summary>
-        /// <see cref="IActionInvoker.Invoke"/>
+        /// Subscribe a channel.
+        /// </summary>
+        /// <param name="action">Subscriber.</param>
+        /// <returns>Unsubscriber.</returns>
+        public IDisposable Subscribe(Action<FiberExecutionEventArgs> action)
+        {
+            return AddHandler(action);
+        }
+
+        /// <summary>
+        /// Add an action.
+        /// </summary>
+        /// <param name="action">An action.</param>
+        /// <returns>Function for removing the action.</returns>
+        private IDisposable AddHandler(Action<FiberExecutionEventArgs> action)
+        {
+            var maskableFilter = new ToggleFilter();
+            Action<FiberExecutionEventArgs> safeAction = (e) =>
+            {
+                var enabled = maskableFilter.IsEnabled;
+                if (enabled)
+                {
+                    action(e);
+                }
+            };
+
+            lock (_lock)
+            {
+                _actions.AddLast(safeAction);
+            }
+
+            var unsubscriber = new Unsubscriber(() =>
+            {
+                maskableFilter.IsEnabled = false;
+                _actions.Remove(safeAction);
+            });
+
+            return unsubscriber;
+        }
+
+        /// <summary>
+        /// Invoke all actions.
         /// </summary>
-        public void Invoke()
+        /// <param name="fiber"></param>
+        public void Invoke(IFiber fiber)
         {
-            _actions.Invoke();
+            lock (_lock)
+            {
+                foreach (var action in _actions)
+                {
+                    fiber.Enqueue(action);
+                }
+            }
         }
 
         ///<summary>
         /// Number of subscribers
         ///</summary>
-        public int NumSubscribers { get { return _actions.Count; } }
+        public int NumSubscribers
+        {
+            get
+            {
+                lock (_lock)
+                {
+                    return _actions.Count;
+                }
+            }
+        }
     }
 }

src/AsyncFiberWorks/Procedures/ActionDriver.cs

@@ -0,0 +1,25 @@
+using System;
+using System.Threading.Tasks;
+
+namespace AsyncFiberWorks.Procedures
+{
+    /// <summary>
+    /// ActionDriver extension.
+    /// </summary>
+    public static class ActionDriverExtensions
+    {
+        /// <summary>
+        /// Subscribe a channel.
+        /// </summary>
+        /// <param name="driver"></param>
+        /// <param name="task">Subscriber.</param>
+        /// <returns>Unsubscriber.</returns>
+        public static IDisposable SubscribeAndReceiveAsTask(this IActionDriverSubscriber driver, Func<Task> task)
+        {
+            return driver.Subscribe((e) =>
+            {
+                e.PauseWhileRunning(task);
+            });
+        }
+    }
+}

src/AsyncFiberWorks/Procedures/ActionDriverExtensions.cs

@@ -22,9 +22,9 @@ public class AsyncRegister : IDisposable
         /// Subscribe a driver.
         /// </summary>
         /// <param name="subscribable"></param>
-        public AsyncRegister(IAsyncActionSubscriber subscribable)
+        public AsyncRegister(IActionDriverSubscriber subscribable)
         {
-            _subscription = subscribable.Subscribe(async () =>
+            _subscription = subscribable.SubscribeAndReceiveAsTask(async () =>
             {
                 await SetFlagAndWaitClearing().ConfigureAwait(false);
             });