Allow cancel Invocable with payload on ConsumeQueueOnShutdown

Src/Coravel/Queuing/Queue.cs

@@ -47,10 +47,16 @@ public Guid QueueInvocable<T>() where T : IInvocable
 
         public Guid QueueInvocableWithPayload<T, TParams>(TParams payload) where T : IInvocable, IInvocableWithPayload<TParams>
         {
+            var tokenSource = new CancellationTokenSource();
             var job = this.EnqueueInvocable<T>(invocable => {
-                IInvocableWithPayload<TParams> invocableWithParams = (IInvocableWithPayload<TParams>) invocable;
+                var invocableWithParams = (IInvocableWithPayload<TParams>) invocable;
                 invocableWithParams.Payload = payload;
+                if (invocableWithParams is ICancellableTask)
+                {
+                    ((ICancellableTask) invocable).Token = tokenSource.Token;
+                }
             });
+            this._tokens.TryAdd(job.Guid, tokenSource);
             return job.Guid;
         }
 

Src/UnitTests/CoravelUnitTests/Queuing/CancellableInvocableForQueueTests.cs

@@ -12,6 +12,25 @@ namespace UnitTests.Queuing
 {
     public class CancellableInvocableForQueueTests
     {
+        [Fact]
+        public async Task CanCancelInvocableWithPayload()
+        {
+            var services = new ServiceCollection();
+            services.AddTransient<TestCancellableInvocable>();
+            services.AddTransient<TestCancellableInvocableWithPayload>();
+            var provider = services.BuildServiceProvider();
+
+            var queue = new Queue(provider.GetRequiredService<IServiceScopeFactory>(), new DispatcherStub());
+            var payload = new TestPayload()
+            {
+                Code = "test"
+            };
+            queue.QueueInvocableWithPayload<TestCancellableInvocableWithPayload, TestPayload>(payload);
+            TestCancellableInvocable.TokensCancelled = 0;
+            await queue.ConsumeQueueOnShutdown();
+            Assert.Equal(1, TestCancellableInvocableWithPayload.TokensCancelled);
+        }
+
         [Fact]
         public async Task CanCancelInvocable()
         {
@@ -82,9 +101,37 @@ public Task Invoke()
                 {
                     Interlocked.Increment(ref TokensCancelled);
                 }
+                return Task.CompletedTask;
+            }
+        }
+
+        private class TestCancellableInvocableWithPayload : IInvocable, IInvocableWithPayload<TestPayload>, ICancellableTask
+        {
+            /// <summary>
+            /// Static fields keeps track of all cancelled tokens count.
+            /// </summary>
+            public static int TokensCancelled = 0;
+            public TestPayload Payload { get; set; }
+            public TestCancellableInvocableWithPayload() {}
+
+            public CancellationToken Token { get; set; }
+
+            public Task Invoke()
+            {
+                if(this.Token.IsCancellationRequested)
+                {
+                    Interlocked.Increment(ref TokensCancelled);
+                }
+                Thread.Sleep(10000);
 
                 return Task.CompletedTask;
             }
+
+        }
+
+        private class TestPayload
+        {
+            public string Code { get; set; }
         }
     }
 }