Docs / Backends / GCP Pub/Sub

GCP Pub/Sub

Topics and subscriptions with DeadLetterPolicy, publish batching and ordering keys

import "go.digitalxero.dev/mq-gcp/v2"

Index

• func NewMessageBuilder() mqtypes.MessageBuilder
• func NewPubSubConsumer(config *viper.Viper) (types.Consumer, error)
• func NewPubSubProducer(config *viper.Viper) (types.Producer, error)
• func NewPubSubTransport() mqtypes.Transport

funcNewMessageBuilder

func NewMessageBuilder() mqtypes.MessageBuilder

NewMessageBuilder creates a new Pub/Sub message builder. The builder provides a fluent interface for constructing Pub/Sub messages with all supported properties and attributes.

Pub/Sub message limitations:

• Message body: max 10MB
• Attributes: max 100 attributes
• Attribute key: max 256 bytes
• Attribute value: max 1024 bytes
• Ordering key: max 1024 bytes

Example:

msg := gcp.NewMessageBuilder().
    WithBody([]byte(`{"event": "user.signup", "userId": "123"}`)).
    WithContentType("application/json").
    WithCorrelationId("signup-123").
    WithRoutingKey("user-123"). // Used as ordering key
    WithHeaders(map[string]interface{}{
        "source": "web-app",
        "version": "1.2.0",
        "region": "us-central1",
    }).
    Build()

funcNewPubSubConsumer

func NewPubSubConsumer(config *viper.Viper) (types.Consumer, error)

NewPubSubConsumer creates a new GCP Pub/Sub consumer with the provided configuration. This function is typically called by mqutils.NewConsumer when it detects a Pub/Sub URL.

The canonical cross-backend keys “destination” (-> topic_id), “max_retries” (-> max_delivery_attempts), and “consumer_group” (-> subscription_id) are accepted alongside the native keys; native keys win when both are set.

Configuration options:

• url: Pub/Sub URL (e.g., “pubsub://project-id/topic-id?subscription=sub-id”; required unless project_id, topic_id, and subscription_id are provided)
• project_id: GCP project ID (extracted from URL if not provided)
• topic_id: Pub/Sub topic ID (extracted from URL if not provided)
• subscription_id: Pub/Sub subscription ID
• create_topic_if_not_exists: Create the topic when missing (default: false)
• create_subscription_if_not_exists: Create the subscription when missing (default: false)
• handler: Name of registered handler function (default: “pubsubLogger”, or “pubsubBatchLogger” when batch processing is enabled)
• max_retries: Retry budget before the runner drops a message (default: max_delivery_attempts, normally 5)
• max_concurrent_handlers: Maximum in-flight handlers or batches (default: effective message_channel_buffer; range: 1–10000)
• max_extension_seconds: Max message deadline extension (default: 300)
• min_retry_delay_seconds / max_retry_delay_seconds: Subscription RetryPolicy backoff bounds (defaults: 10 / 300)
• ack_deadline_seconds: Message acknowledgment deadline (default: 60)
• retention_duration_seconds: Message retention (default: 86400)
• exactly_once_delivery: Enable exactly-once delivery (default: false)
• enable_message_ordering: Enable message ordering by key (default: false)
• dead_letter_topic: Topic for undeliverable messages (optional)
• max_delivery_attempts: Max delivery attempts before dead lettering (default: 5)
• batch_size: Number of messages per batch (default: 5)
• batch_timeout: Batch collection window as a duration (default: “100ms”)
• enable_batch_processing: Enable batch message processing (default: false)

============================ IMPORTANT — IAM ============================ Dead lettering additionally requires granting the project’s Pub/Sub service account roles/pubsub.publisher on the dead letter topic and roles/pubsub.subscriber on the subscription. This library CANNOT grant those roles; see the BindQueue documentation on the transport. =========================================================================

Existing subscription policies are preserved unless their settings are explicitly configured. Explicit ack deadline, retention, exactly-once, and retry-policy changes require subscription update permission; failures are returned.

Returns an error if configuration validation fails or transport creation fails.

funcNewPubSubProducer

func NewPubSubProducer(config *viper.Viper) (types.Producer, error)

NewPubSubProducer creates a new GCP Pub/Sub producer with the provided configuration. This function is typically called by mqutils.NewProducer when it detects a Pub/Sub URL.

The canonical cross-backend key “destination” (-> topic_id) is accepted alongside the native keys; native keys win when both are set.

The batch settings map onto the Pub/Sub topic PublishSettings (DelayThreshold, CountThreshold, ByteThreshold).

Call Close on the returned producer to release transport resources after publishing has finished. Close is idempotent.

Returns an error if configuration validation fails or transport creation fails. The producer must be started with Start() before publishing messages.

funcNewPubSubTransport

func NewPubSubTransport() mqtypes.Transport

NewPubSubTransport creates a new GCP Pub/Sub transport implementation. The transport provides low-level Pub/Sub operations including topic/subscription management, message publishing/receiving, and health monitoring.

The transport manages:

• GCP Pub/Sub client with automatic credential resolution
• Topic and subscription lifecycle management
• Message ordering and exactly-once delivery settings
• Dead letter topic configuration
• Flow control and concurrency settings
• Connection health monitoring

Pub/Sub uses gRPC for communication with automatic reconnection and retry logic built into the client library.

Generated by gomarkdoc