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

Information
Class: AsyncResponse.Transports.SqlServer.SqlServerTransportStore
Assembly: AsyncResponse.Transports.SqlServer
File(s): /home/runner/work/AsyncResponse/AsyncResponse/src/Transports/AsyncResponse.Transports.SqlServer/SqlServerTransportStore.cs
Line coverage
94%
Covered lines: 235
Uncovered lines: 15
Coverable lines: 250
Total lines: 462
Line coverage: 94%
Branch coverage
97%
Covered branches: 41
Total branches: 42
Branch coverage: 97.6%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
.ctor(...)100%11100%
EnsureCreatedAsync()100%66100%
PublishAsync()100%11100%
TryClaimAsync()100%11100%
ClaimBatchAsync()100%66100%
InsertAsync()100%8895.24%
AckAsync()100%11100%
RenewLeaseAsync()100%210%
NakAsync()100%22100%
DeadLetterAsync()75%8880.95%
PruneDeadLettersIfDueAsync()100%44100%
ShouldPruneDeadLetters()100%22100%
OpenConnectionAsync()100%22100%
DeserializeHeaders(...)100%22100%
Sanitize(...)100%11100%
AddMilliseconds(...)100%11100%
SchemaLockResource(...)100%11100%
Quote(...)100%11100%
IndexName(...)100%22100%
.cctor()100%11100%

File(s)

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

