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

Information
Class: AsyncResponse.Transports.PostgreSQL.PostgreSqlReplyTargetOptions
Assembly: AsyncResponse.Transports.PostgreSQL
File(s): /home/runner/work/AsyncResponse/AsyncResponse/src/Transports/AsyncResponse.Transports.PostgreSQL/PostgreSqlAsyncResponseTransportOptions.cs
Line coverage
100%
Covered lines: 1
Uncovered lines: 0
Coverable lines: 1
Total lines: 231
Line coverage: 100%
Branch coverage
N/A
Covered branches: 0
Total branches: 0
Branch coverage: N/A
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

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

File(s)

/home/runner/work/AsyncResponse/AsyncResponse/src/Transports/AsyncResponse.Transports.PostgreSQL/PostgreSqlAsyncResponseTransportOptions.cs

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

Methods/Properties

.ctor()