Merge pull request #12 from jhickson/pausewhileawait

Add ability to pause actor whilst awaiting

Author: Jos Hickson

Winton.Extensions.Threading.Actor.Tests.Unit/ActorTests.cs

@@ -875,6 +875,76 @@ public void ShouldStopActorAndNotProcessAnyAlreadyEnqueuedWorkIfStartWorkCancell
             attempts.Should().Be(0);
         }
 
+        [Fact]
+        public async Task ShouldBeAbleToPauseActorUntilResumeFromAwait()
+        {
+            var actor = CreateActor();
+            var numbers = new List<int>();
+
+            await actor.Start();
+
+            var tasks =
+                new[]
+                {
+                    actor.Enqueue(
+                        async () =>
+                        {
+                            numbers.Add(1);
+
+                            await Task.Delay(TimeSpan.FromSeconds(1)).WhileActorPaused();
+
+                            numbers.Add(2);
+                        }),
+                    actor.Enqueue(() => numbers.Add(3)),
+                    actor.Enqueue(() => numbers.Add(4)),
+                    actor.Enqueue(() => numbers.Add(5))
+                };
+
+            await Task.WhenAll(tasks);
+
+            numbers.Should().Equal(1, 2, 3, 4, 5);
+        }
+
+        [Fact]
+        public async Task ShouldBeAbleToPauseActorUntilResumeFromAwaitReturningData()
+        {
+            var actor = CreateActor();
+            var numbers = new List<int>();
+            var promise = new TaskCompletionSource<int>();
+
+            await actor.Start();
+
+            var tasks =
+                new[]
+                {
+                    actor.Enqueue(
+                        async () =>
+                        {
+                            numbers.Add(1);
+
+                            var next = await promise.Task.WhileActorPaused();
+
+                            numbers.Add(next);
+                        }),
+                    actor.Enqueue(() => numbers.Add(3)),
+                    actor.Enqueue(() => numbers.Add(4)),
+                    actor.Enqueue(() => numbers.Add(5))
+                };
+
+            var promiseSetter =
+                Task.Run(
+                    async () =>
+                    {
+                        await Task.Delay(TimeSpan.FromSeconds(1));
+                        promise.SetResult(2);
+                    });
+
+            await Task.WhenAll(tasks);
+            await promiseSetter;
+
+            numbers.Should().Equal(1, 2, 3, 4, 5);
+        }
+
         [Flags]
         private enum ActorCreateOptions
         {

Winton.Extensions.Threading.Actor.Tests.Unit/Winton.Extensions.Threading.Actor.Tests.Unit.csproj

@@ -11,7 +11,7 @@
   <ItemGroup>
     <PackageReference Include="FluentAssertions" Version="4.19.2" />
     <PackageReference Include="Microsoft.NET.Test.Sdk" Version="15.0.0" />
-    <PackageReference Include="Moq" Version="4.7.1" />
+    <PackageReference Include="Moq" Version="4.7.12" />
     <PackageReference Include="xunit" Version="2.2.0" />
     <PackageReference Include="xunit.runner.visualstudio" Version="2.2.0" />
   </ItemGroup>

Winton.Extensions.Threading.Actor/Actor.cs

@@ -46,8 +46,8 @@ private Actor(ActorImpl impl)
         /// </summary>
         public static ActorId CurrentId
         {
-            get { return _currentId; }
-            internal set { _currentId = value; }
+            get => _currentId;
+            internal set => _currentId = value;
         }
 
         /// <inheritdoc cref="IActor.Id"/>
@@ -56,13 +56,13 @@ public static ActorId CurrentId
         /// <inheritdoc cref="IActor.StartWork"/>
         public ActorStartWork StartWork
         {
-            set { _impl.StartWork = value; }
+            set => _impl.StartWork = value;
         }
 
         /// <inheritdoc cref="IActor.StopWork"/>
         public ActorStopWork StopWork
         {
-            set { _impl.StopWork = value; }
+            set => _impl.StopWork = value;
         }
 
         /// <inheritdoc cref="IActor.Start"/>

Winton.Extensions.Threading.Actor/Internal/ActorImpl.cs

@@ -16,12 +16,12 @@ public ActorImpl(IActorTaskFactory actorTaskFactory)
 
         public ActorStartWork StartWork
         {
-            set { Context.StartWork = value; }
+            set => Context.StartWork = value;
         }
 
         public ActorStopWork StopWork
         {
-            set { Context.StopWork = value; }
+            set => Context.StopWork = value;
         }
 
         public Task Start()

Winton.Extensions.Threading.Actor/Internal/ActorSynchronizationContext.cs

@@ -7,16 +7,23 @@ internal sealed class ActorSynchronizationContext : SynchronizationContext
     {
         private readonly ActorTaskScheduler _scheduler;
         private readonly IActorTaskFactory _actorTaskFactory;
+        private readonly ActorTaskKind _actorTaskKind;
 
-        public ActorSynchronizationContext(ActorTaskScheduler scheduler, IActorTaskFactory actorTaskFactory)
+        public ActorSynchronizationContext(ActorTaskScheduler scheduler, IActorTaskFactory actorTaskFactory, ActorTaskKind actorTaskKind)
         {
             _scheduler = scheduler;
             _actorTaskFactory = actorTaskFactory;
+            _actorTaskKind = actorTaskKind;
+        }
+
+        public ActorSynchronizationContext ChangeActorTaskKind(ActorTaskKind actorTaskKind)
+        {
+            return new ActorSynchronizationContext(_scheduler, _actorTaskFactory, actorTaskKind);
         }
 
         public override void Post(SendOrPostCallback callback, object state)
         {
-            _actorTaskFactory.Create(() => callback(state), CancellationToken.None, TaskCreationOptions.HideScheduler).Start(_scheduler);
+            _actorTaskFactory.Create(() => callback(state), CancellationToken.None, TaskCreationOptions.HideScheduler, _actorTaskKind).Start(_scheduler);
         }
     }
 }

