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

Information
Class: AsyncResponse.Transports.GooglePubSub.GooglePubSubWorkerSubscriber
Assembly: AsyncResponse.Transports.GooglePubSub
File(s): /home/runner/work/AsyncResponse/AsyncResponse/src/Transports/AsyncResponse.Transports.GooglePubSub/GooglePubSubSubscriberServices.cs
Line coverage
100%
Covered lines: 10
Uncovered lines: 0
Coverable lines: 10
Total lines: 232
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%
.ctor(...)100%11100%
get_SubscriptionId()100%11100%
get_SubscriberOptions()100%11100%
get_SubscriberRole()100%11100%
HandleMessageAsync(...)100%11100%

File(s)

/home/runner/work/AsyncResponse/AsyncResponse/src/Transports/AsyncResponse.Transports.GooglePubSub/GooglePubSubSubscriberServices.cs

#LineLine coverage
 1using Google.Api.Gax;
 2using Google.Cloud.PubSub.V1;
 3using Microsoft.Extensions.Hosting;
 4using Microsoft.Extensions.Logging;
 5using Microsoft.Extensions.Options;
 6
 7namespace AsyncResponse.Transports.GooglePubSub;
 8
 9internal abstract class GooglePubSubSubscriberService : BackgroundService
 10{
 11    private readonly Func<SubscriptionName, GooglePubSubSubscriberOptions, Task<IGooglePubSubSubscriberClient>> _subscri
 12
 13    /// <summary>Runs the GooglePubSubSubscriberService operation.</summary>
 14    protected GooglePubSubSubscriberService(
 15        IOptions<GooglePubSubAsyncResponseOptions> options,
 16        ILogger logger)
 17        : this(options, logger, CreateSubscriberAsync)
 18    {
 19    }
 20
 21    /// <summary>Runs the GooglePubSubSubscriberService operation.</summary>
 22    protected GooglePubSubSubscriberService(
 23        IOptions<GooglePubSubAsyncResponseOptions> options,
 24        ILogger logger,
 25        Func<SubscriptionName, GooglePubSubSubscriberOptions, Task<IGooglePubSubSubscriberClient>> subscriberFactory)
 26    {
 27        Options = options.Value;
 28        Logger = logger;
 29        _subscriberFactory = subscriberFactory;
 30    }
 31
 32    protected GooglePubSubAsyncResponseOptions Options { get; }
 33    protected ILogger Logger { get; }
 34
 35    protected abstract string SubscriptionId { get; }
 36    protected abstract GooglePubSubSubscriberOptions SubscriberOptions { get; }
 37    protected abstract GooglePubSubSubscriberRole SubscriberRole { get; }
 38    /// <summary>Handles the delivered message.</summary>
 39    protected abstract Task HandleMessageAsync(PubsubMessage message, CancellationToken cancellationToken);
 40
 41    [System.Diagnostics.CodeAnalysis.ExcludeFromCodeCoverage]
 42    private static async Task<IGooglePubSubSubscriberClient> CreateSubscriberAsync(
 43        SubscriptionName subscriptionName,
 44        GooglePubSubSubscriberOptions subscriberOptions)
 45    {
 46        // EmulatorOrProduction honors PUBSUB_EMULATOR_HOST when present (local dev / tests) and uses
 47        // real Google Cloud otherwise — no behavior change in production.
 48        var builder = new SubscriberClientBuilder
 49        {
 50            SubscriptionName = subscriptionName,
 51            EmulatorDetection = EmulatorDetection.EmulatorOrProduction
 52        };
 53
 54        // In early-ACK mode, bound the streaming pull to the background queue capacity so the client
 55        // never holds more un-ACKed messages than the dispatcher can accept. Combined with the
 56        // dispatcher's write-side backpressure this keeps queue-full NACKs (which burn a configured
 57        // DeadLetterPolicy's delivery attempts) out of steady-state operation.
 58        if (subscriberOptions.AckMode is GooglePubSubAckMode.AckAfterEnqueue)
 59        {
 60            builder.Settings = new SubscriberClient.Settings
 61            {
 62                FlowControlSettings = new Google.Api.Gax.FlowControlSettings(
 63                    maxOutstandingElementCount: subscriberOptions.BackgroundQueueCapacity,
 64                    maxOutstandingByteCount: null)
 65            };
 66        }
 67
 68        var subscriber = await builder.BuildAsync().ConfigureAwait(false);
 69        return new GooglePubSubSubscriberClientAdapter(subscriber);
 70    }
 71
 72    /// <summary>Runs this background operation until cancellation is requested.</summary>
 73    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
 74    {
 75        var projectId = GooglePubSubOptionsValidator.Required(Options.ProjectId, nameof(Options.ProjectId));
 76        var subscriptionId = SubscriptionId;
 77        GooglePubSubMessageDispatcher.ValidateOptions(Options, SubscriberOptions, SubscriberRole);
 78        var subscriptionName = SubscriptionName.FromProjectSubscription(projectId, subscriptionId);
 79
 80        // The transport intentionally has no MaxDeliveryAttempts and no library-managed dead-letter
 81        // queue for Pub/Sub: capping redelivery is delegated to the subscription's native
 82        // DeadLetterPolicy. The client cannot cheaply probe whether one is configured, so tell the
 83        // operator unconditionally instead of failing silently forever on a poison message.
 84        Logger.LogWarning(
 85            "Pub/Sub redelivery is unbounded for subscription {Subscription} ({Role}): the transport enforces no deliver
 86            + "Configure a DeadLetterPolicy on the subscription to cap redeliveries of failing messages.",
 87            subscriptionName.ToString(),
 88            SubscriberRole);
 89
 90        var failures = 0;
 91        while (!stoppingToken.IsCancellationRequested)
 92        {
 93            try
 94            {
 95                await RunSubscriberAsync(subscriptionName, subscriptionId, stoppingToken).ConfigureAwait(false);
 96                return;
 97            }
 98            catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
 99            {
 100                return;
 101            }
 102            catch (Exception ex) when (!stoppingToken.IsCancellationRequested)
 103            {
 104                failures++;
 105                var retryDelay = AsyncResponseRetry.Backoff(
 106                    failures,
 107                    Options.SubscriberRetryBaseDelay,
 108                    Options.SubscriberRetryMaxDelay);
 109                Logger.LogWarning(
 110                    ex,
 111                    "Pub/Sub subscriber failed for subscription {Subscription} ({Role}); retrying in {RetryDelay}.",
 112                    subscriptionName.ToString(),
 113                    SubscriberRole,
 114                    retryDelay);
 115                await Task.Delay(retryDelay, stoppingToken).ConfigureAwait(false);
 116            }
 117        }
 118    }
 119
 120    private async Task RunSubscriberAsync(
 121        SubscriptionName subscriptionName,
 122        string subscriptionId,
 123        CancellationToken stoppingToken)
 124    {
 125        var subscriber = await _subscriberFactory(subscriptionName, SubscriberOptions).ConfigureAwait(false);
 126        await using var dispatcher = GooglePubSubMessageDispatcher.Create(
 127            HandleMessageAsync,
 128            Options,
 129            SubscriberOptions,
 130            Logger,
 131            subscriptionId,
 132            SubscriberRole);
 133
 134        Logger.LogInformation(
 135            "Pub/Sub subscriber started. Subscription: {Subscription}. Role: {Role}. AckMode: {AckMode}.",
 136            subscriptionName.ToString(),
 137            SubscriberRole,
 138            SubscriberOptions.AckMode);
 139
 140        var runTask = subscriber.StartAsync(dispatcher.HandleAsync);
 141
 142        try
 143        {
 144            await runTask.WaitAsync(stoppingToken).ConfigureAwait(false);
 145        }
 146        catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
 147        {
 148            await subscriber.StopAsync(
 149                new SubscriberClient.ShutdownOptions
 150                {
 151                    Timeout = Options.ShutdownTimeout
 152                },
 153                CancellationToken.None).ConfigureAwait(false);
 154            await runTask.ConfigureAwait(false);
 155        }
 156    }
 157
 158}
 159
 160internal sealed class GooglePubSubWorkerSubscriber : GooglePubSubSubscriberService
 161{
 162    private readonly IAsyncResponseIngress _ingress;
 163
 164    /// <summary>Runs the GooglePubSubWorkerSubscriber operation.</summary>
 165    public GooglePubSubWorkerSubscriber(
 166        IOptions<GooglePubSubAsyncResponseOptions> options,
 167        IAsyncResponseIngress ingress,
 168        ILogger<GooglePubSubWorkerSubscriber> logger)
 3169        : base(options, logger)
 170    {
 3171        _ingress = ingress;
 3172    }
 173
 174    internal GooglePubSubWorkerSubscriber(
 175        IOptions<GooglePubSubAsyncResponseOptions> options,
 176        IAsyncResponseIngress ingress,
 177        ILogger<GooglePubSubWorkerSubscriber> logger,
 178        Func<SubscriptionName, GooglePubSubSubscriberOptions, Task<IGooglePubSubSubscriberClient>> subscriberFactory)
 2179        : base(options, logger, subscriberFactory)
 180    {
 3181        _ingress = ingress;
 3182    }
 183
 184    protected override string SubscriptionId
 3185        => GooglePubSubOptionsValidator.Required(Options.WorkerSubscriptionId, nameof(Options.WorkerSubscriptionId));
 186
 3187    protected override GooglePubSubSubscriberOptions SubscriberOptions => Options.WorkerSubscriber;
 3188    protected override GooglePubSubSubscriberRole SubscriberRole => GooglePubSubSubscriberRole.Worker;
 189
 190    /// <summary>Handles the delivered message.</summary>
 191    protected override Task HandleMessageAsync(PubsubMessage message, CancellationToken cancellationToken)
 3192        => _ingress.HandleWorkerMessageAsync(message.Data.ToStringUtf8());
 193}
 194
 195internal sealed class GooglePubSubResponseIngressSubscriber : GooglePubSubSubscriberService
 196{
 197    private readonly IAsyncResponseIngress _ingress;
 198
 199    /// <summary>Runs the GooglePubSubResponseIngressSubscriber operation.</summary>
 200    public GooglePubSubResponseIngressSubscriber(
 201        IOptions<GooglePubSubAsyncResponseOptions> options,
 202        IAsyncResponseIngress ingress,
 203        ILogger<GooglePubSubResponseIngressSubscriber> logger)
 204        : base(options, logger)
 205    {
 206        _ingress = ingress;
 207    }
 208
 209    internal GooglePubSubResponseIngressSubscriber(
 210        IOptions<GooglePubSubAsyncResponseOptions> options,
 211        IAsyncResponseIngress ingress,
 212        ILogger<GooglePubSubResponseIngressSubscriber> logger,
 213        Func<SubscriptionName, GooglePubSubSubscriberOptions, Task<IGooglePubSubSubscriberClient>> subscriberFactory)
 214        : base(options, logger, subscriberFactory)
 215    {
 216        _ingress = ingress;
 217    }
 218
 219    protected override string SubscriptionId
 220        => GooglePubSubOptionsValidator.Required(Options.ResponseSubscriptionId, nameof(Options.ResponseSubscriptionId))
 221
 222    protected override GooglePubSubSubscriberOptions SubscriberOptions => Options.ResponseSubscriber;
 223    protected override GooglePubSubSubscriberRole SubscriberRole => GooglePubSubSubscriberRole.ResponseIngress;
 224
 225    /// <summary>Handles the delivered message.</summary>
 226    protected override Task HandleMessageAsync(PubsubMessage message, CancellationToken cancellationToken)
 227    {
 228        var messageJson = message.Data.ToStringUtf8();
 229        var correlationId = GooglePubSubCorrelationIdExtractor.Extract(message, messageJson, Options);
 230        return _ingress.HandleResponseMessageAsync(messageJson, correlationId);
 231    }
 232}
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.GooglePubSub.GooglePubSubAsyncResponseOptions>, AsyncResponse.IAsyncResponseIngress, Microsoft.Extensions.Logging.ILogger<AsyncResponse.Transports.GooglePubSub.GooglePubSubWorkerSubscriber>)
.ctor(Microsoft.Extensions.Options.IOptions<AsyncResponse.Transports.GooglePubSub.GooglePubSubAsyncResponseOptions>, AsyncResponse.IAsyncResponseIngress, Microsoft.Extensions.Logging.ILogger<AsyncResponse.Transports.GooglePubSub.GooglePubSubWorkerSubscriber>, System.Func<Google.Cloud.PubSub.V1.SubscriptionName, AsyncResponse.Transports.GooglePubSub.GooglePubSubSubscriberOptions, System.Threading.Tasks.Task<AsyncResponse.Transports.GooglePubSub.IGooglePubSubSubscriberClient>>)
get_SubscriptionId()
get_SubscriberOptions()
get_SubscriberRole()
HandleMessageAsync(Google.Cloud.PubSub.V1.PubsubMessage, System.Threading.CancellationToken)