| | | 1 | | using Elsa.Mediator.Contracts; |
| | | 2 | | using Microsoft.Extensions.Logging; |
| | | 3 | | |
| | | 4 | | namespace Elsa.Workflows.Runtime; |
| | | 5 | | |
| | | 6 | | /// <inheritdoc /> |
| | 724 | 7 | | public class WorkflowCommitNotificationBuffer(IMediator mediator, ILogger<WorkflowCommitNotificationBuffer> logger) : IW |
| | | 8 | | { |
| | 724 | 9 | | private readonly IMediator _mediator = mediator; |
| | 724 | 10 | | private readonly ILogger<WorkflowCommitNotificationBuffer> _logger = logger; |
| | 724 | 11 | | private readonly AsyncLocal<Scope?> _currentScope = new(); |
| | | 12 | | |
| | | 13 | | /// <inheritdoc /> |
| | | 14 | | public IWorkflowCommitNotificationScope Begin() |
| | | 15 | | { |
| | 562 | 16 | | var scope = new Scope(this, _currentScope.Value); |
| | 562 | 17 | | _currentScope.Value = scope; |
| | 562 | 18 | | return scope; |
| | | 19 | | } |
| | | 20 | | |
| | | 21 | | internal bool TryAdd(INotification notification, IEventPublishingStrategy? strategy) |
| | | 22 | | { |
| | 17654 | 23 | | var scope = _currentScope.Value; |
| | 17654 | 24 | | if (scope == null) |
| | 14795 | 25 | | return false; |
| | | 26 | | |
| | 2859 | 27 | | scope.Add(notification, strategy); |
| | 2859 | 28 | | return true; |
| | | 29 | | } |
| | | 30 | | |
| | 562 | 31 | | private class Scope(WorkflowCommitNotificationBuffer owner, Scope? parent) : IWorkflowCommitNotificationScope |
| | | 32 | | { |
| | 562 | 33 | | private readonly List<Entry> _entries = []; |
| | | 34 | | private bool _disposed; |
| | | 35 | | |
| | | 36 | | public void Add(INotification notification, IEventPublishingStrategy? strategy) |
| | | 37 | | { |
| | 2859 | 38 | | _entries.Add(new(notification, strategy)); |
| | 2859 | 39 | | } |
| | | 40 | | |
| | | 41 | | public Task FlushAsync(CancellationToken cancellationToken = default) |
| | | 42 | | { |
| | 560 | 43 | | ThrowIfDisposed(); |
| | 560 | 44 | | owner._currentScope.Value = parent; |
| | 560 | 45 | | return FlushEntriesAsync(cancellationToken); |
| | | 46 | | } |
| | | 47 | | |
| | | 48 | | private async Task FlushEntriesAsync(CancellationToken cancellationToken) |
| | | 49 | | { |
| | 560 | 50 | | List<Exception>? exceptions = null; |
| | | 51 | | |
| | 6835 | 52 | | foreach (var entry in _entries) |
| | | 53 | | { |
| | | 54 | | try |
| | | 55 | | { |
| | 2858 | 56 | | await owner._mediator.SendAsync(entry.Notification, entry.Strategy, cancellationToken); |
| | 2855 | 57 | | } |
| | 3 | 58 | | catch (Exception ex) when (ex is not OperationCanceledException and not OutOfMemoryException and not Sta |
| | | 59 | | { |
| | 2 | 60 | | owner._logger.LogError(ex, "Failed to publish buffered workflow commit notification {NotificationTyp |
| | 2 | 61 | | exceptions ??= []; |
| | 2 | 62 | | exceptions.Add(ex); |
| | 2 | 63 | | } |
| | 2857 | 64 | | } |
| | | 65 | | |
| | 559 | 66 | | _entries.Clear(); |
| | | 67 | | |
| | 559 | 68 | | if (exceptions is { Count: > 0 }) |
| | 2 | 69 | | throw new AggregateException("One or more workflow commit notifications failed.", exceptions); |
| | 557 | 70 | | } |
| | | 71 | | |
| | | 72 | | public void Dispose() |
| | | 73 | | { |
| | 562 | 74 | | if (_disposed) |
| | 0 | 75 | | return; |
| | | 76 | | |
| | 562 | 77 | | _disposed = true; |
| | 562 | 78 | | if (ReferenceEquals(owner._currentScope.Value, this)) |
| | 4 | 79 | | owner._currentScope.Value = parent; |
| | 562 | 80 | | } |
| | | 81 | | |
| | | 82 | | private void ThrowIfDisposed() |
| | | 83 | | { |
| | 560 | 84 | | if (_disposed) |
| | 0 | 85 | | throw new ObjectDisposedException(nameof(IWorkflowCommitNotificationScope)); |
| | 560 | 86 | | } |
| | | 87 | | } |
| | | 88 | | |
| | 8577 | 89 | | private record Entry(INotification Notification, IEventPublishingStrategy? Strategy); |
| | | 90 | | } |