BEagle1984 · igorgiovannini · Jul 4, 2023
diff --git a/src/Silverback.Core.EFCore30/EntityFrameworkCore/DbContextEventsPublisher.cs b/src/Silverback.Core.EFCore30/EntityFrameworkCore/DbContextEventsPublisher.cs
@@ -5,6 +5,7 @@
 using System.Collections.Generic;
 using System.Diagnostics.CodeAnalysis;
 using System.Linq;
+using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.EntityFrameworkCore;
 using Silverback.Messaging.Messages;
@@ -96,45 +97,48 @@ public int ExecuteSaveTransaction(Func<int> saveChanges)
         /// <param name="saveChangesAsync">
         ///     The delegate to the original <c>SaveChangesAsync</c> method.
         /// </param>
+        /// <param name="cancellationToken">
+        ///     A <see cref="CancellationToken" /> used to cancel the operation.
+        /// </param>
         /// <returns>
         ///     A <see cref="Task{TResult}" /> representing the asynchronous operation. The task result contains the
         ///     number of entities saved to the database.
         /// </returns>
-        public Task<int> ExecuteSaveTransactionAsync(Func<Task<int>> saveChangesAsync)
+        public Task<int> ExecuteSaveTransactionAsync(Func<Task<int>> saveChangesAsync, CancellationToken cancellationToken = default)
         {
             Check.NotNull(saveChangesAsync, nameof(saveChangesAsync));
 
-            return ExecuteSaveTransactionAsync(saveChangesAsync, true);
+            return ExecuteSaveTransactionAsync(saveChangesAsync, true, cancellationToken);
         }
 
-        private async Task<int> ExecuteSaveTransactionAsync(Func<Task<int>> saveChanges, bool executeAsync)
+        private async Task<int> ExecuteSaveTransactionAsync(Func<Task<int>> saveChanges, bool executeAsync, CancellationToken cancellationToken = default)
         {
-            await PublishEventAsync<TransactionStartedEvent>(executeAsync).ConfigureAwait(false);
+            await PublishEventAsync<TransactionStartedEvent>(executeAsync, cancellationToken).ConfigureAwait(false);
 
             var saved = false;
             try
             {
-                await PublishDomainEventsAsync(executeAsync).ConfigureAwait(false);
+                await PublishDomainEventsAsync(executeAsync, cancellationToken).ConfigureAwait(false);
 
                 int result = await saveChanges().ConfigureAwait(false);
 
                 saved = true;
 
-                await PublishEventAsync<TransactionCompletedEvent>(executeAsync).ConfigureAwait(false);
+                await PublishEventAsync<TransactionCompletedEvent>(executeAsync, cancellationToken).ConfigureAwait(false);
 
                 return result;
             }
             catch
             {
                 if (!saved)
-                    await PublishEventAsync<TransactionAbortedEvent>(executeAsync).ConfigureAwait(false);
+                    await PublishEventAsync<TransactionAbortedEvent>(executeAsync, cancellationToken).ConfigureAwait(false);
 
                 throw;
             }
         }
 
         [SuppressMessage("", "VSTHRD103", Justification = Justifications.ExecutesSyncOrAsync)]
-        private async Task PublishDomainEventsAsync(bool executeAsync)
+        private async Task PublishDomainEventsAsync(bool executeAsync, CancellationToken cancellationToken)
         {
             var events = GetDomainEvents();
 
@@ -143,7 +147,7 @@ private async Task PublishDomainEventsAsync(bool executeAsync)
             {
                 if (executeAsync)
                 {
-                    await events.ForEachAsync(message => _publisher.PublishAsync(message))
+                    await events.ForEachAsync(message => _publisher.PublishAsync(message, cancellationToken))
                         .ConfigureAwait(false);
                 }
                 else
@@ -168,11 +172,11 @@ private IReadOnlyCollection<object> GetDomainEvents() =>
                 }).ToList();
 
         [SuppressMessage("", "VSTHRD103", Justification = Justifications.ExecutesSyncOrAsync)]
-        private async Task PublishEventAsync<TEvent>(bool executeAsync)
+        private async Task PublishEventAsync<TEvent>(bool executeAsync, CancellationToken cancellationToken)
             where TEvent : new()
         {
             if (executeAsync)
-                await _publisher.PublishAsync(new TEvent()).ConfigureAwait(false);
+                await _publisher.PublishAsync(new TEvent(), cancellationToken).ConfigureAwait(false);
             else
                 _publisher.Publish(new TEvent());
         }

