Remove ISequential*Registry.

Author: tosh-coding

src/AsyncFiberWorks/Procedures/FiberAndHandlerPairList.cs

@@ -1,6 +1,7 @@
 using AsyncFiberWorks.Core;
 using System;
 using System.Collections.Generic;
+using System.Threading;
 using System.Threading.Tasks;
 
 namespace AsyncFiberWorks.Procedures
@@ -13,7 +14,7 @@ namespace AsyncFiberWorks.Procedures
     /// If it has not yet been processed, it proceeds to the next step. If already processed, exit at that point.
     /// </summary>
     /// <typeparam name="TMessage">Message type.</typeparam>
-    public class FiberAndHandlerPairList<TMessage> : ISequentialHandlerListRegistry<TMessage>, ISequentialPublisher<TMessage>
+    public class FiberAndHandlerPairList<TMessage> : ISequentialPublisher<TMessage>
     {
         private readonly object _lock = new object();
         private readonly LinkedList<RegisteredHandler> _actions = new LinkedList<RegisteredHandler>();
@@ -125,6 +126,20 @@ public IDisposable Add(Action<IFiberExecutionEventArgs, ProcessedFlagEventArgs<T
             return unsubscriber;
         }
 
+        /// <summary>
+        /// Add a handler to the tail.
+        /// </summary>
+        /// <param name="handler">Message handler. The return value is the processed flag. If this flag is true, subsequent handlers are not called.</param>
+        /// <param name="context">The context in which the handler will execute. if null, the default is used.</param>
+        /// <returns>Handle for canceling registration.</returns>
+        public IDisposable Add(Func<TMessage, Task<bool>> handler, IFiber context = null)
+        {
+            return this.Add((IFiberExecutionEventArgs e, ProcessedFlagEventArgs<TMessage> message) => e.PauseWhileRunning(async () =>
+            {
+                message.Processed = await handler(message.Arg).ConfigureAwait(false);
+            }), context);
+        }
+
         /// <summary>
         /// Distribute one message.
         /// </summary>

src/AsyncFiberWorks/Procedures/FiberAndTaskPairList.cs

@@ -1,6 +1,7 @@
 using AsyncFiberWorks.Core;
 using System;
 using System.Collections.Generic;
+using System.Threading;
 using System.Threading.Tasks;
 
 namespace AsyncFiberWorks.Procedures
@@ -11,7 +12,7 @@ namespace AsyncFiberWorks.Procedures
     /// Can specify the fiber to be executed.
     /// Wait for the calls to complete one by one before proceeding.
     /// </summary>
-    public class FiberAndTaskPairList : ISequentialTaskInvoker, ISequentialTaskListRegistry
+    public class FiberAndTaskPairList : ISequentialTaskInvoker
     {
         private readonly object _lock = new object();
         private readonly LinkedList<RegisteredAction> _actions = new LinkedList<RegisteredAction>();
@@ -121,6 +122,17 @@ public IDisposable Add(Action<IFiberExecutionEventArgs> task, IFiber context = n
             return unsubscriber;
         }
 
+        /// <summary>
+        /// Add a task to the tail.
+        /// </summary>
+        /// <param name="task">Task to be performed.</param>
+        /// <param name="context">The context in which the task will execute. if null, the default is used.</param>
+        /// <returns>Handle for canceling registration.</returns>
+        public IDisposable Add(Func<Task> task, IFiber context = null)
+        {
+            return this.Add((IFiberExecutionEventArgs e) => e.PauseWhileRunning(task), context);
+        }
+
         /// <summary>
         /// Invoke all tasks sequentially.
         /// </summary>

src/AsyncFiberWorks/Procedures/ISequentialHandlerListRegistry.cs

@@ -12,7 +12,6 @@ namespace AsyncFiberWorks.Procedures
     public class SequentialHandlerWaiter<T> : IDisposable
     {
         private readonly object _lockObj = new object();
-        private IDisposable _unsubscriber;
         private bool _executionRequested;
         private bool _isDisposed;
         private readonly ManualResetEventSlim _notifierExecutionRequested = new ManualResetEventSlim();
@@ -24,22 +23,26 @@ public class SequentialHandlerWaiter<T> : IDisposable
         private readonly ProcessedFlagEventArgs<T> _currentValue = new ProcessedFlagEventArgs<T>();
 
         /// <summary>
-        /// Register a handler to the sequential handler list.
+        /// Constructor.
         /// </summary>
-        /// <param name="handlerList"></param>
         /// <param name="cancellationToken"></param>
-        public SequentialHandlerWaiter(ISequentialHandlerListRegistry<T> handlerList, CancellationToken cancellationToken = default)
+        public SequentialHandlerWaiter(CancellationToken cancellationToken = default)
         {
             _thread = UserThreadPool.StartNew(1);
             _cancellationTokenExternal = cancellationToken;
-            _unsubscriber = handlerList.Add(async (arg) =>
-            {
-                _currentValue.Processed = false;
-                _currentValue.Arg = arg;
-                await ExecuteAsync().ConfigureAwait(false);
-                return _currentValue.Processed;
-            });
+        }
 
+        /// <summary>
+        /// Execute a handler.
+        /// </summary>
+        /// <param name="arg">An argument.</param>
+        /// <returns>Indicates whether it has been processed.</returns>
+        public async Task<bool> Handler(T arg)
+        {
+            _currentValue.Processed = false;
+            _currentValue.Arg = arg;
+            await ExecuteAsync().ConfigureAwait(false);
+            return _currentValue.Processed;
         }
 
         /// <summary>
@@ -160,7 +163,6 @@ public void Dispose()
 
                 _executionRequested = false;
                 _inExecuting = false;
-                _unsubscriber.Dispose();
                 _onDispose.Cancel();
                 _onDispose.Dispose();
                 _notifierExecutionRequested.Dispose();

src/AsyncFiberWorks/Procedures/ISequentialTaskListRegistry.cs

@@ -11,7 +11,6 @@ namespace AsyncFiberWorks.Procedures
     public class SequentialTaskWaiter : IDisposable
     {
         private readonly object _lockObj = new object();
-        private IDisposable _unsubscriber;
         private bool _executionRequested;
         private bool _isDisposed;
         private readonly ManualResetEventSlim _notifierExecutionRequested = new ManualResetEventSlim();
@@ -22,23 +21,21 @@ public class SequentialTaskWaiter : IDisposable
         private readonly CancellationTokenSource _onDispose = new CancellationTokenSource();
 
         /// <summary>
-        /// Register a task to the sequential task list.
+        /// Constructor.
         /// </summary>
-        /// <param name="taskList"></param>
         /// <param name="cancellationToken"></param>
-        public SequentialTaskWaiter(ISequentialTaskListRegistry taskList, CancellationToken cancellationToken = default)
+        public SequentialTaskWaiter(CancellationToken cancellationToken = default)
         {
             _thread = UserThreadPool.StartNew(1);
             _cancellationTokenExternal = cancellationToken;
-            _unsubscriber = taskList.Add(ExecuteAsync);
         }
 
         /// <summary>
         /// Execute a task.
         /// </summary>
         /// <returns></returns>
         /// <exception cref="InvalidOperationException"></exception>
-        async Task ExecuteAsync()
+        public async Task ExecuteTask()
         {
             lock (_lockObj)
             {
@@ -150,7 +147,6 @@ public void Dispose()
 
                 _executionRequested = false;
                 _inExecuting = false;
-                _unsubscriber.Dispose();
                 _onDispose.Cancel();
                 _onDispose.Dispose();
                 _notifierExecutionRequested.Dispose();

src/AsyncFiberWorks/Procedures/SequentialHandlerWaiter.cs

@@ -19,7 +19,8 @@ public async Task TestWait()
 
             Func<int, Task> func = async (maxCount) =>
             {
-                using (var activator = new SequentialTaskWaiter(taskList))
+                using (var activator = new SequentialTaskWaiter())
+                using (var unsubscriber = taskList.Add(activator.ExecuteTask))
                 {
                     int counter = 0;
                     while (counter < maxCount)
@@ -59,7 +60,8 @@ public async Task TestCancellation()
 
             var func = new Func<Task>(async () =>
             {
-                using (var activator = new SequentialTaskWaiter(taskList, cancellationToken))
+                using (var activator = new SequentialTaskWaiter(cancellationToken))
+                using (var unsubscriber = taskList.Add(activator.ExecuteTask))
                 {
                     try
                     {
@@ -101,7 +103,8 @@ public async Task TestWaitingOfT()
 
             Func<int, Task> func = async (maxCount) =>
             {
-                using (var reg = new SequentialHandlerWaiter<int>(handlerList))
+                using (var reg = new SequentialHandlerWaiter<int>())
+                using (var unsubscriber = handlerList.Add(reg.Handler))
                 {
                     int counter = 0;
                     while (counter < maxCount)
@@ -147,7 +150,8 @@ public async Task TestCancellationOfT()
 
             var func = new Func<Task>(async () =>
             {
-                using (var reg = new SequentialHandlerWaiter<int>(handlerList, cancellationToken))
+                using (var reg = new SequentialHandlerWaiter<int>(cancellationToken))
+                using (var unsubscriber = handlerList.Add(reg.Handler))
                 {
                     try
                     {
@@ -185,14 +189,15 @@ public async Task TestCancellationOfT()
         [Test]
         public async Task TestWaitingOfProcessedFlagEventArgs()
         {
-            var driver = new FiberAndHandlerPairList<int>();
+            var handlerList = new FiberAndHandlerPairList<int>();
             int resultCounter = 0;
             var lockObj = new object();
 
             var cts = new CancellationTokenSource();
             Func<int, Task> func = async (threshold) =>
             {
-                using (var reg = new SequentialHandlerWaiter<int>(driver, cts.Token))
+                using (var reg = new SequentialHandlerWaiter<int>(cts.Token))
+                using (var unsubscriber = handlerList.Add(reg.Handler))
                 {
                     while (true)
                     {
@@ -222,7 +227,7 @@ public async Task TestWaitingOfProcessedFlagEventArgs()
             for (int i = 0; i < 10; i++)
             {
                 int eventArg = i + 1;
-                await driver.PublishSequentialAsync(eventArg, defaultContext);
+                await handlerList.PublishSequentialAsync(eventArg, defaultContext);
             }
 
             cts.Cancel();