Add IAsyncFiber.EnqueueAsync.

Author: tosh-coding

src/AsyncFiberWorks/Core/AsyncFiberExtentions.cs

@@ -0,0 +1,34 @@
+using System;
+using System.Threading.Tasks;
+
+namespace AsyncFiberWorks.Core
+{
+    /// <summary>
+    /// Extensions to AsyncFiber.
+    /// </summary>
+    public static class AsyncFiberExtentions
+    {
+        /// <summary>
+        /// Enqueue a task.
+        /// </summary>
+        /// <param name="fiber">An fiber.</param>
+        /// <param name="func">A function that returns a task.</param>
+        /// <returns>A task that waits until a given task is finished.</returns>
+        public static async Task EnqueueAsync(this IAsyncFiber fiber, Func<Task> func)
+        {
+            var tcs = new TaskCompletionSource<byte>();
+            fiber.Enqueue(async () =>
+            {
+                try
+                {
+                    await func().ConfigureAwait(false);
+                }
+                finally
+                {
+                    tcs.SetResult(0);
+                }
+            });
+            await tcs.Task;
+        }
+    }
+}

src/AsyncFiberWorks/Core/AsyncNonReentrantExecutor.cs

@@ -0,0 +1,56 @@
+using System;
+using System.Threading.Tasks;
+
+namespace AsyncFiberWorks.Core
+{
+    /// <summary>
+    /// The same instance of this class will not be executed concurrently.
+    /// The one executed later is skipped.
+    /// </summary>
+    public class AsyncNonReentrantFiberFilter : IAsyncFiber
+    {
+        private readonly object _lockObj = new object();
+        private readonly IAsyncFiber _fiber;
+        private bool _executing = false;
+
+        /// <summary>
+        /// Create a filter.
+        /// </summary>
+        /// <param name="fiber"></param>
+        public AsyncNonReentrantFiberFilter(IAsyncFiber fiber)
+        {
+            _fiber = fiber;
+        }
+
+        /// <summary>
+        /// Enqueue a task.
+        /// </summary>
+        /// <param name="func">A function that returns a task.</param>
+        public void Enqueue(Func<Task> func)
+        {
+            lock (_lockObj)
+            {
+                if (_executing)
+                {
+                    return;
+                }
+                _executing = true;
+            }
+
+            _fiber.Enqueue(async () =>
+            {
+                try
+                {
+                    await func().ConfigureAwait(false);
+                }
+                finally
+                {
+                    lock (_lockObj)
+                    {
+                        _executing = false;
+                    }
+                }
+            });
+        }
+    }
+}

src/AsyncFiberWorks/Core/AsyncNonReentrantFiberFilter.cs

@@ -0,0 +1,55 @@
+using System;
+
+namespace AsyncFiberWorks.Core
+{
+    /// <summary>
+    /// The same instance of this class will not be executed concurrently.
+    /// The one executed later is skipped.
+    /// </summary>
+    public class NonReentrantFiberFilter : IExecutionContext
+    {
+        private readonly object _lockObj = new object();
+        private readonly IExecutionContext _fiber;
+        private bool _executing = false;
+
+        /// <summary>
+        /// Create a filter.
+        /// </summary>
+        /// <param name="fiber"></param>
+        public NonReentrantFiberFilter(IExecutionContext fiber)
+        {
+            _fiber = fiber;
+        }
+
+        /// <summary>
+        /// Enqueue an action.
+        /// </summary>
+        /// <param name="action">Action to be executed.</param>
+        public void Enqueue(Action action)
+        {
+            lock (_lockObj)
+            {
+                if (_executing)
+                {
+                    return;
+                }
+                _executing = true;
+            }
+
+            _fiber.Enqueue(() =>
+            {
+                try
+                {
+                    action();
+                }
+                finally
+                {
+                    lock (_lockObj)
+                    {
+                        _executing = false;
+                    }
+                }
+            });
+        }
+    }
+}

src/AsyncFiberWorks/Core/NonReentrantFiberFilter.cs

@@ -1,3 +1,4 @@
+using AsyncFiberWorks.Channels;
 using AsyncFiberWorks.Core;
 using System;
 
@@ -29,14 +30,23 @@ public IDisposable Subscribe(Action action)
         {
             if (_executor != null)
             {
-                return _actions.AddHandler(() =>
+                var maskableFilter = new MaskableExecutor();
+                var disposable = _actions.AddHandler(() => maskableFilter.Execute(() => _executor.Execute(action)));
+                return new Unsubscriber(() =>
                 {
-                    _executor.Execute(action);
+                    maskableFilter.IsEnabled = false;
+                    disposable.Dispose();
                 });
             }
             else
             {
-                return _actions.AddHandler(action);
+                var maskableFilter = new MaskableExecutor();
+                var disposable = _actions.AddHandler(() => maskableFilter.Execute(action));
+                return new Unsubscriber(() =>
+                {
+                    maskableFilter.IsEnabled = false;
+                    disposable.Dispose();
+                });
             }
         }
 

src/AsyncFiberWorks/Procedures/ActionDriver.cs

@@ -1,5 +1,6 @@
 using System;
 using System.Threading.Tasks;
+using AsyncFiberWorks.Channels;
 using AsyncFiberWorks.Core;
 
 namespace AsyncFiberWorks.Procedures
@@ -43,7 +44,13 @@ public AsyncActionDriver()
         /// <returns>Unsubscriber.</returns>
         public IDisposable Subscribe(Func<Task> action)
         {
-            return _actions.AddHandler(action);
+            var maskableFilter = new AsyncMaskableExecutor();
+            var disposable = _actions.AddHandler(() => maskableFilter.Execute(action));
+            return new Unsubscriber(() =>
+            {
+                maskableFilter.IsEnabled = false;
+                disposable.Dispose();
+            });
         }
 
         /// <summary>

src/AsyncFiberWorks/Procedures/AsyncActionDriver.cs

@@ -44,12 +44,11 @@ async Task AsyncFiberTest(IAsyncFiber fiber)
                 counter += 100;
                 await Task.Yield();
             });
-#pragma warning disable 1998
-            fiber.Enqueue(async () =>
+            fiber.Enqueue(() =>
             {
                 tcs.SetResult(0);
+                return Task.CompletedTask;
             });
-#pragma warning restore 1998
 
             await tcs.Task;
             Assert.AreEqual(120, counter);
@@ -118,5 +117,29 @@ public async Task RepeatingTimer()
             Console.WriteLine($"as{sw.Elapsed}");
             Assert.AreEqual(3, counter);
         }
+
+        [Test]
+        public async Task EnqueueAsyncTest()
+        {
+            var fiber = new ChainAsyncFiber();
+            int counter = 0;
+            var sw = Stopwatch.StartNew();
+            var t1 = fiber.EnqueueAsync(async () =>
+            {
+                counter = 1;
+                await Task.Delay(300).ConfigureAwait(false);
+            });
+            await Task.Delay(10).ConfigureAwait(false);
+            fiber.Enqueue(() =>
+            {
+                counter = 2;
+                return Task.CompletedTask;
+            });
+            Assert.AreEqual(1, counter);
+            await t1;
+            Assert.GreaterOrEqual(sw.Elapsed, TimeSpan.FromMilliseconds(300));
+            await Task.Delay(10).ConfigureAwait(false);
+            Assert.AreEqual(2, counter);
+        }
     }
 }