| | | 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) |
| | 3 | 29 | | : base( |
| | 3 | 30 | | scopeFactory, |
| | 3 | 31 | | store, |
| | 3 | 32 | | recoveryStateStore, |
| | 3 | 33 | | options.Value, |
| | 3 | 34 | | propagation, |
| | 3 | 35 | | logger, |
| | 3 | 36 | | channelTypeName: nameof(MongoDbAsyncResponseChannel), |
| | 3 | 37 | | providerName: "MongoDB", |
| | 3 | 38 | | activityTag: "mongodb", |
| | 3 | 39 | | subscriberRecordNoun: "document", |
| | 3 | 40 | | localDispatchRetryHint: "listener retry will pick it up") |
| | | 41 | | { |
| | 3 | 42 | | } |
| | | 43 | | |
| | | 44 | | /// <inheritdoc /> |
| | 3 | 45 | | protected override string ChannelName(string correlationId) => $"{_options.MessageCollection}:{correlationId}"; |
| | | 46 | | |
| | | 47 | | /// <inheritdoc /> |
| | 3 | 48 | | protected override TimeSpan CurrentPollInterval() => _options.ListenerPollInterval; |
| | | 49 | | |
| | | 50 | | /// <inheritdoc /> |
| | | 51 | | protected override Task? StartWakeListener(CancellationToken cancellationToken) |
| | 3 | 52 | | => _options.UseChangeStreams |
| | 1 | 53 | | ? Task.Run(() => ListenLoopAsync(cancellationToken)) |
| | 3 | 54 | | : Task.CompletedTask; |
| | | 55 | | |
| | | 56 | | /// <inheritdoc /> |
| | | 57 | | protected override IAsyncResponseWaiter<T> CreateWaiter<T>(Task<T> responseTask, Func<ValueTask> cleanupAsync) |
| | 3 | 58 | | => new MongoDbAsyncResponseWaiter<T>(responseTask, cleanupAsync); |
| | | 59 | | |
| | | 60 | | private async Task ListenLoopAsync(CancellationToken cancellationToken) |
| | | 61 | | { |
| | 1 | 62 | | var failures = 0; |
| | 1 | 63 | | while (!cancellationToken.IsCancellationRequested) |
| | | 64 | | { |
| | | 65 | | try |
| | | 66 | | { |
| | 1 | 67 | | await _store.WatchMessagesAsync(payload => |
| | 1 | 68 | | { |
| | 1 | 69 | | SignalDispatcher(string.IsNullOrEmpty(payload) ? null : payload); |
| | 1 | 70 | | return Task.CompletedTask; |
| | 1 | 71 | | }, cancellationToken).ConfigureAwait(false); |
| | 1 | 72 | | failures = 0; |
| | 1 | 73 | | } |
| | 1 | 74 | | catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) |
| | | 75 | | { |
| | 1 | 76 | | return; |
| | | 77 | | } |
| | 0 | 78 | | catch (Exception ex) when (MongoDbChannelStore.IsChangeStreamUnsupported(ex)) |
| | | 79 | | { |
| | | 80 | | // Standalone server: change streams need a replica set. The dispatch loop's |
| | | 81 | | // ListenerPollInterval sweep still delivers, so degrade to polling instead of |
| | | 82 | | // retry-spamming an error the server will keep returning. |
| | 0 | 83 | | _logger.LogWarning( |
| | 0 | 84 | | ex, |
| | 0 | 85 | | "MongoDB change streams are unavailable (the server is not a replica set); response wakes fall back |
| | 0 | 86 | | _options.ListenerPollInterval); |
| | 0 | 87 | | return; |
| | | 88 | | } |
| | 0 | 89 | | catch (Exception ex) |
| | | 90 | | { |
| | 0 | 91 | | failures++; |
| | 0 | 92 | | var delay = AsyncResponseRetry.Backoff(failures, TimeSpan.FromMilliseconds(100), TimeSpan.FromSeconds(5) |
| | 0 | 93 | | _logger.LogWarning(ex, "MongoDB change-stream loop failed; retrying in {Delay}.", delay); |
| | 0 | 94 | | await Task.Delay(delay, cancellationToken).ConfigureAwait(false); |
| | | 95 | | } |
| | | 96 | | } |
| | 1 | 97 | | } |
| | | 98 | | } |