| | | 1 | | namespace MediatR; |
| | | 2 | | |
| | | 3 | | internal class WhenAllPublisher : INotificationPublisher |
| | | 4 | | { |
| | | 5 | | public async Task Publish( |
| | | 6 | | IEnumerable<NotificationHandlerExecutor> handlerExecutors, |
| | | 7 | | INotification notification, |
| | | 8 | | CancellationToken cancellationToken) |
| | | 9 | | { |
| | 9 | 10 | | List<Exception>? exceptions = null; |
| | 9 | 11 | | var tasks = new List<Task>(); |
| | | 12 | | |
| | | 13 | | // Some of the tasks may throw an immediate exception, synchronously, and we can't guarantee they're utilizing a |
| | | 14 | | // e.g. public Task Handle(CancellationToken cancellationToken) => throw new Exception(); |
| | | 15 | | // In that case, WhenAll won't even be executed. So, we must iterate and catch these exceptions beforehand. |
| | 60 | 16 | | foreach (var handlerExecutor in handlerExecutors) |
| | | 17 | | { |
| | | 18 | | try |
| | | 19 | | { |
| | 21 | 20 | | tasks.Add(handlerExecutor.HandlerCallback(notification, cancellationToken)); |
| | 16 | 21 | | } |
| | 2 | 22 | | catch (AggregateException ex) |
| | | 23 | | { |
| | 2 | 24 | | (exceptions ??= []).AddRange(ex.Flatten().InnerExceptions); |
| | 2 | 25 | | } |
| | 3 | 26 | | catch (Exception ex) |
| | | 27 | | { |
| | 3 | 28 | | (exceptions ??= []).Add(ex); |
| | 3 | 29 | | } |
| | | 30 | | } |
| | | 31 | | |
| | | 32 | | try |
| | | 33 | | { |
| | 9 | 34 | | await Task.WhenAll(tasks).ConfigureAwait(false); |
| | 4 | 35 | | } |
| | 5 | 36 | | catch |
| | | 37 | | { |
| | 34 | 38 | | foreach (var task in tasks) |
| | | 39 | | { |
| | 12 | 40 | | if (task.IsFaulted) |
| | | 41 | | { |
| | 5 | 42 | | if (task.Exception.InnerExceptions.Count > 1 || task.Exception.InnerException is AggregateException) |
| | | 43 | | { |
| | 3 | 44 | | (exceptions ??= []).AddRange(task.Exception.Flatten().InnerExceptions); |
| | | 45 | | } |
| | 2 | 46 | | else if (task.Exception.InnerException is not null) |
| | | 47 | | { |
| | 2 | 48 | | (exceptions ??= []).Add(task.Exception.InnerException); |
| | | 49 | | } |
| | | 50 | | } |
| | 7 | 51 | | else if (task.IsCanceled) |
| | | 52 | | { |
| | | 53 | | try |
| | | 54 | | { |
| | | 55 | | // This will force the task to throw the exception if it's canceled. |
| | | 56 | | // This will preserve all the information compared to creating a new TaskCanceledException manua |
| | 3 | 57 | | task.GetAwaiter().GetResult(); |
| | 0 | 58 | | } |
| | 3 | 59 | | catch (Exception ex) |
| | | 60 | | { |
| | 3 | 61 | | (exceptions ??= []).Add(ex); |
| | 3 | 62 | | } |
| | | 63 | | } |
| | | 64 | | } |
| | 5 | 65 | | } |
| | | 66 | | |
| | 9 | 67 | | if (exceptions?.Count > 0) |
| | | 68 | | { |
| | 7 | 69 | | throw new AggregateException(exceptions); |
| | | 70 | | } |
| | 2 | 71 | | } |
| | | 72 | | } |