diff --git a/src/Silverback.Core.Model/Messaging/Publishing/CommandPublisher.cs b/src/Silverback.Core.Model/Messaging/Publishing/CommandPublisher.cs
@@ -2,6 +2,7 @@
 // This code is licensed under MIT license (see LICENSE file for details)
 
 using System.Linq;
+using System.Threading;
 using System.Threading.Tasks;
 using Silverback.Messaging.Messages;
 
@@ -38,24 +39,31 @@ public TResult Execute<TResult>(ICommand<TResult> commandMessage) =>
         public TResult Execute<TResult>(ICommand<TResult> commandMessage, bool throwIfUnhandled) =>
             _publisher.Publish<TResult>(commandMessage, throwIfUnhandled).SingleOrDefault();
 
-        /// <inheritdoc cref="ICommandPublisher.ExecuteAsync(ICommand)" />
-        public Task ExecuteAsync(ICommand commandMessage) => _publisher.PublishAsync(commandMessage);
+        /// <inheritdoc cref="ICommandPublisher.ExecuteAsync(ICommand, CancellationToken)" />
+        public Task ExecuteAsync(ICommand commandMessage, CancellationToken cancellationToken = default) =>
+            _publisher.PublishAsync(commandMessage, cancellationToken);
 
-        /// <inheritdoc cref="ICommandPublisher.ExecuteAsync(ICommand, bool)" />
-        public Task ExecuteAsync(ICommand commandMessage, bool throwIfUnhandled) =>
-            _publisher.PublishAsync(commandMessage, throwIfUnhandled);
+        /// <inheritdoc cref="ICommandPublisher.ExecuteAsync(ICommand, bool, CancellationToken)" />
+        public Task ExecuteAsync(
+            ICommand commandMessage,
+            bool throwIfUnhandled,
+            CancellationToken cancellationToken = default) =>
+            _publisher.PublishAsync(commandMessage, throwIfUnhandled, cancellationToken);
 
-        /// <inheritdoc cref="ICommandPublisher.ExecuteAsync{TResult}(ICommand{TResult})" />
-        public async Task<TResult> ExecuteAsync<TResult>(ICommand<TResult> commandMessage) =>
-            (await _publisher.PublishAsync<TResult>(commandMessage)
+        /// <inheritdoc cref="ICommandPublisher.ExecuteAsync{TResult}(ICommand{TResult}, CancellationToken)" />
+        public async Task<TResult> ExecuteAsync<TResult>(
+            ICommand<TResult> commandMessage,
+            CancellationToken cancellationToken = default) =>
+            (await _publisher.PublishAsync<TResult>(commandMessage, cancellationToken)
                 .ConfigureAwait(false))
             .SingleOrDefault();
 
-        /// <inheritdoc cref="ICommandPublisher.ExecuteAsync{TResult}(ICommand{TResult}, bool)" />
+        /// <inheritdoc cref="ICommandPublisher.ExecuteAsync{TResult}(ICommand{TResult}, bool, CancellationToken)" />
         public async Task<TResult> ExecuteAsync<TResult>(
             ICommand<TResult> commandMessage,
-            bool throwIfUnhandled) =>
-            (await _publisher.PublishAsync<TResult>(commandMessage, throwIfUnhandled)
+            bool throwIfUnhandled,
+            CancellationToken cancellationToken = default) =>
+            (await _publisher.PublishAsync<TResult>(commandMessage, throwIfUnhandled, cancellationToken)
                 .ConfigureAwait(false))
             .SingleOrDefault();
     }

diff --git a/src/Silverback.Core.Model/Messaging/Publishing/EventPublisher.cs b/src/Silverback.Core.Model/Messaging/Publishing/EventPublisher.cs
@@ -1,6 +1,7 @@
 // Copyright (c) 2020 Sergio Aquilini
 // This code is licensed under MIT license (see LICENSE file for details)
 
+using System.Threading;
 using System.Threading.Tasks;
 using Silverback.Messaging.Messages;
 
@@ -29,11 +30,15 @@ public EventPublisher(IPublisher publisher)
         public void Publish(IEvent eventMessage, bool throwIfUnhandled) =>
             _publisher.Publish(eventMessage, throwIfUnhandled);
 
