< Summary - AsyncResponse (Release / net8.0+net10.0 / unit+integration)

Information
Class: AsyncResponse.Channels.MongoDB.MongoDbAsyncResponseChannelOptions
Assembly: AsyncResponse.Channels.MongoDB
File(s): /_/src/Channels/AsyncResponse.Channels.MongoDB/MongoDbAsyncResponseChannelOptions.cs
Line coverage
100%
Covered lines: 67
Uncovered lines: 0
Coverable lines: 67
Total lines: 215
Line coverage: 100%
Branch coverage
95%
Covered branches: 23
Total branches: 24
Branch coverage: 95.8%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
get_ConnectionString()100%11100%
get_DatabaseName()100%11100%
get_RecoveryStateCollection()100%11100%
get_MessageCollection()100%11100%
get_SubscriberCollection()100%11100%
get_AutoCreateIndexes()100%11100%
get_UseOwnershipLedger()100%11100%
get_UseChangeStreams()100%11100%
get_MessageRetention()100%11100%
get_DeliveryConfirmationTimeout()100%11100%
get_DeliveryConfirmationPollInterval()100%11100%
get_ListenerPollInterval()100%11100%
get_FullSweepInterval()100%11100%
get_PendingMessageBatchSize()100%11100%
get_HistoryReconciliationInterval()100%11100%
get_SubscriberHeartbeatInterval()100%11100%
get_SubscriberHeartbeatTimeout()100%11100%
get_PublishMaxAttempts()100%11100%
get_PublishRetryBaseDelay()100%11100%
get_PublishRetryMaxDelay()100%11100%
Validate()95.83%2424100%

File(s)

/_/src/Channels/AsyncResponse.Channels.MongoDB/MongoDbAsyncResponseChannelOptions.cs

