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

Information
Class: AsyncResponse.Transports.GooglePubSub.GooglePubSubWorkerTransport
Assembly: AsyncResponse.Transports.GooglePubSub
File(s): /_/src/Transports/AsyncResponse.Transports.GooglePubSub/GooglePubSubWorkerTransport.cs
Line coverage
100%
Covered lines: 66
Uncovered lines: 0
Coverable lines: 66
Total lines: 183
Line coverage: 100%
Branch coverage
100%
Covered branches: 20
Total branches: 20
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%
.ctor(...)100%11100%
get_MaxInFlightDuration()100%22100%
GetPublisherAsync()100%44100%
PublishAsync()100%1010100%
DisposeAsync()100%44100%

File(s)

/_/src/Transports/AsyncResponse.Transports.GooglePubSub/GooglePubSubWorkerTransport.cs

#LineLine coverage
 1using Google.Api.Gax;
 2using Google.Cloud.PubSub.V1;
 3using Google.Protobuf;
 4using Microsoft.Extensions.Options;
 5using System.Diagnostics;
 6using System.Text.Json;
 7
 8namespace AsyncResponse.Transports.GooglePubSub;
 9
 10/// <summary>
 11/// Publishes <see cref="WorkerJobEnvelope"/> messages to a Google Pub/Sub topic.
 12/// </summary>
 13/// <remarks>
 14/// The publisher client is created lazily and re-created on demand: a transient build failure when the
 15/// first job is published does not permanently break the transport (a faulted build attempt is not cached).
 16/// </remarks>
 17public sealed class GooglePubSubWorkerTransport : IWorkerTransportInFlightLimit, IAsyncDisposable
 18{
 19    private readonly GooglePubSubAsyncResponseOptions _options;
 20    private readonly Func<CancellationToken, Task<IGooglePubSubPublisherClient>> _publisherFactory;
 24121    private readonly SemaphoreSlim _publisherGate = new(1, 1);
 22    private IGooglePubSubPublisherClient? _publisher;
 23    private int _disposeGate;
 24    private bool _disposed;
 25
 26    /// <summary>Runs the GooglePubSubWorkerTransport operation.</summary>
 27    public GooglePubSubWorkerTransport(IOptions<GooglePubSubAsyncResponseOptions> options)
 40728        : this(options, cancellationToken => CreatePublisherAsync(options.Value, cancellationToken))
 29    {
 20530    }
 31
 24132    internal GooglePubSubWorkerTransport(
 24133        IOptions<GooglePubSubAsyncResponseOptions> options,
 24134        Func<CancellationToken, Task<IGooglePubSubPublisherClient>> publisherFactory)
 35    {
 24136        _options = options.Value;
 24137        _ = GooglePubSubOptionsValidator.Required(_options.ProjectId, nameof(_options.ProjectId));
 23738        _ = GooglePubSubOptionsValidator.Required(_options.WorkerTopicId, nameof(_options.WorkerTopicId));
 23339        GooglePubSubOptionsValidator.ValidateTimeouts(_options);
 40        // Advertised below as the in-flight ceiling, whose contract is "positive": check it here
 41        // as well as in the subscriber, which may start later than the first consumer of the value.
 23342        GooglePubSubOptionsValidator.ValidateMaxTotalAckExtension(
 23343            _options.WorkerSubscriber,
 23344            $"{nameof(GooglePubSubAsyncResponseOptions)}.{nameof(GooglePubSubAsyncResponseOptions.WorkerSubscriber)}");
 23145        _publisherFactory = publisherFactory;
 23146    }
 47
 48    /// <summary>
 49    /// The worker subscriber's <see cref="GooglePubSubSubscriberOptions.MaxTotalAckExtension"/>: past
 50    /// it the Pub/Sub client stops extending the job's ack deadline and Pub/Sub redelivers the same
 51    /// job while its first handler is still running. <c>null</c> when the worker subscriber uses
 52    /// <see cref="GooglePubSubAckMode.AckAfterEnqueue"/> — the delivery is settled at enqueue, so no
 53    /// handler is ever held against the ceiling.
 54    /// <para>
 55    /// The subscriber and this publisher read the same options instance, so the value is the one the
 56    /// consuming side applies. Pub/Sub redelivers up to one ack deadline (60 seconds) <em>after</em>
 57    /// the extension lapses; advertising the extension itself keeps the engine on the safe side.
 58    /// </para>
 59    /// </summary>
 60    public TimeSpan? MaxInFlightDuration
 24661        => _options.WorkerSubscriber.AckMode is GooglePubSubAckMode.AckAfterEnqueue
 24662            ? null
 24663            : _options.WorkerSubscriber.MaxTotalAckExtension;
 64
 65    private async Task<IGooglePubSubPublisherClient> GetPublisherAsync(CancellationToken cancellationToken)
 66    {
 43767        var publisher = Volatile.Read(ref _publisher);
 43768        if (publisher is not null)
 21569            return publisher;
 70
 22271        await _publisherGate.WaitAsync(cancellationToken).ConfigureAwait(false);
 72        try
 73        {
 21874            ObjectDisposedException.ThrowIf(_disposed, this);
 21475            if (_publisher is not null)
 276                return _publisher;
 77
 78            // Assign only after the await succeeds, so a faulted build attempt is not cached and the next
 79            // publish retries instead of awaiting a permanently faulted task. The token reaches the
 80            // build itself (sibling parity): a stalled credential/metadata lookup or gRPC handshake
 81            // under this gate otherwise ignored the caller's — and the host's stopping — token.
 21282            var created = await _publisherFactory(cancellationToken).ConfigureAwait(false);
 20883            _publisher = created;
 20884            return created;
 85        }
 86        finally
 87        {
 21888            _publisherGate.Release();
 89        }
 42590    }
 91
 92    [System.Diagnostics.CodeAnalysis.ExcludeFromCodeCoverage]
 93    private static async Task<IGooglePubSubPublisherClient> CreatePublisherAsync(
 94        GooglePubSubAsyncResponseOptions options,
 95        CancellationToken cancellationToken)
 96    {
 97        var projectId = GooglePubSubOptionsValidator.Required(options.ProjectId, nameof(options.ProjectId));
 98        var topicId = GooglePubSubOptionsValidator.Required(options.WorkerTopicId, nameof(options.WorkerTopicId));
 99        var topicName = TopicName.FromProjectTopic(projectId, topicId);
 100        // EmulatorOrProduction honors PUBSUB_EMULATOR_HOST when present (local dev / tests) and uses
 101        // real Google Cloud otherwise — no behavior change in production.
 102        var publisher = await new PublisherClientBuilder
 103        {
 104            TopicName = topicName,
 105            EmulatorDetection = EmulatorDetection.EmulatorOrProduction
 106        }.BuildAsync(cancellationToken).ConfigureAwait(false);
 107        return new GooglePubSubPublisherClientAdapter(publisher);
 108    }
 109
 110    /// <summary>Publishes the supplied message.</summary>
 111    public async Task PublishAsync(WorkerJobEnvelope job, CancellationToken cancellationToken = default)
 112    {
 439113        ArgumentNullException.ThrowIfNull(job);
 114
 437115        using var activity = AsyncResponseDiagnostics.StartActivity(
 437116            "asyncresponse.worker.publish",
 437117            ActivityKind.Producer,
 437118            job.CorrelationId);
 437119        activity?.SetTag("asyncresponse.transport", "google_pubsub");
 437120        activity?.SetTag("messaging.system", "gcp_pubsub");
 437121        activity?.SetTag("messaging.destination.name", _options.WorkerTopicId);
 437122        AsyncResponseDiagnostics.SetReplyTarget(activity, job.ReplyTarget);
 437123        AsyncResponseDiagnostics.SetWorker(activity, job.Call);
 124
 125        try
 126        {
 437127            var message = new PubsubMessage
 437128            {
 437129                Data = ByteString.CopyFromUtf8(AsyncResponseJson.Serialize(job))
 437130            };
 131
 437132            if (!string.IsNullOrWhiteSpace(job.CorrelationId))
 124133                message.Attributes[_options.CorrelationIdAttribute] = job.CorrelationId;
 134
 437135            var publisher = await GetPublisherAsync(cancellationToken).ConfigureAwait(false);
 425136            var messageId = await publisher.PublishAsync(message, cancellationToken).ConfigureAwait(false);
 423137            activity?.SetTag("messaging.message.id", messageId);
 423138        }
 14139        catch (Exception ex)
 140        {
 14141            AsyncResponseDiagnostics.SetError(activity, ex);
 14142            throw;
 143        }
 423144    }
 145
 146    /// <summary>Releases resources held by this instance.</summary>
 147    public async ValueTask DisposeAsync()
 148    {
 414149        if (Interlocked.Exchange(ref _disposeGate, 1) != 0)
 197150            return;
 151
 217152        await _publisherGate.WaitAsync().ConfigureAwait(false);
 153        try
 154        {
 217155            _disposed = true;
 217156            if (_publisher is not null)
 157            {
 158                // Best effort, like the RabbitMQ/Azure Service Bus worker transports' closes: the
 159                // SDK CANCELS the returned task when the timeout expires before the backlog
 160                // flushes, and this is a container-created singleton — a throw here escapes
 161                // ServiceProvider.DisposeAsync and aborts the disposal of every service after it.
 162                try
 163                {
 200164                    await _publisher.ShutdownAsync(_options.ShutdownTimeout).ConfigureAwait(false);
 198165                }
 2166                catch
 167                {
 168                    // Best effort.
 2169                }
 170            }
 217171        }
 172        finally
 173        {
 174            // Release, never Dispose: SemaphoreSlim.Dispose does not complete pending WaitAsync
 175            // waiters, so disposing here would strand publishers parked on the gate forever (and
 176            // the first woken waiter's finally would throw trying to Release a disposed
 177            // semaphore, never handing the permit on). Released, each parked waiter wakes in
 178            // turn and observes _disposed; the gate holds no unmanaged resources, so leaving it
 179            // undisposed leaks nothing.
 217180            _publisherGate.Release();
 181        }
 414182    }
 183}
Generated by: ReportGenerator 5.5.11.0
09/22/2026 - 09:41:04
GitHub | reportgenerator.io

Methods/Properties

.ctor(Microsoft.Extensions.Options.IOptions`1<AsyncResponse.Transports.GooglePubSub.GooglePubSubAsyncResponseOptions>,System.Func`2<System.Threading.CancellationToken,System.Threading.Tasks.Task`1<AsyncResponse.Transports.GooglePubSub.IGooglePubSubPublisherClient>>)
.ctor(Microsoft.Extensions.Options.IOptions`1<AsyncResponse.Transports.GooglePubSub.GooglePubSubAsyncResponseOptions>)
get_MaxInFlightDuration()
GetPublisherAsync()
PublishAsync()
DisposeAsync()