-        /// <inheritdoc cref="IEventPublisher.PublishAsync(IEvent)" />
-        public Task PublishAsync(IEvent eventMessage) => _publisher.PublishAsync(eventMessage);
+        /// <inheritdoc cref="IEventPublisher.PublishAsync(IEvent, CancellationToken)" />
+        public Task PublishAsync(IEvent eventMessage, CancellationToken cancellationToken = default) =>
+            _publisher.PublishAsync(eventMessage, cancellationToken);
 
-        /// <inheritdoc cref="IEventPublisher.PublishAsync(IEvent, bool)" />
-        public Task PublishAsync(IEvent eventMessage, bool throwIfUnhandled) =>
-            _publisher.PublishAsync(eventMessage, throwIfUnhandled);
+        /// <inheritdoc cref="IEventPublisher.PublishAsync(IEvent, bool, CancellationToken)" />
+        public Task PublishAsync(
+            IEvent eventMessage,
+            bool throwIfUnhandled,
+            CancellationToken cancellationToken = default) =>
+            _publisher.PublishAsync(eventMessage, throwIfUnhandled, cancellationToken);
     }
 }
diff --git a/src/Silverback.Core.Model/Messaging/Publishing/ICommandPublisher.cs b/src/Silverback.Core.Model/Messaging/Publishing/ICommandPublisher.cs
@@ -1,6 +1,7 @@
 // Copyright (c) 2020 Sergio Aquilini
 // This code is licensed under MIT license (see LICENSE file for details)
 
+using System.Threading;
 using System.Threading.Tasks;
 using Silverback.Messaging.Messages;
 
@@ -80,10 +81,13 @@ public interface ICommandPublisher
         /// <param name="commandMessage">
         ///     The command to be executed.
         /// </param>
+        /// <param name="cancellationToken">
+        ///     A <see cref="CancellationToken" /> used to cancel the operation.
+        /// </param>
         /// <returns>
         ///     A <see cref="Task" /> representing the asynchronous operation.
         /// </returns>
-        Task ExecuteAsync(ICommand commandMessage);
+        Task ExecuteAsync(ICommand commandMessage, CancellationToken cancellationToken = default);
 
         /// <summary>
         ///     Executes the specified command publishing it to the internal bus. The message will be forwarded to
@@ -98,10 +102,13 @@ public interface ICommandPublisher
         ///     A boolean value indicating whether an exception must be thrown if no subscriber is handling the
         ///     message.
         /// </param>
+        /// <param name="cancellationToken">
+        ///     A <see cref="CancellationToken" /> used to cancel the operation.
+        /// </param>
         /// <returns>
         ///     A <see cref="Task" /> representing the asynchronous operation.
         /// </returns>
-        Task ExecuteAsync(ICommand commandMessage, bool throwIfUnhandled);
+        Task ExecuteAsync(ICommand commandMessage, bool throwIfUnhandled, CancellationToken cancellationToken = default);
 
         /// <summary>
         ///     Executes the specified command publishing it to the internal bus. The message will be forwarded to
@@ -115,11 +122,14 @@ public interface ICommandPublisher
         /// <param name="commandMessage">
         ///     The command to be executed.
         /// </param>
+        /// <param name="cancellationToken">
+        ///     A <see cref="CancellationToken" /> used to cancel the operation.
+        /// </param>
         /// <returns>
         ///     A <see cref="Task{TResult}" /> representing the asynchronous operation. The task result contains the
         ///     command result.
         /// </returns>
-        Task<TResult> ExecuteAsync<TResult>(ICommand<TResult> commandMessage);
+        Task<TResult> ExecuteAsync<TResult>(ICommand<TResult> commandMessage, CancellationToken cancellationToken = default);
 
         /// <summary>
         ///     Executes the specified command publishing it to the internal bus. The message will be forwarded to
@@ -137,10 +147,13 @@ public interface ICommandPublisher
         ///     A boolean value indicating whether an exception must be thrown if no subscriber is handling the
         ///     message.
         /// </param>
+        /// <param name="cancellationToken">
+        ///     A <see cref="CancellationToken" /> used to cancel the operation.
+        /// </param>
         /// <returns>
         ///     A <see cref="Task{TResult}" /> representing the asynchronous operation. The task result contains the
         ///     command result.
         /// </returns>