#LineLine coverage
 1namespace AsyncResponse.Channels.MongoDB;
 2
 3/// <summary>
 4/// Options for the MongoDB-backed async-response channel.
 5/// <para>
 6/// Active waiters are woken with a MongoDB change stream watching inserts to the response-message
 7/// collection; the change event carries the correlation id, so only the signaled waiter's messages
 8/// are scanned. Response envelopes are stored as documents, and durable <see cref="RecoveryState"/>
 9/// entries live in a TTL-indexed collection so late responses can resume or fail flows after the
 10/// original waiter process dies. Change streams require the server to run as a replica set (a
 11/// single-node replica set is sufficient); without one the channel degrades to interval polling.
 12/// </para>
 13/// </summary>
 14public sealed class MongoDbAsyncResponseChannelOptions : DurableAsyncResponseChannelOptions
 15{
 16    /// <summary>The channel name reported to the startup validator.</summary>
 17    public const string ChannelName = "MongoDB";
 18
 19    /// <summary>
 20    /// Optional MongoDB connection string used when no <c>IMongoDatabase</c> or <c>IMongoClient</c>
 21    /// is registered with the host.
 22    /// </summary>
 1023    public string? ConnectionString { get; set; }
 24
 25    /// <summary>Optional database name used when no <c>IMongoDatabase</c> is registered.</summary>
 46826    public string? DatabaseName { get; set; }
 27
 28    /// <summary>
 29    /// Collection storing durable recovery registrations. Each waiter registration is one document
 30    /// keyed by correlation id and registration id, expired natively by a TTL index.
 31    /// </summary>
 770132    public string RecoveryStateCollection { get; set; } = "asyncresponse_recovery_state";
 33
 34    /// <summary>
 35    /// Collection storing response envelopes until they expire. The change stream watches inserts to
 36    /// this collection; waiters load the envelope documents from it.
 37    /// </summary>
 1690638    public string MessageCollection { get; set; } = "asyncresponse_channel_messages";
 39
 40    /// <summary>
 41    /// Collection storing short-lived live-subscriber heartbeats for watchdog liveness and the
 42    /// publish fast path, expired natively by a TTL index.
 43    /// </summary>
 759244    public string SubscriberCollection { get; set; } = "asyncresponse_channel_subscribers";
 45
 46    /// <summary>
 47    /// Creates the TTL and lookup indexes on first use. Disable when provisioning owns index DDL.
 48    /// </summary>
 122149    public bool AutoCreateIndexes { get; set; } = true;
 50
 51    /// <summary>
 52    /// Claims this component's collections (the derived ack-counter collection included) in the
 53    /// persisted cross-component ownership ledger (<c>asyncresponse_ownership</c>) at first use,
 54    /// so another AsyncResponse component — in this or any other process — misconfigured onto
 55    /// the same collection fails startup instead of silently corrupting data. Independent of
 56    /// <see cref="AutoCreateIndexes"/>: disabling index DDL must not disable collision
 57    /// protection. Disable only for least-privilege deployments that cannot write the ledger
 58    /// collection and audit their collection layout externally. Default: <c>true</c>.
 59    /// </summary>
 114260    public bool UseOwnershipLedger { get; set; } = true;
 61
 62    /// <summary>
 63    /// Watches the message collection with a change stream so active waiters are woken with
 64    /// broker-grade latency. Requires a replica set. When disabled — or when the server reports
 65    /// change streams as unsupported — waiters fall back to <see cref="ListenerPollInterval"/>
 66    /// polling. Default: <c>true</c>.
 67    /// </summary>
 436368    public bool UseChangeStreams { get; set; } = true;
 69
 70    /// <summary>
 71    /// How long response-envelope documents are retained for active waiter delivery and
 72    /// missed-notification recovery. Expired documents are reaped by the TTL index.
 73    /// </summary>
 502474    public TimeSpan MessageRetention { get; set; } = TimeSpan.FromHours(1);
 75
 76    /// <summary>
 77    /// How long a publisher waits for a live waiter to acknowledge loading a response envelope
 78    /// before treating the response as lost-subscriber delivery. Default: 5 seconds.
 79    /// </summary>
 384480    public TimeSpan DeliveryConfirmationTimeout { get; set; } = TimeSpan.FromSeconds(5);
 81
 82    /// <summary>
 83    /// Poll interval used while a publisher waits for delivery acknowledgement. Default: 50 ms.
 84    /// </summary>
 330185    public TimeSpan DeliveryConfirmationPollInterval { get; set; } = TimeSpan.FromMilliseconds(50);
 86
 87    /// <summary>
 88    /// Fallback poll interval used by the dispatch loop to catch messages if a change-stream event
 89    /// is missed during reconnect (or change streams are unavailable). Default: 250 ms.
 90    /// </summary>
 877591    public TimeSpan ListenerPollInterval { get; set; } = TimeSpan.FromMilliseconds(250);
 92
 93    /// <summary>
 94    /// Minimum interval between full safety-net sweeps. A full sweep queries the store once per
 95    /// subscribed correlation id, so its idle cost is W queries per <see cref="ListenerPollInterval"/>
 96    /// tick with W in-flight waiters. While change streams carry normal delivery the sweep only
 97    /// covers wakes lost in failure windows, so this bounds idle database load without touching
 98    /// normal delivery latency — it stretches only the worst-case recovery of a LOST wake. It is
 99    /// ignored (the sweep runs on every <see cref="ListenerPollInterval"/> tick) whenever change
 100    /// streams are not carrying delivery: <see cref="UseChangeStreams"/> off, or a server that
 101    /// reports them unsupported — there the sweep is the only cross-process wake, and a throttle
 102    /// equal to <see cref="DeliveryConfirmationTimeout"/> routed live waiters into recovery.
 103    /// Default: 5 seconds (an unbounded null swept every waiter on every 250 ms tick — W
 104    /// sequential queries per tick of pure idle load). Set null to sweep on every poll tick.
 105    /// </summary>
 4905106    public TimeSpan? FullSweepInterval { get; set; } = TimeSpan.FromSeconds(5);
 107
 108    /// <summary>
 109    /// Number of pending response messages loaded per subscribed correlation id per dispatch pass.
 110    /// Default: 64.
 111    /// </summary>
 4720112    public int PendingMessageBatchSize { get; set; } = 64;
 113
 114    /// <summary>
 115    /// Minimum interval between incremental reconciliation passes over retained message history.
 116    /// Normal scans keep their forward cursor; reconciliation catches late commits behind it,
 117    /// including rows already acknowledged by another process. One history page is read per
 118    /// dispatch pass so a long history cannot monopolize delivery to other correlations.
 119    /// Default: 5 seconds; large histories take additional poll intervals to reconcile.
 120    /// </summary>
 2095121    public TimeSpan HistoryReconciliationInterval { get; set; } = TimeSpan.FromSeconds(5);
 122
 123
 124    /// <summary>
 125    /// How often a live waiter refreshes its subscriber heartbeat document. Default: 10 seconds.
 126    /// </summary>
 5029127    public TimeSpan SubscriberHeartbeatInterval { get; set; } = TimeSpan.FromSeconds(10);
 128
 129    /// <summary>
 130    /// How long a subscriber heartbeat remains live without refresh. Keep this above
 131    /// <see cref="SubscriberHeartbeatInterval"/>. Default: 30 seconds.
 132    /// </summary>
 3890133    public TimeSpan SubscriberHeartbeatTimeout { get; set; } = TimeSpan.FromSeconds(30);
 134
 135    /// <summary>Maximum attempts for a response-document insert. Set to 1 to disable publish retries. Default: 3.</summ
 2226136    public int PublishMaxAttempts { get; set; } = 3;
 137
 138    /// <summary>Initial delay before retrying a failed response-document insert. Default: 50 ms.</summary>
 3273139    public TimeSpan PublishRetryBaseDelay { get; set; } = TimeSpan.FromMilliseconds(50);
 140
 141    /// <summary>Maximum delay between response-document insert retries. Default: 1 second.</summary>
 3275142    public TimeSpan PublishRetryMaxDelay { get; set; } = TimeSpan.FromSeconds(1);
 143
 144    /// <summary>Validates the option values and throws on misconfiguration.</summary>
 145    public void Validate()
 146    {
 147        // Shared channel knobs (RecoveryStateExpiry, DefaultTimeout, DisposalDrainTimeout) go
 148        // through the ONE base guard set — a bespoke duplicate here silently missed every knob
 149        // added to the base later (DisposalDrainTimeout was validated nowhere on this provider).
 1099150        ValidateShared(nameof(MongoDbAsyncResponseChannelOptions));
 151
 1095152        MongoDbChannelStore.ValidateCollectionName(RecoveryStateCollection, nameof(RecoveryStateCollection));
 1093153        MongoDbChannelStore.ValidateCollectionName(MessageCollection, nameof(MessageCollection));
 1091154        MongoDbChannelStore.ValidateCollectionName(SubscriberCollection, nameof(SubscriberCollection));
 155
 1087156        if (StringComparer.Ordinal.Equals(RecoveryStateCollection, MessageCollection)
 1087157            || StringComparer.Ordinal.Equals(RecoveryStateCollection, SubscriberCollection)
 1087158            || StringComparer.Ordinal.Equals(MessageCollection, SubscriberCollection))
 159        {
 2160            throw new InvalidOperationException(
 2161                $"{nameof(MongoDbAsyncResponseChannelOptions)}.{nameof(RecoveryStateCollection)}, " +
 2162                $"{nameof(MessageCollection)}, and {nameof(SubscriberCollection)} must be distinct collections.");
 163        }
 164
 165        // The derived "{MessageCollection}_counters" collection is part of the effective name
 166        // plan: a configured collection occupying it would receive the ack-counter document —
 167        // in the TTL-indexed recovery collection the reaper would silently delete the counter
 168        // and reset the same-tick delivery tie-breaker.
 1085169        var countersCollection = MongoDbChannelStore.CountersCollectionName(MessageCollection);
 1085170        if (StringComparer.Ordinal.Equals(RecoveryStateCollection, countersCollection)
 1085171            || StringComparer.Ordinal.Equals(SubscriberCollection, countersCollection))
 172        {
 4173            throw new InvalidOperationException(
 4174                $"{nameof(MongoDbAsyncResponseChannelOptions)}: '{countersCollection}' is reserved for the ack counter "
 4175                $"(derived from {nameof(MessageCollection)}); {nameof(RecoveryStateCollection)} and {nameof(SubscriberCo
 176        }
 177
 1081178        EnsurePersistedTtl(MessageRetention, nameof(MongoDbAsyncResponseChannelOptions), nameof(MessageRetention));
 1077179        EnsurePersistedTtl(DeliveryConfirmationTimeout, nameof(MongoDbAsyncResponseChannelOptions), nameof(DeliveryConfi
 1073180        if (MessageRetention <= DeliveryConfirmationTimeout)
 2181            throw new InvalidOperationException(
 2182                $"{nameof(MongoDbAsyncResponseChannelOptions)}.{nameof(MessageRetention)} must exceed " +
 2183                $"{nameof(DeliveryConfirmationTimeout)}: a message document pruned inside the confirmation window is " +
 2184                "indistinguishable from an acknowledged one, so the response would be reported delivered and " +
 2185                "lost-response recovery silently skipped.");
 1071186        EnsureTimerBacked(DeliveryConfirmationPollInterval, nameof(MongoDbAsyncResponseChannelOptions), nameof(DeliveryC
 1067187        EnsureTimerBacked(ListenerPollInterval, nameof(MongoDbAsyncResponseChannelOptions), nameof(ListenerPollInterval)
 1063188        EnsureTimerBacked(HistoryReconciliationInterval, nameof(MongoDbAsyncResponseChannelOptions), nameof(HistoryRecon
 1061189        if (FullSweepInterval is { } fullSweepInterval)
 957190            EnsureTimerBacked(fullSweepInterval, nameof(MongoDbAsyncResponseChannelOptions), nameof(FullSweepInterval));
 1061191        EnsureTimerBacked(SubscriberHeartbeatInterval, nameof(MongoDbAsyncResponseChannelOptions), nameof(SubscriberHear
 1061192        EnsurePersistedTtl(SubscriberHeartbeatTimeout, nameof(MongoDbAsyncResponseChannelOptions), nameof(SubscriberHear
 193
 1059194        if (MaxRemoteStackTraceLength < 0)
 2195            throw new InvalidOperationException($"{nameof(MongoDbAsyncResponseChannelOptions)}.{nameof(MaxRemoteStackTra
 196
 1057197        if (PendingMessageBatchSize <= 0)
 2198            throw new InvalidOperationException($"{nameof(MongoDbAsyncResponseChannelOptions)}.{nameof(PendingMessageBat
 199
 1055200        if (SubscriberHeartbeatInterval >= SubscriberHeartbeatTimeout)
 2201            throw new InvalidOperationException(
 2202                $"{nameof(MongoDbAsyncResponseChannelOptions)}.{nameof(SubscriberHeartbeatInterval)} must be less than "
 2203                $"{nameof(MongoDbAsyncResponseChannelOptions)}.{nameof(SubscriberHeartbeatTimeout)}.");
 204
 1053205        if (PublishMaxAttempts <= 0)
 2206            throw new InvalidOperationException($"{nameof(MongoDbAsyncResponseChannelOptions)}.{nameof(PublishMaxAttempt
 207
 1051208        EnsureTimerBacked(PublishRetryBaseDelay, nameof(MongoDbAsyncResponseChannelOptions), nameof(PublishRetryBaseDela
 1051209        EnsureTimerBacked(PublishRetryMaxDelay, nameof(MongoDbAsyncResponseChannelOptions), nameof(PublishRetryMaxDelay)
 1049210        if (PublishRetryBaseDelay > PublishRetryMaxDelay)
 2211            throw new InvalidOperationException(
 2212                $"{nameof(MongoDbAsyncResponseChannelOptions)}.{nameof(PublishRetryBaseDelay)} cannot exceed {nameof(Pub
 1047213    }
 214
 215}
Generated by: ReportGenerator 5.5.11.0
09/22/2026 - 09:41:04
GitHub | reportgenerator.io

Methods/Properties

get_ConnectionString()
get_DatabaseName()
get_RecoveryStateCollection()
get_MessageCollection()
get_SubscriberCollection()
get_AutoCreateIndexes()
get_UseOwnershipLedger()
get_UseChangeStreams()
get_MessageRetention()
get_DeliveryConfirmationTimeout()
get_DeliveryConfirmationPollInterval()
get_ListenerPollInterval()
get_FullSweepInterval()
get_PendingMessageBatchSize()
get_HistoryReconciliationInterval()
get_SubscriberHeartbeatInterval()
get_SubscriberHeartbeatTimeout()
get_PublishMaxAttempts()
get_PublishRetryBaseDelay()
get_PublishRetryMaxDelay()
Validate()