#LineLine coverage
 1using Microsoft.Data.SqlClient;
 2using Microsoft.Extensions.Logging;
 3using Microsoft.Extensions.Options;
 4using System.Runtime.CompilerServices;
 5using System.Text.Json;
 6
 7namespace AsyncResponse.Transports.SqlServer;
 8
 9internal enum SqlServerSubscriberRole
 10{
 11    Worker,
 12    ResponseIngress
 13}
 14
 15/// <summary>A claimed SQL Server transport row, decoupled from SqlClient types for dispatch tests.</summary>
 16/// <remarks>
 17/// <c>RenewAsync</c> extends the claim's lease (<c>locked_until</c>) by the original lock timeout,
 18/// fenced on the claim's <c>lock_id</c>; it returns <c>false</c> when the fence no longer matches
 19/// (the lease lapsed and another subscriber re-claimed the row).
 20/// </remarks>
 21internal sealed record SqlServerTransportDelivery(
 22    Guid Id,
 23    string Queue,
 24    string Payload,
 25    IReadOnlyDictionary<string, string> Headers,
 26    int Attempt,
 27    Func<ValueTask> AckAsync,
 28    Func<TimeSpan, ValueTask> NakAsync,
 29    Func<Exception, bool, CancellationToken, ValueTask<bool>> DeadLetterAsync,
 30    Func<ValueTask<bool>> RenewAsync);
 31
 32/// <summary>Small SQL adapter for the SQL Server transport queue table.</summary>
 33internal sealed class SqlServerTransportStore
 34{
 35    // SQL Server duplicate-key error numbers: 2627 = PRIMARY KEY/UNIQUE constraint violation,
 36    // 2601 = unique index violation. A concurrent retry of the same idempotent publish can lose the
 37    // WHERE NOT EXISTS race; the duplicate is the outcome the caller asked for, not a failure.
 38    private const int PrimaryKeyViolation = 2627;
 39    private const int UniqueIndexViolation = 2601;
 40
 41    // Interpolated into DDL: literal braces cannot appear directly inside the interpolated raw string.
 42    private const string EmptyJsonObject = "{}";
 43
 44    private readonly string _connectionString;
 45    private readonly SqlServerAsyncResponseTransportOptions _options;
 46    private readonly ILogger<SqlServerTransportStore>? _logger;
 347    private readonly SemaphoreSlim _ensureGate = new(1, 1);
 48    private bool _created;
 49    private long _lastDeadLetterPruneTicks;
 50
 351    public SqlServerTransportStore(
 352        IOptions<SqlServerAsyncResponseTransportOptions> options,
 353        ILogger<SqlServerTransportStore>? logger = null)
 54    {
 355        _options = options.Value;
 356        _logger = logger;
 357        SqlServerTransportOptionsValidator.ValidateCommon(_options);
 358        _connectionString = _options.ConnectionString!;
 359        Schema = Quote(_options.SchemaName);
 360        MessageTable = $"{Schema}.{Quote(_options.MessageTable)}";
 361    }
 62
 63    public string Schema { get; }
 64    public string MessageTable { get; }
 65
 66    /// <summary>
 67    /// Raised after a row is inserted (with the logical queue name) or released for retry
 68    /// (<c>null</c>). Same-process subscribers use it to wake immediately instead of waiting out
 69    /// their empty-poll delay; SQL Server has no LISTEN/NOTIFY, so cross-process wakes rely on polling.
 70    /// </summary>
 71    public event Action<string?>? MessagePublished;
 72
 73    public async Task EnsureCreatedAsync(CancellationToken cancellationToken = default)
 74    {
 375        if (_created || !_options.AutoCreateSchema)
 376            return;
 77
 378        await _ensureGate.WaitAsync(cancellationToken).ConfigureAwait(false);
 79        try
 80        {
 381            if (_created)
 182                return;
 83
 384            await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 185            await using var transaction = (SqlTransaction)await connection.BeginTransactionAsync(cancellationToken).Conf
 86
 87            // Serialize schema creation across processes. The IF-NOT-EXISTS guards are not atomic
 88            // against a concurrent create of the same object: two instances starting together both
 89            // pass the existence check and collide on the catalog (error 2714/2627). A
 90            // transaction-scoped application lock (keyed by schema, shared with the channel store)
 91            // lets one instance build the schema while the rest wait and then find it already present.
 192            await using (var lockCommand = connection.CreateCommand())
 93            {
 194                lockCommand.Transaction = transaction;
 195                lockCommand.CommandText =
 196                    """
 197                    DECLARE @lock_result int;
 198                    EXEC @lock_result = sp_getapplock
 199                        @Resource = @lock_resource,
 1100                        @LockMode = 'Exclusive',
 1101                        @LockOwner = 'Transaction',
 1102                        @LockTimeout = 60000;
 1103                    IF @lock_result < 0
 1104                        THROW 51000, N'Failed to acquire the AsyncResponse DDL application lock.', 1;
 1105                    """;
 1106                lockCommand.Parameters.AddWithValue("@lock_resource", SchemaLockResource(_options.SchemaName));
 1107                await lockCommand.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 108            }
 109
 1110            await using var command = connection.CreateCommand();
 1111            command.Transaction = transaction;
 1112            command.CommandText =
 1113                $"""
 1114                IF SCHEMA_ID(N'{_options.SchemaName}') IS NULL
 1115                    EXEC(N'CREATE SCHEMA {Schema}');
 1116
 1117                IF OBJECT_ID(N'{MessageTable}', N'U') IS NULL
 1118                CREATE TABLE {MessageTable} (
 1119                    id uniqueidentifier NOT NULL PRIMARY KEY NONCLUSTERED,
 1120                    queue nvarchar(200) NOT NULL,
 1121                    payload_json nvarchar(max) NOT NULL,
 1122                    headers_json nvarchar(max) NOT NULL DEFAULT N'{EmptyJsonObject}',
 1123                    created_at datetime2 NOT NULL DEFAULT SYSUTCDATETIME(),
 1124                    available_at datetime2 NOT NULL DEFAULT SYSUTCDATETIME(),
 1125                    locked_until datetime2 NULL,
 1126                    lock_id uniqueidentifier NULL,
 1127                    attempts int NOT NULL DEFAULT 0,
 1128                    dead_letter_reason nvarchar(max) NULL
 1129                );
 1130
 1131                IF NOT EXISTS (SELECT 1 FROM sys.indexes WHERE name = N'{IndexName(_options.MessageTable, "claim")}' AND
 1132                    CREATE INDEX {Quote(IndexName(_options.MessageTable, "claim"))}
 1133                        ON {MessageTable} (queue, available_at, locked_until, created_at);
 1134                IF NOT EXISTS (SELECT 1 FROM sys.indexes WHERE name = N'{IndexName(_options.MessageTable, "created")}' A
 1135                    CREATE INDEX {Quote(IndexName(_options.MessageTable, "created"))}
 1136                        ON {MessageTable} (created_at);
 1137                """;
 1138            await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 1139            await transaction.CommitAsync(cancellationToken).ConfigureAwait(false);
 1140            _created = true;
 1141        }
 142        finally
 143        {
 3144            _ensureGate.Release();
 145        }
 3146    }
 147
 148    /// <summary>
 149    /// Publishes a queue row. The caller supplies the id so a retried publish is idempotent
 150    /// (insert-if-absent) rather than inserting a duplicate job.
 151    /// </summary>
 152    public async Task PublishAsync(
 153        Guid id,
 154        string queue,
 155        string payload,
 156        IReadOnlyDictionary<string, string>? headers,
 157        CancellationToken cancellationToken)
 158    {
 3159        await InsertAsync(id, queue, payload, headers, deadLetterReason: null, notify: true, cancellationToken).Configur
 1160        await PruneDeadLettersIfDueAsync(cancellationToken).ConfigureAwait(false);
 1161    }
 162
 163    public async Task<SqlServerTransportDelivery?> TryClaimAsync(string queue, TimeSpan lockTimeout, CancellationToken c
 164    {
 1165        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 1166        var lockId = Guid.NewGuid();
 167
 1168        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1169        await using var command = connection.CreateCommand();
 170        // READPAST skips rows other subscribers hold UPDLOCK on — SQL Server's equivalent of
 171        // PostgreSQL's FOR UPDATE SKIP LOCKED — so competing consumers never block on each other.
 1172        command.CommandText =
 1173            $"""
 1174            WITH next AS (
 1175                SELECT TOP (1) id, queue, payload_json, headers_json, attempts, locked_until, lock_id
 1176                FROM {MessageTable} WITH (UPDLOCK, ROWLOCK, READPAST)
 1177                WHERE queue = @queue
 1178                  AND available_at <= SYSUTCDATETIME()
 1179                  AND (locked_until IS NULL OR locked_until <= SYSUTCDATETIME())
 1180                ORDER BY created_at
 1181            )
 1182            UPDATE next
 1183            SET attempts = attempts + 1,
 1184                locked_until = {AddMilliseconds("@lock_timeout_ms")},
 1185                lock_id = @lock_id
 1186            OUTPUT inserted.id, inserted.queue, inserted.payload_json, inserted.headers_json, inserted.attempts;
 1187            """;
 1188        command.Parameters.AddWithValue("@queue", queue);
 1189        command.Parameters.AddWithValue("@lock_timeout_ms", (long)lockTimeout.TotalMilliseconds);
 1190        command.Parameters.AddWithValue("@lock_id", lockId);
 191
 1192        await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 1193        if (!await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 1194            return null;
 195
 1196        var id = reader.GetGuid(0);
 1197        var payload = reader.GetString(2);
 1198        var headerJson = reader.GetString(3);
 1199        var attempt = reader.GetInt32(4);
 1200        var headers = DeserializeHeaders(headerJson);
 201
 1202        return new SqlServerTransportDelivery(
 1203            id,
 1204            reader.GetString(1),
 1205            payload,
 1206            headers,
 1207            attempt,
 1208            () => AckAsync(id, lockId),
 1209            delay => NakAsync(id, lockId, delay),
 1210            (exception, deleteOriginal, token) => DeadLetterAsync(id, lockId, queue, payload, headers, exception, delete
 1211            () => RenewLeaseAsync(id, lockId, lockTimeout));
 1212    }
 213
 214    public async IAsyncEnumerable<SqlServerTransportDelivery> ClaimBatchAsync(
 215        string queue,
 216        int batchSize,
 217        TimeSpan lockTimeout,
 218        [EnumeratorCancellation] CancellationToken cancellationToken)
 219    {
 1220        for (var i = 0; i < batchSize; i++)
 221        {
 1222            var delivery = await TryClaimAsync(queue, lockTimeout, cancellationToken).ConfigureAwait(false);
 1223            if (delivery is null)
 1224                yield break;
 1225            yield return delivery;
 226        }
 1227    }
 228
 229    private async Task InsertAsync(
 230        Guid id,
 231        string queue,
 232        string payload,
 233        IReadOnlyDictionary<string, string>? headers,
 234        string? deadLetterReason,
 235        bool notify,
 236        CancellationToken cancellationToken)
 237    {
 3238        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 3239        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1240        await using var command = connection.CreateCommand();
 241        // Insert-if-absent keeps a retried publish idempotent. The UPDLOCK/HOLDLOCK hints make the
 242        // existence check and the insert atomic; a concurrent same-id insert that still slips through
 243        // surfaces as a duplicate-key error, which is treated as success below.
 1244        command.CommandText =
 1245            $"""
 1246            INSERT INTO {MessageTable} (id, queue, payload_json, headers_json, dead_letter_reason)
 1247            SELECT @id, @queue, @payload_json, @headers_json, @dead_letter_reason
 1248            WHERE NOT EXISTS (SELECT 1 FROM {MessageTable} WITH (UPDLOCK, HOLDLOCK) WHERE id = @id);
 1249            """;
 1250        command.Parameters.AddWithValue("@id", id);
 1251        command.Parameters.AddWithValue("@queue", queue);
 1252        command.Parameters.AddWithValue("@payload_json", payload);
 1253        command.Parameters.AddWithValue("@headers_json", AsyncResponseJson.Serialize(headers ?? EmptyHeaders));
 1254        command.Parameters.AddWithValue("@dead_letter_reason", (object?)deadLetterReason ?? DBNull.Value);
 255
 256        try
 257        {
 1258            await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 1259        }
 0260        catch (SqlException ex) when (ex.Number is PrimaryKeyViolation or UniqueIndexViolation)
 261        {
 1262        }
 263
 1264        if (notify)
 1265            MessagePublished?.Invoke(queue);
 1266    }
 267
 268    private async ValueTask AckAsync(Guid id, Guid lockId)
 269    {
 1270        await using var connection = await OpenConnectionAsync(CancellationToken.None).ConfigureAwait(false);
 1271        await using var command = connection.CreateCommand();
 1272        command.CommandText = $"DELETE FROM {MessageTable} WHERE id = @id AND lock_id = @lock_id;";
 1273        command.Parameters.AddWithValue("@id", id);
 1274        command.Parameters.AddWithValue("@lock_id", lockId);
 1275        await command.ExecuteNonQueryAsync(CancellationToken.None).ConfigureAwait(false);
 1276    }
 277
 278    private async ValueTask<bool> RenewLeaseAsync(Guid id, Guid lockId, TimeSpan lockTimeout)
 279    {
 0280        await using var connection = await OpenConnectionAsync(CancellationToken.None).ConfigureAwait(false);
 0281        await using var command = connection.CreateCommand();
 0282        command.CommandText =
 0283            $"""
 0284            UPDATE {MessageTable}
 0285            SET locked_until = {AddMilliseconds("@lock_timeout_ms")}
 0286            WHERE id = @id AND lock_id = @lock_id;
 0287            """;
 0288        command.Parameters.AddWithValue("@id", id);
 0289        command.Parameters.AddWithValue("@lock_id", lockId);
 0290        command.Parameters.AddWithValue("@lock_timeout_ms", (long)lockTimeout.TotalMilliseconds);
 0291        return await command.ExecuteNonQueryAsync(CancellationToken.None).ConfigureAwait(false) > 0;
 0292    }
 293
 294    private async ValueTask NakAsync(Guid id, Guid lockId, TimeSpan delay)
 295    {
 1296        await using var connection = await OpenConnectionAsync(CancellationToken.None).ConfigureAwait(false);
 1297        await using var command = connection.CreateCommand();
 1298        command.CommandText =
 1299            $"""
 1300            UPDATE {MessageTable}
 1301            SET available_at = {AddMilliseconds("@delay_ms")},
 1302                locked_until = NULL,
 1303                lock_id = NULL
 1304            WHERE id = @id AND lock_id = @lock_id;
 1305            """;
 1306        command.Parameters.AddWithValue("@id", id);
 1307        command.Parameters.AddWithValue("@lock_id", lockId);
 1308        command.Parameters.AddWithValue("@delay_ms", (long)delay.TotalMilliseconds);
 1309        await command.ExecuteNonQueryAsync(CancellationToken.None).ConfigureAwait(false);
 1310        MessagePublished?.Invoke(null);
 1311    }
 312
 313    private async ValueTask<bool> DeadLetterAsync(
 314        Guid id,
 315        Guid lockId,
 316        string sourceQueue,
 317        string payload,
 318        IReadOnlyDictionary<string, string> headers,
 319        Exception exception,
 320        bool deleteOriginal,
 321        CancellationToken cancellationToken)
 322    {
 3323        if (!_options.DeadLetterEnabled)
 324        {
 1325            if (deleteOriginal)
 1326                await AckAsync(id, lockId).ConfigureAwait(false);
 1327            return true;
 328        }
 329
 3330        var deadHeaders = new Dictionary<string, string>(headers, StringComparer.OrdinalIgnoreCase)
 3331        {
 3332            ["AR-DeadLetter-Reason"] = Sanitize(exception.Message),
 3333            ["AR-DeadLetter-Source-Queue"] = sourceQueue
 3334        };
 335
 336        try
 337        {
 3338            if (!deleteOriginal)
 339            {
 3340                await InsertAsync(Guid.NewGuid(), _options.DeadLetterQueue, payload, deadHeaders, exception.Message, not
 0341                return true;
 342            }
 343
 344            // The DLQ insert and the original-row delete must commit atomically: split across two
 345            // connections, a crash between them leaves the original row to be redelivered and
 346            // dead-lettered again, duplicating the DLQ entry.
 3347            await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 3348            await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1349            await using var transaction = (SqlTransaction)await connection.BeginTransactionAsync(cancellationToken).Conf
 1350            await using var command = connection.CreateCommand();
 1351            command.Transaction = transaction;
 1352            command.CommandText =
 1353                $"""
 1354                INSERT INTO {MessageTable} (id, queue, payload_json, headers_json, dead_letter_reason)
 1355                VALUES (@id, @queue, @payload_json, @headers_json, @dead_letter_reason);
 1356                DELETE FROM {MessageTable} WHERE id = @source_id AND lock_id = @lock_id;
 1357                """;
 1358            command.Parameters.AddWithValue("@id", Guid.NewGuid());
 1359            command.Parameters.AddWithValue("@queue", _options.DeadLetterQueue);
 1360            command.Parameters.AddWithValue("@payload_json", payload);
 1361            command.Parameters.AddWithValue("@headers_json", AsyncResponseJson.Serialize(deadHeaders));
 1362            command.Parameters.AddWithValue("@dead_letter_reason", exception.Message);
 1363            command.Parameters.AddWithValue("@source_id", id);
 1364            command.Parameters.AddWithValue("@lock_id", lockId);
 1365            await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 1366            await transaction.CommitAsync(cancellationToken).ConfigureAwait(false);
 1367            return true;
 1368        }
 2369        catch (Exception ex)
 370        {
 371            // Callers decide the redelivery consequence from the false return; log the cause here so
 372            // a failing dead-letter write is never silent.
 2373            _logger?.LogError(
 2374                ex,
 2375                "Failed to write SQL Server dead-letter row for message {MessageId} from queue {SourceQueue}.",
 2376                id,
 2377                sourceQueue);
 2378            return false;
 379        }
 3380    }
 381
 382    /// <summary>
 383    /// Opportunistically deletes dead-letter rows older than the configured retention. No-op unless
 384    /// <see cref="SqlServerAsyncResponseTransportOptions.DeadLetterRetention"/> is set, and throttled
 385    /// so the DELETE runs at most once per minute regardless of publish rate.
 386    /// </summary>
 387    private async Task PruneDeadLettersIfDueAsync(CancellationToken cancellationToken)
 388    {
 1389        if (_options.DeadLetterRetention is not { } retention || !ShouldPruneDeadLetters())
 1390            return;
 391
 1392        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1393        await using var command = connection.CreateCommand();
 1394        command.CommandText = $"DELETE FROM {MessageTable} WHERE queue = @queue AND created_at < {AddMilliseconds("@nega
 1395        command.Parameters.AddWithValue("@queue", _options.DeadLetterQueue);
 1396        command.Parameters.AddWithValue("@negative_retention_ms", -(long)retention.TotalMilliseconds);
 1397        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 1398    }
 399
 400    private bool ShouldPruneDeadLetters()
 401    {
 1402        var now = DateTime.UtcNow.Ticks;
 1403        var last = Interlocked.Read(ref _lastDeadLetterPruneTicks);
 1404        return now - last >= DeadLetterPruneThrottle.Ticks
 1405            && Interlocked.CompareExchange(ref _lastDeadLetterPruneTicks, now, last) == last;
 406    }
 407
 408    private async Task<SqlConnection> OpenConnectionAsync(CancellationToken cancellationToken)
 409    {
 3410        var connection = new SqlConnection(_connectionString);
 411        try
 412        {
 3413            await connection.OpenAsync(cancellationToken).ConfigureAwait(false);
 1414            return connection;
 415        }
 3416        catch
 417        {
 3418            await connection.DisposeAsync().ConfigureAwait(false);
 3419            throw;
 420        }
 1421    }
 422
 423    private static IReadOnlyDictionary<string, string> DeserializeHeaders(string json)
 424    {
 1425        var parsed = AsyncResponseJson.Deserialize<Dictionary<string, string>>(json);
 1426        return parsed is null
 1427            ? EmptyHeaders
 1428            : new Dictionary<string, string>(parsed, StringComparer.OrdinalIgnoreCase);
 429    }
 430
 3431    private static string Sanitize(string value) => value.Replace('\r', ' ').Replace('\n', ' ');
 432
 433    /// <summary>
 434    /// SQL expression adding a millisecond bigint parameter to the database clock. DATEADD only takes
 435    /// int arguments, so the value is split into whole seconds and a sub-second remainder — intervals
 436    /// (lock timeouts, redelivery delays, retentions) stay on the database clock, immune to app-side
 437    /// clock skew, without overflowing on long spans.
 438    /// </summary>
 439    internal static string AddMilliseconds(string parameterName)
 3440        => $"DATEADD(SECOND, CAST({parameterName} / 1000 AS int), DATEADD(MILLISECOND, CAST({parameterName} % 1000 AS in
 441
 442    /// <summary>
 443    /// Stable application-lock resource for serializing schema creation. Must be byte-for-byte
 444    /// identical to the channel store's resource so that, for a shared schema, the channel and
 445    /// transport take the same lock and never race each other on CREATE SCHEMA.
 446    /// </summary>
 447    internal static string SchemaLockResource(string schemaName)
 3448        => $"asyncresponse:ddl:{schemaName}";
 449
 3450    private static string Quote(string identifier) => "[" + identifier + "]";
 451
 452    private static string IndexName(string table, string suffix)
 453    {
 1454        var name = $"{table}_{suffix}_idx";
 1455        return name.Length <= 128 ? name : name[..128];
 456    }
 457
 3458    private static readonly TimeSpan DeadLetterPruneThrottle = TimeSpan.FromMinutes(1);
 459
 3460    private static readonly IReadOnlyDictionary<string, string> EmptyHeaders =
 3461        new Dictionary<string, string>(0, StringComparer.OrdinalIgnoreCase);
 462}
Generated by: ReportGenerator 5.5.10.0
08/03/2026 - 08:59:44
GitHub | reportgenerator.io

Methods/Properties

.ctor(Microsoft.Extensions.Options.IOptions<AsyncResponse.Transports.SqlServer.SqlServerAsyncResponseTransportOptions>, Microsoft.Extensions.Logging.ILogger<AsyncResponse.Transports.SqlServer.SqlServerTransportStore>)
EnsureCreatedAsync()
PublishAsync()
TryClaimAsync()
ClaimBatchAsync()
InsertAsync()
AckAsync()
RenewLeaseAsync()
NakAsync()
DeadLetterAsync()
PruneDeadLettersIfDueAsync()
ShouldPruneDeadLetters()
OpenConnectionAsync()
DeserializeHeaders(string)
Sanitize(string)
AddMilliseconds(string)
SchemaLockResource(string)
Quote(string)
IndexName(string, string)
.cctor()