Skip to main content

Google Pub/Sub Transport

Google Cloud Pub/Sub transport for scalable, GCP-native messaging with global availability.

Before You Start​

Installation​

dotnet add package Excalibur.Dispatch.Transport.GooglePubSub

Quick Start​

services.AddDispatch(dispatch =>
{
    dispatch.AddHandlersFromAssembly(typeof(Program).Assembly);
    dispatch.UseGooglePubSub(pubsub =>
    {
        pubsub.ProjectId("my-gcp-project")
              .TopicId("dispatch-events")
              .SubscriptionId("dispatch-events-sub");
    });
});

Standalone Registration​

services.AddDispatch(dispatch =>
{
    dispatch.AddHandlersFromAssembly(typeof(Program).Assembly);
});

services.AddGooglePubSubTransport("dispatch-events", pubsub =>
{
    pubsub.ProjectId("my-gcp-project")
          .TopicId("dispatch-events")
          .SubscriptionId("dispatch-events-sub")
          .MapTopic<OrderCreated>("orders-topic");
});

Google Pub/Sub registers a keyed IMessageBus named GooglePubSub or GooglePubSub:{TopicId} when TopicId is set:

var bus = serviceProvider.GetRequiredKeyedService<IMessageBus>("GooglePubSub:dispatch-events");

Configuration​

Pub/Sub Options​

Configure core transport settings with GooglePubSubOptions:

services.Configure<GooglePubSubOptions>("google-pubsub", options =>
{
    options.Connection.ProjectId = "my-gcp-project";
    options.Connection.TopicId = "dispatch-events";
    options.Connection.SubscriptionId = "dispatch-events-sub";

    options.Subscriber.MaxPullMessages = 100;
    options.Subscriber.AckDeadlineSeconds = 60;
    options.Subscriber.EnableAutoAckExtension = true;
    options.Subscriber.MaxConcurrentAcks = 10;
});

CloudEvents Configuration​

Via Transport Builder​

Configure CloudEvents settings directly on the transport builder:

services.AddGooglePubSubTransport("events", pubsub =>
{
    pubsub.ProjectId("my-gcp-project")
          .TopicId("dispatch-events")
          .SubscriptionId("dispatch-events-sub")
          .ConfigureOptions(options => options.Subscriber.EnableExactlyOnceDelivery = true)
          .ConfigureCloudEvents(ce =>
          {
              ce.UseOrderingKeys = true;
              ce.Transport.EnableCompression = true;
              ce.Transport.CompressionThreshold = 1024 * 1024; // 1MB
          });
});

Standalone CloudEvents Registration​

Use AddCloudEventsForPubSub for standalone CloudEvents configuration:

Trimming and Native AOT

The CloudEvents mapper bundled with this transport serializes the message payload with reflection-based JSON, so these registrations carry [RequiresUnreferencedCode] and [RequiresDynamicCode]. A host that trims or publishes ahead of time gets a warning at the call. To compose without the requirement, register your own ICloudEventEncoder<TOutbound> backed by a source-generated serializer.

services.AddCloudEventsForPubSub(options => options.UseOrderingKeys = true);

When UseOrderingKeys is enabled, CloudEvents use the partition key as the Pub/Sub ordering key to preserve ordering for related messages.

Dead Letter Topics​

Configure dead letter handling via the transport options:

services.AddGooglePubSubTransport("events", pubsub =>
{
    pubsub.ProjectId("my-gcp-project")
          .TopicId("dispatch-events")
          .EnableDeadLetter("dispatch-events-dlq");
});

Auto-Apply Dead Letter Policy​

By default the transport only references the dead letter topic; attaching the dead letter policy to the subscription is normally an infrastructure-as-code concern. You can opt in to having the transport automatically apply the policy to the subscription at startup (a GetSubscription + UpdateSubscription) so it is actually honored rather than configured but never attached:

services.AddGooglePubSubTransport("events", pubsub =>
{
    pubsub.ProjectId("my-gcp-project")
          .TopicId("dispatch-events")
          .SubscriptionId("dispatch-events-sub")
          .EnableDeadLetter("dispatch-events-dlq")
          .ConfigureOptions(options =>
          {
              options.AutoApplyDeadLetterPolicy = true;   // attach the policy at startup
              options.DeadLetterMaxDeliveryAttempts = 5;  // attempts before dead-lettering
          });
});

The policy is applied only when AutoApplyDeadLetterPolicy is enabled, a dead letter topic is configured, and a subscription id is set. DeadLetterMaxDeliveryAttempts controls how many delivery attempts occur before a message is dead-lettered.

OptionTypeDefaultDescription
AutoApplyDeadLetterPolicyboolfalseAttaches the configured dead letter policy to the subscription at startup.
DeadLetterMaxDeliveryAttemptsint5Delivery attempts before a message is dead-lettered (applies when auto-apply is enabled).

Publish Failures and Retryability​

When a publish fails, the returned SendError.IsRetryable tells you whether trying again can succeed. It is classified from the gRPC status code the Pub/Sub service returned, so you do not have to interpret status codes yourself:

Status codeIsRetryableWhy
UNAVAILABLEtrueTransient. Retry with exponential backoff.
DEADLINE_EXCEEDEDtrueThe deadline passed; the outcome is unknown.
INTERNALtrueTransient server-side condition.
RESOURCE_EXHAUSTEDtrueA traffic spike; backoff is the correct response.
CANCELLEDtrueTransient, unless you cancelled — see below.
INVALID_ARGUMENTfalseThe message is malformed; it will fail again unchanged.
PERMISSION_DENIED / UNAUTHENTICATEDfalseFix the credentials, not the retry policy.
FAILED_PRECONDITION / ALREADY_EXISTSfalseRetrying cannot change the condition.
NOT_FOUNDfalseTreated as permanent for a configured topic.
anything elsefalseUnrecognised codes are conservatively non-retryable.
var result = await sender.SendAsync(message, cancellationToken);

if (!result.IsSuccess && result.Error?.IsRetryable == true)
{
    // Safe to retry with backoff.
}
Retryable does not mean the publish did not happen

CANCELLED and DEADLINE_EXCEEDED are ambiguous outcomes: the service may already have accepted the message before the call ended. Retrying one of these can therefore deliver the message twice. That is consistent with the at-least-once delivery this transport provides — your consumers must be idempotent regardless.

Cancellation you requested is never reported as retryable. If the cancellation token you passed is already cancelled, whether to try again is a decision you have made, and the token would still be cancelled on a retry — so the failure is reported as non-retryable rather than inviting a loop.

Health Checks​

When using transport adapters, register aggregate health checks:

services.AddHealthChecks()
    .AddTransportHealthChecks();

Observability​

services.AddOpenTelemetry()
    .WithTracing(tracing =>
    {
        tracing.AddSource("Excalibur.Dispatch");
        tracing.AddGoogleCloudTraceExporter();
    })
    .WithMetrics(metrics =>
    {
        metrics.AddDispatchMetrics();
    });

Configure telemetry options via GooglePubSubOptions:

services.Configure<GooglePubSubOptions>("google-pubsub", options =>
{
    options.Telemetry.EnableOpenTelemetry = true;
    options.Telemetry.ExportToCloudMonitoring = true;
    options.Telemetry.TracingSamplingRatio = 0.1; // 10% sampling
    options.Telemetry.EnableTracePropagation = true;
});

Production Checklist​

  • Use Workload Identity or managed credentials
  • Configure Subscriber.EnableExactlyOnceDelivery for critical streams
  • Enable ordering keys for strict ordering requirements
  • Configure dead letter topics for failed messages
  • Enable OpenTelemetry and Cloud Monitoring

Next Steps​

See Also​