Pause the fiber while ActionDriver.Invoke is executing.

tosh-coding · Jun 18, 2024 · 9e32494 · 9e32494
1 parent 2a253b6
commit 9e32494
Show file tree

Hide file tree

Showing 5 changed files with 24 additions and 11 deletions.
diff --git a/src/AsyncFiberWorks/Core/FiberExecutionEventArgs.cs b/src/AsyncFiberWorks/Core/FiberExecutionEventArgs.cs
@@ -87,5 +87,13 @@ public void PauseWhileRunning(Func<Task> runningTask)
                 }
             });
         }
+
+        /// <summary>
+        /// Source thread.
+        /// </summary>
+        public IThreadPool SourceThread
+        {
+            get { return _threadPool; }
+        }
     }
 }
diff --git a/src/AsyncFiberWorks/Procedures/ActionDriver.cs b/src/AsyncFiberWorks/Procedures/ActionDriver.cs
@@ -72,18 +72,22 @@ private IDisposable AddHandler(Action<FiberExecutionEventArgs> action)
         }
 
         /// <summary>
-        /// Invoke all actions.
+        /// Invoke all subscribers.
+        /// The fiber passed in the argument may be paused.
         /// </summary>
-        /// <param name="fiber"></param>
-        public void Invoke(IFiber fiber)
+        /// <param name="eventArgs">Handle for fiber pause.</param>
+        public void InvokeAsync(FiberExecutionEventArgs eventArgs)
         {
+            eventArgs.Pause();
+            var tmpFiber = new PoolFiber(eventArgs.SourceThread);
             lock (_lock)
             {
                 foreach (var action in _actions)
                 {
-                    fiber.Enqueue(action);
+                    tmpFiber.Enqueue(action);
                 }
             }
+            tmpFiber.Enqueue(() => eventArgs.Resume());
         }
 
         ///<summary>

diff --git a/src/AsyncFiberWorks/Procedures/IActionDriverInvoker.cs b/src/AsyncFiberWorks/Procedures/IActionDriverInvoker.cs
@@ -1,4 +1,4 @@
-using AsyncFiberWorks.Fibers;
+using AsyncFiberWorks.Core;
 
 namespace AsyncFiberWorks.Procedures
 {
@@ -9,8 +9,9 @@ public interface IActionDriverInvoker
     {
         /// <summary>
         /// Invoke all subscribers.
+        /// The fiber passed in the argument may be paused.
         /// </summary>
-        /// <param name="fiber"></param>
-        void Invoke(IFiber fiber);
+        /// <param name="eventArgs">Handle for fiber pause.</param>
+        void InvokeAsync(FiberExecutionEventArgs eventArgs);
     }
 }
diff --git a/src/AsyncFiberWorksTests/ActionDriverTests.cs b/src/AsyncFiberWorksTests/ActionDriverTests.cs
@@ -36,7 +36,7 @@ public void Invoking()
 
             var loop = new ThreadPoolAdaptor();
             var fiber = new PoolFiber(loop);
-            driver.Invoke(fiber);
+            fiber.Enqueue((e) => driver.InvokeAsync(e));
             fiber.Enqueue(() => loop.Stop());
             loop.Run();
             Assert.AreEqual(301, counter);
@@ -67,7 +67,7 @@ public async Task AsyncInvoking()
             var disposable2 = driver.SubscribeAndReceiveAsTask(action2);
 
             var fiber = new PoolFiber();
-            driver.Invoke(fiber);
+            fiber.Enqueue((e) => driver.InvokeAsync(e));
             await fiber.EnqueueTaskAsync(() => Task.CompletedTask);
             Assert.AreEqual(301, counter);
         }
@@ -122,7 +122,7 @@ public void ToggleAtUnsubscribe()
 
             var loop = new ThreadPoolAdaptor();
             var fiber = new PoolFiber(loop);
-            driver.Invoke(fiber);
+            fiber.Enqueue((e) => driver.InvokeAsync(e));
             fiber.Enqueue(() => loop.Stop());
             loop.Run();
             Assert.AreEqual(1, counter);

diff --git a/src/AsyncFiberWorksTests/AsyncRegisterTests.cs b/src/AsyncFiberWorksTests/AsyncRegisterTests.cs
@@ -47,7 +47,7 @@ public async Task WaitingOfAsyncRegister()
             var fiber = new PoolFiber();
             for (int i = 0; i < 10; i++)
             {
-                driver.Invoke(fiber);
+                fiber.Enqueue((e) => driver.InvokeAsync(e));
                 await fiber.EnqueueTaskAsync(() => Task.CompletedTask);
             }