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

Information
Class: AsyncResponse.Transports.SqlServer.SqlServerSubscriberOptions
Assembly: AsyncResponse.Transports.SqlServer
File(s): /home/runner/work/AsyncResponse/AsyncResponse/src/Transports/AsyncResponse.Transports.SqlServer/SqlServerAsyncResponseTransportOptions.cs
Line coverage
100%
Covered lines: 15
Uncovered lines: 0
Coverable lines: 15
Total lines: 231
Line coverage: 100%
Branch coverage
100%
Covered branches: 6
Total branches: 6
Branch coverage: 100%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
.ctor()100%11100%
UseAckAfterEnqueue(...)100%66100%

File(s)

/home/runner/work/AsyncResponse/AsyncResponse/src/Transports/AsyncResponse.Transports.SqlServer/SqlServerAsyncResponseTransportOptions.cs

#LineLine coverage
 1namespace AsyncResponse.Transports.SqlServer;
 2
 3/// <summary>Options for the Microsoft SQL Server AsyncResponse transport.</summary>
 4public sealed class SqlServerAsyncResponseTransportOptions
 5{
 6    /// <summary>The transport name reported to reply targets and the startup validator.</summary>
 7    public const string TransportName = "SqlServer";
 8
 9    /// <summary>
 10    /// SQL Server connection string used for every queue-table operation. Required. The database it
 11    /// targets must already exist; the transport creates only its schema, table, and indexes.
 12    /// </summary>
 13    public string? ConnectionString { get; set; }
 14
 15    /// <summary>Database schema that contains the transport table. Default: <c>dbo</c>.</summary>
 16    public string SchemaName { get; set; } = "dbo";
 17
 18    /// <summary>Table storing worker, response-ingress, and dead-letter queue rows.</summary>
 19    public string MessageTable { get; set; } = "asyncresponse_transport_messages";
 20
 21    /// <summary>Creates the schema, table, and indexes on first use. Disable when migrations own DDL.</summary>
 22    public bool AutoCreateSchema { get; set; } = true;
 23
 24    /// <summary>Logical queue name used by <see cref="SqlServerWorkerTransport"/>.</summary>
 25    public string WorkerQueue { get; set; } = "worker";
 26
 27    /// <summary>Worker queue handling options.</summary>
 28    public SqlServerSubscriberOptions WorkerSubscriber { get; } = new();
 29
 30    /// <summary>Logical queue name consumed by the hosted response-ingress subscriber.</summary>
 31    public string ResponseQueue { get; set; } = "response";
 32
 33    /// <summary>Response queue handling options.</summary>
 34    public SqlServerSubscriberOptions ResponseSubscriber { get; } = new();
 35
 36    /// <summary>Logical queue name that receives poison messages and already-ACKed background failures.</summary>
 37    public string DeadLetterQueue { get; set; } = "deadletter";
 38
 39    /// <summary>Enables dead-lettering when a message exhausts attempts or fails after early ACK.</summary>
 40    public bool DeadLetterEnabled { get; set; } = true;
 41
 42    /// <summary>
 43    /// How long dead-letter rows are retained before being pruned opportunistically. The dead-letter
 44    /// queue has no consumer by default, so without a retention its rows accumulate indefinitely.
 45    /// Leave <c>null</c> (the default) to keep them forever for manual inspection.
 46    /// </summary>
 47    public TimeSpan? DeadLetterRetention { get; set; }
 48
 49    /// <summary>How long a claimed row remains locked before another subscriber may retry it.</summary>
 50    public TimeSpan LockTimeout { get; set; } = TimeSpan.FromSeconds(30);
 51
 52    /// <summary>The logical reply target name used by <c>WithReplyTarget()</c>. Default: <c>default</c>.</summary>
 53    public string DefaultReplyTargetName { get; set; } = "default";
 54
 55    /// <summary>
 56    /// Named reply targets exposed to Core through <see cref="IAsyncResponseReplyTargetProvider"/>.
 57    /// When empty, <see cref="ResponseQueue"/> becomes the default target.
 58    /// </summary>
 59    public Dictionary<string, SqlServerReplyTargetOptions> ReplyTargets { get; } = new(StringComparer.Ordinal);
 60
 61    /// <summary>
 62    /// Header key stored in the row metadata for the AsyncResponse correlation id. Response messages
 63    /// may omit it when the id is present in the JSON body via <see cref="CorrelationIdJsonPaths"/>.
 64    /// </summary>
 65    public string CorrelationIdHeader { get; set; } = "AR-Correlation-Id";
 66
 67    /// <summary>
 68    /// JSON paths inspected when a response message does not carry the correlation id in
 69    /// <see cref="CorrelationIdHeader"/>. Paths are case-insensitive and support nested JSON strings.
 70    /// </summary>
 71    public string[] CorrelationIdJsonPaths { get; set; } =
 72    [
 73        "CorrelationId",
 74        "CustomParameters",
 75        "CustomParameters.CorrelationId",
 76        "PubSubParams.CustomParameters",
 77        "PubSubParams.CustomParameters.CorrelationId",
 78        "DagJsonParameters.CorrelationId"
 79    ];
 80
 81    /// <summary>Maximum attempts for SQL Server publish commands. Set to 1 to disable publish retries.</summary>
 82    public int PublishMaxAttempts { get; set; } = 3;
 83
 84    /// <summary>Initial delay before retrying a failed publish command.</summary>
 85    public TimeSpan PublishRetryBaseDelay { get; set; } = TimeSpan.FromMilliseconds(50);
 86
 87    /// <summary>Maximum delay between publish retry attempts.</summary>
 88    public TimeSpan PublishRetryMaxDelay { get; set; } = TimeSpan.FromSeconds(1);
 89
 90    /// <summary>Initial delay after a subscriber loop failure.</summary>
 91    public TimeSpan SubscriberRetryBaseDelay { get; set; } = TimeSpan.FromMilliseconds(100);
 92
 93    /// <summary>Maximum delay after repeated subscriber loop failures.</summary>
 94    public TimeSpan SubscriberRetryMaxDelay { get; set; } = TimeSpan.FromSeconds(5);
 95
 96    /// <summary>
 97    /// The hosting shutdown budget that must contain
 98    /// <see cref="SqlServerSubscriberOptions.BackgroundDrainTimeout"/> when a subscriber uses
 99    /// <see cref="SqlServerAckMode.AckAfterEnqueue"/>. Defaults to the Generic Host default of
 100    /// 30 seconds. Set to <c>null</c> only when this budget is validated externally.
 101    /// </summary>
 102    public TimeSpan? HostShutdownTimeout { get; set; } = TimeSpan.FromSeconds(30);
 103
 104    /// <summary>Adds or replaces a named SQL Server reply target.</summary>
 105    public SqlServerAsyncResponseTransportOptions AddReplyTarget(string name, string responseQueue)
 106    {
 107        ArgumentException.ThrowIfNullOrWhiteSpace(name);
 108        ArgumentException.ThrowIfNullOrWhiteSpace(responseQueue);
 109
 110        ReplyTargets[name] = new SqlServerReplyTargetOptions { ResponseQueue = responseQueue };
 111        return this;
 112    }
 113}
 114
 115/// <summary>Options for one named SQL Server async-response reply target.</summary>
 116public sealed class SqlServerReplyTargetOptions
 117{
 118    /// <summary>Queue remote systems should insert response payload rows into.</summary>
 119    public string? ResponseQueue { get; set; }
 120
 121    /// <summary>Additional values copied to the transport-neutral reply target.</summary>
 122    public Dictionary<string, string> Properties { get; } = new(StringComparer.Ordinal);
 123}
 124
 125/// <summary>Controls when a SQL Server transport row is acknowledged relative to handling.</summary>
 126public enum SqlServerAckMode
 127{
 128    /// <summary>
 129    /// Delete the row only after the AsyncResponse handler completes successfully. Handler failures
 130    /// reschedule the row until <see cref="SqlServerSubscriberOptions.MaxDeliveryAttempts"/> is reached.
 131    /// </summary>
 132    AckAfterHandlerCompletes = 0,
 133
 134    /// <summary>
 135    /// Delete the row immediately after it is accepted into a bounded in-process background queue.
 136    /// Handler failures are logged, reported, and dead-lettered when enabled because the row has
 137    /// already been acknowledged.
 138    /// </summary>
 139    AckAfterEnqueue = 1
 140}
 141
 142/// <summary>Describes a handler failure that happened after a SQL Server row was already acknowledged.</summary>
 143public sealed class SqlServerBackgroundFailureContext
 144{
 145    internal SqlServerBackgroundFailureContext(string queue, string subscriberRole, int attempt, string? correlationId, 
 146    {
 147        Queue = queue;
 148        SubscriberRole = subscriberRole;
 149        Attempt = attempt;
 150        CorrelationId = correlationId;
 151        Exception = exception;
 152    }
 153
 154    /// <summary>The logical queue the row came from.</summary>
 155    public string Queue { get; }
 156
 157    /// <summary>The logical subscriber role, such as <c>Worker</c> or <c>ResponseIngress</c>.</summary>
 158    public string SubscriberRole { get; }
 159
 160    /// <summary>The delivery attempt count for the row.</summary>
 161    public int Attempt { get; }
 162
 163    /// <summary>The AsyncResponse correlation id, when one was available.</summary>
 164    public string? CorrelationId { get; }
 165
 166    /// <summary>The exception thrown by the background handler.</summary>
 167    public Exception Exception { get; }
 168}
 169
 170/// <summary>Per-queue SQL Server subscriber behavior.</summary>
 171public sealed class SqlServerSubscriberOptions
 172{
 173    /// <summary>Controls when a row is acknowledged. Defaults to <see cref="SqlServerAckMode.AckAfterHandlerCompletes"/
 174    public SqlServerAckMode AckMode { get; set; } = SqlServerAckMode.AckAfterHandlerCompletes;
 175
 176    /// <summary>
 177    /// Maximum rows claimed per subscriber loop pass. In the default
 178    /// <see cref="SqlServerAckMode.AckAfterHandlerCompletes"/> mode the claimed rows are handled
 179    /// one at a time, so this bounds claim round-trips, not handler concurrency. Use
 180    /// <see cref="SqlServerAckMode.AckAfterEnqueue"/> (or run multiple subscriber instances) to
 181    /// process messages in parallel. Default: <c>16</c>.
 182    /// </summary>
 3183    public int BatchSize { get; set; } = 16;
 184
 185    /// <summary>
 186    /// Maximum delivery attempts before a failing row is deleted and written to the dead-letter
 187    /// queue. <c>0</c> means unlimited retries. Default: <c>5</c>.
 188    /// </summary>
 3189    public int MaxDeliveryAttempts { get; set; } = 5;
 190
 191    /// <summary>Delay before a failed row becomes available for redelivery. Default: <c>5s</c>.</summary>
 3192    public TimeSpan RedeliveryDelay { get; set; } = TimeSpan.FromSeconds(5);
 193
 194    /// <summary>
 195    /// Delay after an empty poll before checking again. SQL Server has no server-push notification
 196    /// (unlike PostgreSQL <c>LISTEN/NOTIFY</c>), so cross-process publishes are picked up within this
 197    /// delay; same-process publishes wake the subscriber immediately. Default: <c>250ms</c>.
 198    /// </summary>
 3199    public TimeSpan EmptyPollDelay { get; set; } = TimeSpan.FromMilliseconds(250);
 200
 201    /// <summary>Number of background workers used by <see cref="SqlServerAckMode.AckAfterEnqueue"/>.</summary>
 202    public int BackgroundWorkerCount { get; set; }
 203
 204    /// <summary>Maximum number of ACKed rows waiting in the background queue.</summary>
 205    public int BackgroundQueueCapacity { get; set; }
 206
 207    /// <summary>Maximum time to wait for queued/running background handlers while stopping.</summary>
 3208    public TimeSpan BackgroundDrainTimeout { get; set; } = TimeSpan.FromSeconds(20);
 209
 210    /// <summary>Optional callback invoked when a background handler fails after the row was already acknowledged.</summ
 211    public Func<SqlServerBackgroundFailureContext, ValueTask>? OnBackgroundFailure { get; set; }
 212
 213    /// <summary>Explicitly opts this subscriber into ACK-after-enqueue behavior.</summary>
 214    public SqlServerSubscriberOptions UseAckAfterEnqueue(
 215        int backgroundWorkerCount,
 216        int backgroundQueueCapacity,
 217        TimeSpan? backgroundDrainTimeout = null)
 218    {
 3219        ArgumentOutOfRangeException.ThrowIfNegativeOrZero(backgroundWorkerCount);
 3220        ArgumentOutOfRangeException.ThrowIfNegativeOrZero(backgroundQueueCapacity);
 3221        if (backgroundDrainTimeout is { } timeout && timeout <= TimeSpan.Zero)
 3222            throw new ArgumentOutOfRangeException(nameof(backgroundDrainTimeout), timeout, "Drain timeout must be positi
 223
 3224        AckMode = SqlServerAckMode.AckAfterEnqueue;
 3225        BackgroundWorkerCount = backgroundWorkerCount;
 3226        BackgroundQueueCapacity = backgroundQueueCapacity;
 3227        if (backgroundDrainTimeout is not null)
 3228            BackgroundDrainTimeout = backgroundDrainTimeout.Value;
 3229        return this;
 230    }
 231}
Generated by: ReportGenerator 5.5.10.0
08/03/2026 - 08:59:44
GitHub | reportgenerator.io

Methods/Properties

.ctor()
UseAckAfterEnqueue(int, int, System.Nullable<System.TimeSpan>)