| | | 1 | | using Microsoft.Extensions.DependencyInjection; |
| | | 2 | | using Microsoft.Extensions.Logging; |
| | | 3 | | using ThrottleDebounce; |
| | | 4 | | |
| | | 5 | | namespace Elsa.Workflows.Runtime; |
| | | 6 | | |
| | | 7 | | public class BookmarkQueueWorker : IBookmarkQueueWorker |
| | | 8 | | { |
| | | 9 | | private readonly RateLimitedFunc<CancellationToken, Task> _rateLimitedProcessAsync; |
| | | 10 | | private CancellationTokenSource _cts = null!; |
| | | 11 | | private bool _running; |
| | | 12 | | private readonly IBookmarkQueueSignaler _signaler; |
| | | 13 | | private readonly IServiceScopeFactory _scopeFactory; |
| | | 14 | | private readonly ILogger<BookmarkQueueWorker> _logger; |
| | | 15 | | |
| | 11 | 16 | | public BookmarkQueueWorker(IBookmarkQueueSignaler signaler, IServiceScopeFactory scopeFactory, ILogger<BookmarkQueue |
| | | 17 | | { |
| | 11 | 18 | | _signaler = signaler; |
| | 11 | 19 | | _scopeFactory = scopeFactory; |
| | 11 | 20 | | _logger = logger; |
| | 11 | 21 | | _rateLimitedProcessAsync = Throttler.Throttle<CancellationToken, Task>(ProcessAsync, TimeSpan.FromMilliseconds(5 |
| | 11 | 22 | | } |
| | | 23 | | |
| | | 24 | | public void Start() |
| | | 25 | | { |
| | 11 | 26 | | if (_running) |
| | 0 | 27 | | return; |
| | | 28 | | |
| | 11 | 29 | | _cts = new(); |
| | 11 | 30 | | _running = true; |
| | | 31 | | |
| | 11 | 32 | | _ = Task.Run(AwaitSignalAsync); |
| | 11 | 33 | | } |
| | | 34 | | |
| | | 35 | | public void Stop() |
| | | 36 | | { |
| | 7 | 37 | | if (_running) |
| | | 38 | | { |
| | 7 | 39 | | _running = false; |
| | 7 | 40 | | _cts.Cancel(); |
| | | 41 | | } |
| | | 42 | | |
| | 7 | 43 | | _cts.Dispose(); |
| | 7 | 44 | | } |
| | | 45 | | |
| | | 46 | | private async Task AwaitSignalAsync() |
| | | 47 | | { |
| | 599 | 48 | | while (!_cts.IsCancellationRequested) |
| | | 49 | | { |
| | | 50 | | try |
| | | 51 | | { |
| | 598 | 52 | | await _signaler.AwaitAsync(_cts.Token); |
| | 588 | 53 | | await _rateLimitedProcessAsync.InvokeAsync(_cts.Token); |
| | 588 | 54 | | } |
| | 6 | 55 | | catch (OperationCanceledException) |
| | | 56 | | { |
| | 6 | 57 | | break; // Stop() was called |
| | | 58 | | } |
| | 0 | 59 | | catch (Exception ex) |
| | | 60 | | { |
| | 0 | 61 | | _logger.LogError(ex, "BookmarkQueueWorker error – continuing loop"); |
| | 0 | 62 | | } |
| | | 63 | | } |
| | 7 | 64 | | } |
| | | 65 | | |
| | | 66 | | protected virtual async Task ProcessAsync(CancellationToken cancellationToken) |
| | | 67 | | { |
| | 62 | 68 | | _logger.LogDebug("Processing bookmark queue..."); |
| | 62 | 69 | | using var scope = _scopeFactory.CreateScope(); |
| | 62 | 70 | | var processor = scope.ServiceProvider.GetRequiredService<IBookmarkQueueProcessor>(); |
| | 62 | 71 | | await processor.ProcessAsync(cancellationToken); |
| | 62 | 72 | | _logger.LogDebug("Processed bookmark queue."); |
| | 62 | 73 | | } |
| | | 74 | | } |