-        Task<TResult> ExecuteAsync<TResult>(ICommand<TResult> commandMessage, bool throwIfUnhandled);
+        Task<TResult> ExecuteAsync<TResult>(ICommand<TResult> commandMessage, bool throwIfUnhandled, CancellationToken cancellationToken = default);
     }
 }
diff --git a/src/Silverback.Core.Model/Messaging/Publishing/IEventPublisher.cs b/src/Silverback.Core.Model/Messaging/Publishing/IEventPublisher.cs
@@ -1,6 +1,7 @@
 // Copyright (c) 2020 Sergio Aquilini
 // This code is licensed under MIT license (see LICENSE file for details)
 
+using System.Threading;
 using System.Threading.Tasks;
 using Silverback.Messaging.Messages;
 
@@ -43,10 +44,13 @@ public interface IEventPublisher
         /// <param name="eventMessage">
         ///     The event to be executed.
         /// </param>
+        /// <param name="cancellationToken">
+        ///     A <see cref="CancellationToken" /> used to cancel the operation.
+        /// </param>
         /// <returns>
         ///     A <see cref="Task" /> representing the asynchronous operation.
         /// </returns>
-        Task PublishAsync(IEvent eventMessage);
+        Task PublishAsync(IEvent eventMessage, CancellationToken cancellationToken = default);
 
         /// <summary>
         ///     Publishes the specified event to the internal bus. The message will be forwarded to its subscribers
@@ -60,9 +64,15 @@ public interface IEventPublisher
         ///     A boolean value indicating whether an exception must be thrown if no subscriber is handling the
         ///     message.
         /// </param>
+        /// <param name="cancellationToken">
+        ///     A <see cref="CancellationToken" /> used to cancel the operation.
+        /// </param>
         /// <returns>
         ///     A <see cref="Task" /> representing the asynchronous operation.
         /// </returns>
-        Task PublishAsync(IEvent eventMessage, bool throwIfUnhandled);
+        Task PublishAsync(
+            IEvent eventMessage,
+            bool throwIfUnhandled,
+            CancellationToken cancellationToken = default);
     }
 }
diff --git a/src/Silverback.Core.Model/Messaging/Publishing/IQueryPublisher.cs b/src/Silverback.Core.Model/Messaging/Publishing/IQueryPublisher.cs
@@ -1,6 +1,7 @@
 // Copyright (c) 2020 Sergio Aquilini
 // This code is licensed under MIT license (see LICENSE file for details)
 
+using System.Threading;
 using System.Threading.Tasks;
 using Silverback.Messaging.Messages;
 
@@ -58,11 +59,16 @@ public interface IQueryPublisher
         /// <param name="queryMessage">
         ///     The query to be executed.
         /// </param>
+        /// <param name="cancellationToken">
+        ///     A <see cref="CancellationToken" /> used to cancel the operation.
+        /// </param>
         /// <returns>
         ///     A <see cref="Task{TResult}" /> representing the asynchronous operation. The task result contains the
         ///     query result.
         /// </returns>
-        Task<TResult> ExecuteAsync<TResult>(IQuery<TResult> queryMessage);
+        Task<TResult> ExecuteAsync<TResult>(
+            IQuery<TResult> queryMessage,
+            CancellationToken cancellationToken = default);
 
         /// <summary>
         ///     Executes the specified query publishing it to the internal bus. The message will be forwarded to its
@@ -79,10 +85,16 @@ public interface IQueryPublisher
         ///     A boolean value indicating whether an exception must be thrown if no subscriber is handling the
         ///     message.
         /// </param>
+        /// <param name="cancellationToken">
+        ///     A <see cref="CancellationToken" /> used to cancel the operation.
+        /// </param>
         /// <returns>
         ///     A <see cref="Task{TResult}" /> representing the asynchronous operation. The task result contains the
         ///     query result.
         /// </returns>
-        Task<TResult> ExecuteAsync<TResult>(IQuery<TResult> queryMessage, bool throwIfUnhandled);
+        Task<TResult> ExecuteAsync<TResult>(
+            IQuery<TResult> queryMessage,
+            bool throwIfUnhandled,
+            CancellationToken cancellationToken = default);
     }
 }