Winton.Extensions.Threading.Actor/Internal/ActorTask.cs

@@ -1,3 +1,4 @@
+using System;
 using System.Threading.Tasks;
 
 namespace Winton.Extensions.Threading.Actor.Internal
@@ -9,6 +10,11 @@ public static bool IsActorTask(this Task self)
             return self.AsyncState is ActorTaskContext;
         }
 
+        public static ActorTaskContext GetActorTaskContext(this Task self)
+        {
+            return self.AsyncState as ActorTaskContext ?? throw new InvalidOperationException("Task is not an actor task.");
+        }
+
         public static void Cancel(this Task self)
         {
             var context = self.AsyncState as ActorTaskContext;

Winton.Extensions.Threading.Actor/Internal/ActorTaskContext.cs

@@ -4,11 +4,14 @@ namespace Winton.Extensions.Threading.Actor.Internal
 {
     internal sealed class ActorTaskContext
     {
-        public ActorTaskContext(CancellationToken cancellationToken)
+        public ActorTaskContext(CancellationToken cancellationToken, ActorTaskKind kind)
         {
+            Kind = kind;
             Canceller = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
         }
 
         public CancellationTokenSource Canceller { get; }
+
+        public ActorTaskKind Kind { get; }
     }
 }

Winton.Extensions.Threading.Actor/Internal/ActorTaskFactoryExtensions.cs

@@ -8,9 +8,9 @@ internal static class ActorTaskFactoryExtensions
     {
         private const TaskCreationOptions BaseCreationOptions = TaskCreationOptions.HideScheduler;
 
-        public static Task Create(this IActorTaskFactory self, Action work, CancellationToken cancellationToken, TaskCreationOptions taskCreationOptions)
+        public static Task Create(this IActorTaskFactory self, Action work, CancellationToken cancellationToken, TaskCreationOptions taskCreationOptions, ActorTaskKind kind = ActorTaskKind.Standard)
         {
-            var actorTaskState = new ActorTaskContext(cancellationToken);
+            var actorTaskState = new ActorTaskContext(cancellationToken, kind);
             return self.Create(state =>
                                {
                                    try
@@ -25,9 +25,9 @@ public static Task Create(this IActorTaskFactory self, Action work, Cancellation
                                actorTaskState.Canceller.Token, taskCreationOptions | BaseCreationOptions, actorTaskState);
         }
 
-        public static Task<T> Create<T>(this IActorTaskFactory self, Func<T> work, CancellationToken cancellationToken, TaskCreationOptions taskCreationOptions)
+        public static Task<T> Create<T>(this IActorTaskFactory self, Func<T> work, CancellationToken cancellationToken, TaskCreationOptions taskCreationOptions, ActorTaskKind kind = ActorTaskKind.Standard)
         {
-            var actorTaskState = new ActorTaskContext(cancellationToken);
+            var actorTaskState = new ActorTaskContext(cancellationToken, kind);
             return self.Create(state =>
                                {
                                    try

Winton.Extensions.Threading.Actor/Internal/ActorTaskKind.cs

@@ -0,0 +1,8 @@
+namespace Winton.Extensions.Threading.Actor.Internal
+{
+    internal enum ActorTaskKind
+    {
+        Standard,
+        Resumer
+    }
+}