| | | 1 | | // Binds the shared database-channel machinery (src/Channels/Shared/DbChannelShared.cs, compiled |
| | | 2 | | // into this project) to this provider's concrete seam types. See the note atop the shared file. |
| | | 3 | | global using DbChannelStore = AsyncResponse.Channels.MongoDB.MongoDbChannelStore; |
| | | 4 | | global using DbChannelMessage = AsyncResponse.Channels.MongoDB.MongoDbChannelMessage; |
| | | 5 | | global using DbChannelOptions = AsyncResponse.Channels.MongoDB.MongoDbAsyncResponseChannelOptions; |
| | | 6 | | |
| | | 7 | | using Microsoft.Extensions.DependencyInjection; |
| | | 8 | | using Microsoft.Extensions.Logging; |
| | | 9 | | using Microsoft.Extensions.Options; |
| | | 10 | | |
| | | 11 | | namespace AsyncResponse.Channels.MongoDB; |
| | | 12 | | |
| | | 13 | | /// <summary> |
| | | 14 | | /// MongoDB-backed response channel using change streams for active waiter wakeups and TTL-indexed |
| | | 15 | | /// collections for durable recovery state. Requires the server to run as a replica set (a |
| | | 16 | | /// single-node replica set is sufficient) for change-stream wakes; without one the channel degrades |
| | | 17 | | /// to interval polling. |
| | | 18 | | /// </summary> |
| | | 19 | | internal sealed class MongoDbAsyncResponseChannel : DbAsyncResponseChannelBase |
| | | 20 | | { |
| | | 21 | | /// <summary>Creates a MongoDB-backed async-response channel.</summary> |
| | | 22 | | public MongoDbAsyncResponseChannel( |
| | | 23 | | IServiceScopeFactory scopeFactory, |
| | | 24 | | MongoDbChannelStore store, |
| | | 25 | | IRecoveryStateStore recoveryStateStore, |
| | | 26 | | IOptions<MongoDbAsyncResponseChannelOptions> options, |
| | | 27 | | AsyncResponseContextPropagation propagation, |
| | | 28 | | ILogger<MongoDbAsyncResponseChannel> logger, |
| | | 29 | | TimeProvider? timeProvider = null) |
| | 503 | 30 | | : base( |
| | 503 | 31 | | scopeFactory, |
| | 503 | 32 | | store, |
| | 503 | 33 | | recoveryStateStore, |
| | 503 | 34 | | options.Value, |
| | 503 | 35 | | propagation, |
| | 503 | 36 | | logger, |
| | 503 | 37 | | channelTypeName: nameof(MongoDbAsyncResponseChannel), |
| | 503 | 38 | | providerName: "MongoDB", |
| | 503 | 39 | | activityTag: "mongodb", |
| | 503 | 40 | | subscriberRecordNoun: "document", |
| | 503 | 41 | | localDispatchRetryHint: "listener retry will pick it up", |
| | 503 | 42 | | timeProvider) |
| | | 43 | | { |
| | 503 | 44 | | } |
| | | 45 | | |
| | | 46 | | /// <inheritdoc /> |
| | 7390 | 47 | | protected override string ChannelName(string correlationId) => $"{_options.MessageCollection}:{correlationId}"; |
| | | 48 | | |
| | | 49 | | /// <inheritdoc /> |
| | 6597 | 50 | | protected override TimeSpan CurrentPollInterval() => _options.ListenerPollInterval; |
| | | 51 | | |
| | | 52 | | // Set once the server reports change streams unsupported (a standalone). Read on every poll |
| | | 53 | | // tick, so volatile. |
| | | 54 | | private volatile bool _changeStreamsUnavailable; |
| | | 55 | | |
| | | 56 | | /// <summary> |
| | | 57 | | /// The throttled sweep applies only while change streams carry normal delivery. With |
| | | 58 | | /// <see cref="MongoDbAsyncResponseChannelOptions.UseChangeStreams"/> off, or once the server |
| | | 59 | | /// has reported them unsupported, the sweep is the ONLY cross-process wake — and the 5s |
| | | 60 | | /// default equalled <c>DeliveryConfirmationTimeout</c>, so the publisher gave up, claimed the |
| | | 61 | | /// message for recovery and fired the lost-subscriber callback a beat before the healthy |
| | | 62 | | /// waiter's sweep found it. Every tick, as the <c>UseChangeStreams</c> doc promises. |
| | | 63 | | /// </summary> |
| | | 64 | | protected override TimeSpan? CurrentFullSweepInterval() |
| | 3248 | 65 | | => _options.UseChangeStreams && !_changeStreamsUnavailable ? _options.FullSweepInterval : null; |
| | | 66 | | |
| | | 67 | | /// <inheritdoc /> |
| | | 68 | | protected override Task? StartWakeListener(CancellationToken cancellationToken) |
| | 378 | 69 | | => _options.UseChangeStreams |
| | 357 | 70 | | ? Task.Run(() => ListenLoopAsync(cancellationToken)) |
| | 378 | 71 | | : Task.CompletedTask; |
| | | 72 | | |
| | | 73 | | /// <inheritdoc /> |
| | | 74 | | protected override IAsyncResponseWaiter<T> CreateWaiter<T>(Task<T> responseTask, Func<ValueTask> cleanupAsync) |
| | 387 | 75 | | => new MongoDbAsyncResponseWaiter<T>(responseTask, cleanupAsync); |
| | | 76 | | |
| | | 77 | | private async Task ListenLoopAsync(CancellationToken cancellationToken) |
| | | 78 | | { |
| | 357 | 79 | | var failures = 0; |
| | 357 | 80 | | while (!cancellationToken.IsCancellationRequested) |
| | | 81 | | { |
| | | 82 | | try |
| | | 83 | | { |
| | 356 | 84 | | await _store.WatchMessagesAsync(payload => |
| | 356 | 85 | | { |
| | 472 | 86 | | SignalDispatcher(string.IsNullOrEmpty(payload) ? null : payload); |
| | 472 | 87 | | return Task.CompletedTask; |
| | 356 | 88 | | }, cancellationToken).ConfigureAwait(false); |
| | 0 | 89 | | failures = 0; |
| | 0 | 90 | | } |
| | 356 | 91 | | catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) |
| | | 92 | | { |
| | 356 | 93 | | return; |
| | | 94 | | } |
| | 0 | 95 | | catch (Exception ex) when (MongoDbChannelStore.IsChangeStreamUnsupported(ex)) |
| | | 96 | | { |
| | | 97 | | // Standalone server: change streams need a replica set. The dispatch loop's |
| | | 98 | | // ListenerPollInterval sweep still delivers, so degrade to polling instead of |
| | | 99 | | // retry-spamming an error the server will keep returning. Flagged so the sweep |
| | | 100 | | // throttle lifts (see CurrentFullSweepInterval): it is now the only wake. |
| | 0 | 101 | | _changeStreamsUnavailable = true; |
| | 0 | 102 | | _logger.LogWarning( |
| | 0 | 103 | | ex, |
| | 0 | 104 | | "MongoDB change streams are unavailable (the server is not a replica set); response wakes fall back |
| | 0 | 105 | | _options.ListenerPollInterval); |
| | 0 | 106 | | return; |
| | | 107 | | } |
| | 0 | 108 | | catch (Exception ex) |
| | | 109 | | { |
| | 0 | 110 | | failures++; |
| | 0 | 111 | | var delay = AsyncResponseRetry.Backoff(failures, TimeSpan.FromMilliseconds(100), TimeSpan.FromSeconds(5) |
| | 0 | 112 | | _logger.LogWarning(ex, "MongoDB change-stream loop failed; retrying in {Delay}.", delay); |
| | 0 | 113 | | await Task.Delay(delay, cancellationToken).ConfigureAwait(false); |
| | | 114 | | } |
| | | 115 | | } |
| | 357 | 116 | | } |
| | | 117 | | } |