diff --git a/src/Silverback.Core.Model/Messaging/Publishing/QueryPublisher.cs b/src/Silverback.Core.Model/Messaging/Publishing/QueryPublisher.cs
@@ -2,6 +2,7 @@
 // This code is licensed under MIT license (see LICENSE file for details)
 
 using System.Linq;
+using System.Threading;
 using System.Threading.Tasks;
 using Silverback.Messaging.Messages;
 
@@ -31,17 +32,20 @@ public TResult Execute<TResult>(IQuery<TResult> queryMessage) =>
         public TResult Execute<TResult>(IQuery<TResult> queryMessage, bool throwIfUnhandled) =>
             _publisher.Publish<TResult>(queryMessage, throwIfUnhandled).SingleOrDefault();
 
-        /// <inheritdoc cref="IQueryPublisher.ExecuteAsync{TResult}(IQuery{TResult})" />
-        public async Task<TResult> ExecuteAsync<TResult>(IQuery<TResult> queryMessage) =>
-            (await _publisher.PublishAsync<TResult>(queryMessage)
+        /// <inheritdoc cref="IQueryPublisher.ExecuteAsync{TResult}(IQuery{TResult}, CancellationToken)" />
+        public async Task<TResult> ExecuteAsync<TResult>(
+            IQuery<TResult> queryMessage,
+            CancellationToken cancellationToken = default) =>
+            (await _publisher.PublishAsync<TResult>(queryMessage, cancellationToken)
                 .ConfigureAwait(false))
             .SingleOrDefault();
 
-        /// <inheritdoc cref="IQueryPublisher.ExecuteAsync{TResult}(IQuery{TResult}, bool)" />
+        /// <inheritdoc cref="IQueryPublisher.ExecuteAsync{TResult}(IQuery{TResult}, bool, CancellationToken)" />
         public async Task<TResult> ExecuteAsync<TResult>(
             IQuery<TResult> queryMessage,
-            bool throwIfUnhandled) =>
-            (await _publisher.PublishAsync<TResult>(queryMessage, throwIfUnhandled)
+            bool throwIfUnhandled,
+            CancellationToken cancellationToken = default) =>
+            (await _publisher.PublishAsync<TResult>(queryMessage, throwIfUnhandled, cancellationToken)
                 .ConfigureAwait(false))
             .SingleOrDefault();
     }

diff --git a/...Core.Rx/Messaging/Subscribers/ReturnValueHandlers/ObservableMessagesReturnValueHandler.cs b/...Core.Rx/Messaging/Subscribers/ReturnValueHandlers/ObservableMessagesReturnValueHandler.cs
@@ -4,6 +4,7 @@
 using System;
 using System.Linq;
 using System.Reactive.Linq;
+using System.Threading;
 using System.Threading.Tasks;
 using Silverback.Messaging.Configuration;
 using Silverback.Messaging.Publishing;
@@ -35,7 +36,7 @@ public ObservableMessagesReturnValueHandler(IPublisher publisher, IBusOptions bu
         }
 
         /// <inheritdoc cref="IReturnValueHandler.CanHandle" />
-        public bool CanHandle(object returnValue) =>
+        public bool CanHandle(object? returnValue) =>
             returnValue != null &&
             returnValue.GetType().GetInterfaces().Any(
                 i => i.IsGenericType &&
@@ -45,11 +46,11 @@ public bool CanHandle(object returnValue) =>
                              messageType.IsAssignableFrom(i.GenericTypeArguments[0])));
 
         /// <inheritdoc cref="IReturnValueHandler.Handle" />
-        public void Handle(object returnValue) =>
-            _publisher.Publish<object>(((IObservable<object>)returnValue).ToEnumerable());
+        public void Handle(object? returnValue) =>
+            _publisher.Publish<object?>(((IObservable<object?>)returnValue!).ToEnumerable());
 
         /// <inheritdoc cref="IReturnValueHandler.HandleAsync" />
-        public Task HandleAsync(object returnValue) =>
-            _publisher.PublishAsync<object>(((IObservable<object>)returnValue).ToEnumerable());
+        public Task HandleAsync(object? returnValue, CancellationToken cancellationToken = default) =>
+            _publisher.PublishAsync<object>(((IObservable<object?>)returnValue!).ToEnumerable(), cancellationToken);
     }
 }