pscompat

package
v1.10.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 15, 2026 License: Apache-2.0 Imports: 28 Imported by: 10

Documentation

Overview

Package pscompat contains clients for publishing and subscribing using the Pub/Sub Lite service.

This package is designed to compatible with the Cloud Pub/Sub library: https://pkg.go.dev/cloud.google.com/go/pubsub. If interfaces are defined by the client application, PublisherClient and SubscriberClient can be used as substitutions for pubsub.Topic.Publish() and pubsub.Subscription.Receive(), respectively, from the pubsub package. See the following examples: https://pkg.go.dev/cloud.google.com/go/pubsublite/pscompat#example-NewPublisherClient-Interface and https://pkg.go.dev/cloud.google.com/go/pubsublite/pscompat#example-NewSubscriberClient-Interface.

The Cloud Pub/Sub and Pub/Sub Lite services have some differences:

  • Pub/Sub Lite does not support NACK for messages. By default, this will terminate the SubscriberClient. A custom function can be provided for ReceiveSettings.NackHandler to handle NACKed messages.
  • Pub/Sub Lite has no concept of ACK deadlines. Subscribers must ACK or NACK every message received and can take as much time as they need to process the message.
  • Pub/Sub Lite PublisherClients and SubscriberClients can fail permanently when an unretryable error occurs.
  • Publishers and subscribers will be throttled if Pub/Sub Lite publish or subscribe throughput limits are exceeded. Thus publishing can be more sensitive to buffer overflow than Cloud Pub/Sub.
  • Pub/Sub Lite utilizes bidirectional gRPC streams extensively to maximize publish and subscribe throughput.

More information about Pub/Sub Lite is available at https://cloud.google.com/pubsub/lite.

Information about choosing between Cloud Pub/Sub vs Pub/Sub Lite is available at https://cloud.google.com/pubsub/docs/choosing-pubsub-or-lite.

Complete sample programs can be found at https://github.com/GoogleCloudPlatform/golang-samples/tree/master/pubsublite.

See https://pkg.go.dev/cloud.google.com/go for authentication, timeouts, connection pooling and similar aspects of this package.

Index

Examples

Constants

View Source 
const (
	// MaxPublishRequestCount is the maximum number of messages that can be
	// batched in a single publish request.
	MaxPublishRequestCount = wire.MaxPublishRequestCount

	// MaxPublishRequestBytes is the maximum allowed serialized size of a single
	// publish request (containing a batch of messages) in bytes.
	MaxPublishRequestBytes = wire.MaxPublishRequestBytes
)
View Source 
const EventTimeAttributeKey = "x-goog-pubsublite-event-time-timestamp-proto"

EventTimeAttributeKey is the key of the attribute whose value is an encoded Timestamp proto. The value will be used to set the event time property of a Pub/Sub Lite message.

Variables

View Source 
var (
	// ErrOverflow is set for a PublishResult when publish buffers overflow. This
	// can occur when backends are unavailable or the actual publish throughput
	// of clients exceeds the allocated publish throughput for the Pub/Sub Lite
	// topic. Use errors.Is for comparing errors.
	ErrOverflow = wire.ErrOverflow

	// ErrOversizedMessage is set for a PublishResult when a published message
	// exceeds MaxPublishRequestBytes. Publishing this message will never succeed.
	// Use errors.Is for comparing errors.
	ErrOversizedMessage = wire.ErrOversizedMessage

	// ErrPublisherStopped is set for a PublishResult when a message cannot be
	// published because the publisher client has stopped or is in the process of
	// stopping. It may be stopping due to a fatal error. PublisherClient.Error()
	// returns the error that caused the publisher client to terminate (if any).
	// Use errors.Is for comparing errors.
	ErrPublisherStopped = wire.ErrServiceStopped

	// ErrBackendUnavailable indicates that the backend service has been
	// unavailable for a period of time. The timeout can be configured using
	// PublishSettings.Timeout or ReceiveSettings.Timeout. Use errors.Is for
	// comparing errors.
	ErrBackendUnavailable = wire.ErrBackendUnavailable
)
View Source 
var DefaultPublishSettings = PublishSettings{
	DelayThreshold:    10 * time.Millisecond,
	CountThreshold:    100,
	ByteThreshold:     1e6,
	Timeout:           7 * 24 * time.Hour,
	BufferedByteLimit: 1e9,
	EnableIdempotence: true,
}

DefaultPublishSettings holds the default values for PublishSettings.

View Source 
var DefaultReceiveSettings = ReceiveSettings{
	MaxOutstandingMessages: 1000,
	MaxOutstandingBytes:    1e9,
	Timeout:                7 * 24 * time.Hour,
}

DefaultReceiveSettings holds the default values for ReceiveSettings.

View Source 
var ErrUnsupported = errors.New("gmk: operation unsupported on Managed Kafka backend")

ErrUnsupported indicates an operation that has no meaningful Kafka analogue. Callers should either omit the operation or handle this error explicitly.

Functions

func BuildGMKBootstrapServer added in v1.9.0 

func BuildGMKBootstrapServer(projectID, region, clusterID string) string

BuildGMKBootstrapServer constructs the bootstrap server URL for a Google Managed Kafka cluster.

func DecodeEventTimeAttribute added in v1.8.0 

func DecodeEventTimeAttribute(value string) (*tspb.Timestamp, error)

DecodeEventTimeAttribute decodes a timestamp that was encoded with EncodeEventTimeAttribute.

Example 
package main

import (
	"context"
	"fmt"

	"cloud.google.com/go/pubsub"
	"cloud.google.com/go/pubsublite/pscompat"
)

func main() {
	ctx := context.Background()
	const subscription = "projects/my-project/locations/region-or-zone/subscriptions/my-subscription"
	subscriber, err := pscompat.NewSubscriberClient(ctx, subscription)
	if err != nil {
		// TODO: Handle error.
	}
	cctx, cancel := context.WithCancel(ctx)
	err = subscriber.Receive(cctx, func(ctx context.Context, m *pubsub.Message) {
		m.Ack()
		if v, ok := m.Attributes[pscompat.EventTimeAttributeKey]; ok {
			eventTime, err := pscompat.DecodeEventTimeAttribute(v)
			if err != nil {
				// TODO: Handle error.
			}
			fmt.Printf("Received message with event time: %v\n", eventTime)
		}
	})
	if err != nil {
		// TODO: Handle error.
	}

	// Call cancel from the receiver callback or another goroutine to stop
	// receiving.
	cancel()
}

func EncodeEventTimeAttribute added in v1.8.0 

func EncodeEventTimeAttribute(eventTime *tspb.Timestamp) (string, error)

EncodeEventTimeAttribute encodes a timestamp in a way that it will be interpreted as an event time if published on a message with an attribute named EventTimeAttributeKey.

Example 
package main

import (
	"context"

	"cloud.google.com/go/pubsub"
	"cloud.google.com/go/pubsublite/pscompat"
	"google.golang.org/protobuf/types/known/timestamppb"
)

func main() {
	ctx := context.Background()
	const topic = "projects/my-project/locations/region-or-zone/topics/my-topic"
	publisher, err := pscompat.NewPublisherClient(ctx, topic)
	if err != nil {
		// TODO: Handle error.
	}
	defer publisher.Stop()

	v, err := pscompat.EncodeEventTimeAttribute(&timestamppb.Timestamp{
		Seconds: 1672531200,
		Nanos:   500000000,
	})
	if err != nil {
		// TODO: Handle error.
	}

	r := publisher.Publish(ctx, &pubsub.Message{
		Data: []byte("hello world"),
		Attributes: map[string]string{
			pscompat.EventTimeAttributeKey: v,
		},
	})
	_, err = r.Get(ctx)
	if err != nil {
		// TODO: Handle error.
	}
}

func MigrateOffsets added in v1.9.0 

func MigrateOffsets(ctx context.Context, cfg *MigrationConfig) (map[int32]MigrationResult, error)

MigrateOffsets translates the PSL committed cursor for each partition to a Kafka offset via publish-time timestamps and commits the resolved offsets to the Kafka consumer group in a single batch. The implementation mirrors Java's OffsetMigrationHelper and runs in five steps:

  1. List PSL committed offsets for the subscription.
  2. For each partition, look up the message at that PSL offset and read its publish timestamp.
  3. Resolve the corresponding Kafka offset by timestamp lookup on the Kafka topic.
  4. Commit the resolved offsets to the Kafka consumer group in one batch.
  5. Optionally re-read the committed offsets to confirm.

func NewGMKSaramaConfig added in v1.9.0 

func NewGMKSaramaConfig(ctx context.Context) (*sarama.Config, error)

NewGMKSaramaConfig creates a Sarama configuration pre-configured for Google Managed Kafka with SASL_SSL/OAUTHBEARER authentication using GCP default credentials.

Types

type BacklogInfo added in v1.9.0 

type BacklogInfo struct {
	// MessageCount is the number of messages between the committed offset and
	// the latest (head) offset.
	MessageCount int64
	// ByteCount is the estimated byte size of that backlog.
	ByteCount int64
}

BacklogInfo reports how far behind head a consumer group is on a given partition.

type KafkaAdminClient added in v1.9.0 

type KafkaAdminClient struct {
	// contains filtered or unexported fields
}

KafkaAdminClient is a Managed Kafka-backed admin client providing PSL-style topic and subscription management. Subscriptions map to Kafka consumer groups; topics map to Kafka topics.

Methods that have no Kafka analogue (updateSubscription, reservation ops, seekSubscription) either return ErrUnsupported or log a WARNING and succeed as a no-op, matching the documented degraded-behavior contract.

func NewKafkaAdminClient added in v1.9.0 

func NewKafkaAdminClient(ctx context.Context, cfg *KafkaAdminClientConfig) (*KafkaAdminClient, error)

NewKafkaAdminClient creates an admin client connected to Managed Kafka.

func (*KafkaAdminClient) Close added in v1.9.0 

func (c *KafkaAdminClient) Close() error

Close releases the underlying connection.

func (*KafkaAdminClient) CreateReservation added in v1.9.0 

func (c *KafkaAdminClient) CreateReservation(_ context.Context, name string) error

CreateReservation is a no-op on Managed Kafka.

func (*KafkaAdminClient) CreateSubscription added in v1.9.0 

func (c *KafkaAdminClient) CreateSubscription(_ context.Context, groupID string) error

CreateSubscription is a no-op on Managed Kafka. Consumer groups are created implicitly when a consumer joins them for the first time.

func (*KafkaAdminClient) CreateTopic added in v1.9.0 

func (c *KafkaAdminClient) CreateTopic(_ context.Context, topic string, numPartitions int32, replicationFactor int16) error

CreateTopic creates a Kafka topic with the given partition count and replication factor.

func (*KafkaAdminClient) DeleteReservation added in v1.9.0 

func (c *KafkaAdminClient) DeleteReservation(_ context.Context, name string) error

DeleteReservation is a no-op on Managed Kafka.

func (*KafkaAdminClient) DeleteSubscription added in v1.9.0 

func (c *KafkaAdminClient) DeleteSubscription(_ context.Context, groupID string) error

DeleteSubscription deletes a Kafka consumer group.

func (*KafkaAdminClient) DeleteTopic added in v1.9.0 

func (c *KafkaAdminClient) DeleteTopic(_ context.Context, topic string) error

DeleteTopic deletes a Kafka topic.

func (*KafkaAdminClient) GetReservation added in v1.9.0 

func (c *KafkaAdminClient) GetReservation(_ context.Context, name string) error

GetReservation is a no-op on Managed Kafka.

func (*KafkaAdminClient) GetSubscription added in v1.9.0 

func (c *KafkaAdminClient) GetSubscription(_ context.Context, groupID, topic string) (map[int32]int64, error)

GetSubscription returns the offsets currently committed for a consumer group on a given topic. Matches PSL's GetSubscription semantics insofar as Kafka offers an analogue.

func (*KafkaAdminClient) GetTopic added in v1.9.0 

func (c *KafkaAdminClient) GetTopic(_ context.Context, topic string) (*sarama.TopicMetadata, error)

GetTopic returns metadata for a single topic.

func (*KafkaAdminClient) IncreasePartitions added in v1.9.0 

func (c *KafkaAdminClient) IncreasePartitions(_ context.Context, topic string, newCount int32) error

IncreasePartitions grows a topic's partition count. Kafka only supports monotonically increasing partition counts; shrinking is not possible.

Note: increasing partitions on a live topic does not redistribute existing data. Producers using the default partitioner will start routing keys to the new partition layout immediately, which can break per-key ordering guarantees for keys whose hash now maps to a different partition. Existing records remain on their original partitions and are not rebalanced. Most callers will want to coordinate this with their producers/consumers (and any downstream stateful processors) before calling.

func (*KafkaAdminClient) ListReservations added in v1.9.0 

func (c *KafkaAdminClient) ListReservations(_ context.Context) ([]string, error)

ListReservations is a no-op on Managed Kafka.

func (*KafkaAdminClient) ListSubscriptions added in v1.9.0 

func (c *KafkaAdminClient) ListSubscriptions(_ context.Context) ([]string, error)

ListSubscriptions returns all Kafka consumer group names in the cluster.

func (*KafkaAdminClient) ListTopics added in v1.9.0 

func (c *KafkaAdminClient) ListTopics(_ context.Context) ([]string, error)

ListTopics returns all non-internal topic names in the cluster. Internal topics (those starting with "__") are filtered out, matching the Java KafkaAdminClient behavior. Optionally accepts an "expect" topic that must be present; retries briefly to tolerate post-CreateTopic metadata propagation.

func (*KafkaAdminClient) ListTopicsAwait added in v1.9.0 

func (c *KafkaAdminClient) ListTopicsAwait(ctx context.Context, expect string) ([]string, error)

ListTopicsAwait is like ListTopics but retries until the named topic shows up in the listing or attempts are exhausted. Useful for workflows that create a topic and then immediately list it.

func (*KafkaAdminClient) SeekSubscription added in v1.9.0 

func (c *KafkaAdminClient) SeekSubscription(_ context.Context, _ string) error

SeekSubscription returns ErrUnsupported. Admin-level seeks are not exposed by the Kafka admin surface; use KafkaCursorClient for offset management.

func (*KafkaAdminClient) TopicPartitionCount added in v1.9.0 

func (c *KafkaAdminClient) TopicPartitionCount(_ context.Context, topic string) (int, error)

TopicPartitionCount returns the number of partitions for the given topic. Retries briefly to tolerate post-CreateTopic metadata propagation.

func (*KafkaAdminClient) UpdateReservation added in v1.9.0 

func (c *KafkaAdminClient) UpdateReservation(_ context.Context, name string) error

UpdateReservation is a no-op on Managed Kafka.

func (*KafkaAdminClient) UpdateSubscription added in v1.9.0 

func (c *KafkaAdminClient) UpdateSubscription(_ context.Context, groupID string) error

UpdateSubscription is a no-op on Managed Kafka. Kafka consumer groups lack centralized mutable configuration.

func (*KafkaAdminClient) UpdateTopic added in v1.9.0 

func (c *KafkaAdminClient) UpdateTopic(_ context.Context, topic string) error

UpdateTopic is a no-op on Managed Kafka. Kafka does not support mutable topic configurations. A WARNING is logged for visibility.

type KafkaAdminClientConfig added in v1.9.0 

type KafkaAdminClientConfig struct {
	// BootstrapServers is the Kafka bootstrap server address.
	// Use BuildGMKBootstrapServer() to construct this for GMK clusters.
	BootstrapServers string

	// SaramaConfig is an optional pre-built Sarama configuration. If nil,
	// NewGMKSaramaConfig() is called to build one with GCP OAUTHBEARER auth.
	SaramaConfig *sarama.Config
}

KafkaAdminClientConfig configures a KafkaAdminClient.

type KafkaCursorClient added in v1.9.0 

type KafkaCursorClient struct {
	// contains filtered or unexported fields
}

KafkaCursorClient manages consumer group offsets on Managed Kafka. It provides PSL CursorClient-equivalent operations (commit, read, reset, seek to end), mapping PSL "subscription" to a Kafka consumer group ID.

Consumers must not be actively joined to the group when calling Reset* methods; Kafka disallows external offset commits on groups with live members. For typical migration flows this is not a concern because the cutover sequence terminates PSL subscribers before invoking the migration tooling.

func NewKafkaCursorClient added in v1.9.0 

func NewKafkaCursorClient(ctx context.Context, cfg *KafkaCursorClientConfig) (*KafkaCursorClient, error)

NewKafkaCursorClient creates a cursor client connected to Managed Kafka.

func (*KafkaCursorClient) Close added in v1.9.0 

func (c *KafkaCursorClient) Close() error

Close releases the underlying connection.

func (*KafkaCursorClient) CommitOffset added in v1.9.0 

func (c *KafkaCursorClient) CommitOffset(_ context.Context, groupID, topic string, partition int32, offset int64) error

CommitOffset commits a single partition's offset for the given consumer group.

func (*KafkaCursorClient) GetCommittedOffset added in v1.9.0 

func (c *KafkaCursorClient) GetCommittedOffset(_ context.Context, groupID, topic string, partition int32) (int64, error)

GetCommittedOffset returns the committed offset for a single partition. Returns -1 (no offset) if the group has never committed on that partition.

func (*KafkaCursorClient) ReadCommittedOffsets added in v1.9.0 

func (c *KafkaCursorClient) ReadCommittedOffsets(_ context.Context, groupID, topic string) (map[int32]int64, error)

ReadCommittedOffsets returns the currently committed offsets for the group on the given topic. Partitions with no committed offset are omitted from the result.

func (*KafkaCursorClient) ResetOffsets added in v1.9.0 

func (c *KafkaCursorClient) ResetOffsets(_ context.Context, groupID, topic string, offsets map[int32]int64) error

ResetOffsets commits the provided partition→offset map for the given group and topic in a single OffsetCommit RPC.

func (*KafkaCursorClient) SeekToEnd added in v1.9.0 

func (c *KafkaCursorClient) SeekToEnd(ctx context.Context, groupID, topic string, partitionCount int) (map[int32]int64, error)

SeekToEnd resets the consumer group's committed offset to the latest offset on each of the given partitions. Returns the offsets that were committed.

type KafkaCursorClientConfig added in v1.9.0 

type KafkaCursorClientConfig struct {
	BootstrapServers string
	SaramaConfig     *sarama.Config
}

KafkaCursorClientConfig configures a KafkaCursorClient.

type KafkaPublishConfig added in v1.9.0 

type KafkaPublishConfig struct {
	// BootstrapServers is the Kafka bootstrap server address.
	// Use BuildGMKBootstrapServer() to construct this for GMK clusters.
	BootstrapServers string

	// TopicName is the Kafka topic name to publish to.
	TopicName string

	// SaramaConfig is an optional pre-built Sarama configuration. If nil,
	// NewGMKSaramaConfig() will be called to build one with GCP OAUTHBEARER
	// authentication.
	SaramaConfig *sarama.Config
}

KafkaPublishConfig holds configuration for connecting to Google Managed Kafka for publishing.

type KafkaSubscribeConfig added in v1.9.0 

type KafkaSubscribeConfig struct {
	// BootstrapServers is the Kafka bootstrap server address.
	// Use BuildGMKBootstrapServer() to construct this for GMK clusters.
	BootstrapServers string

	// TopicName is the Kafka topic name to subscribe to.
	TopicName string

	// SubscriptionName is used as the Kafka consumer group ID. Consumers with the
	// same SubscriptionName share the load of consuming from the topic.
	SubscriptionName string

	// SaramaConfig is an optional pre-built Sarama configuration. If nil,
	// NewGMKSaramaConfig() will be called to build one with GCP OAUTHBEARER
	// authentication. Auto-commit is disabled for PSL-like semantics.
	SaramaConfig *sarama.Config
}

KafkaSubscribeConfig holds configuration for connecting to Google Managed Kafka for receiving messages.

type KafkaTopicStatsClient added in v1.9.0 

type KafkaTopicStatsClient struct {
	// contains filtered or unexported fields
}

KafkaTopicStatsClient exposes PSL-style offset and stats APIs backed by Kafka's ListOffsets protocol.

Degraded semantics (documented in design doc):

  • ComputeMessageStats does not populate min_publish_time (would require a message read).
  • Byte estimates default to 1 KB/message since Kafka provides no low-latency per-offset byte count.
  • ComputeCursorForEventTime is aliased to ComputeCursorForPublishTime (Kafka lacks a native event-time primitive).

func NewKafkaTopicStatsClient added in v1.9.0 

func NewKafkaTopicStatsClient(ctx context.Context, cfg *KafkaTopicStatsClientConfig) (*KafkaTopicStatsClient, error)

NewKafkaTopicStatsClient creates a TopicStats client connected to Managed Kafka.

func (*KafkaTopicStatsClient) Close added in v1.9.0 

func (c *KafkaTopicStatsClient) Close() error

Close releases the underlying connection.

func (*KafkaTopicStatsClient) ComputeBacklogBytes added in v1.9.0 

func (c *KafkaTopicStatsClient) ComputeBacklogBytes(_ context.Context, topic string, partition int32, groupID string, avgBytesPerMessage int64) (*BacklogInfo, error)

ComputeBacklogBytes returns backlog stats for a consumer group on a partition. If avgBytesPerMessage <= 0, a 1 KB default is used.

func (*KafkaTopicStatsClient) ComputeCursorForEventTime added in v1.9.0 

func (c *KafkaTopicStatsClient) ComputeCursorForEventTime(ctx context.Context, topic string, partition int32, t time.Time) (int64, bool, error)

ComputeCursorForEventTime aliases to ComputeCursorForPublishTime since Kafka lacks a native event-time primitive. All record timestamps are treated as publish time.

func (*KafkaTopicStatsClient) ComputeCursorForPublishTime added in v1.9.0 

func (c *KafkaTopicStatsClient) ComputeCursorForPublishTime(ctx context.Context, topic string, partition int32, t time.Time) (int64, bool, error)

ComputeCursorForPublishTime returns the earliest offset whose record timestamp is >= t. Equivalent to GetOffsetForTimestamp with a time.Time parameter for caller ergonomics.

func (*KafkaTopicStatsClient) ComputeMessageStats added in v1.9.0 

func (c *KafkaTopicStatsClient) ComputeMessageStats(_ context.Context, topic string, partition int32, startOffset, endOffset int64, avgBytesPerMessage int64) (*MessageStats, error)

ComputeMessageStats returns approximate stats for the offset range [startOffset, endOffset). Uses a default estimate of 1 KB per message for byte sizing; pass a non-zero avgBytesPerMessage to override.

func (*KafkaTopicStatsClient) GetEarliestOffset added in v1.9.0 

func (c *KafkaTopicStatsClient) GetEarliestOffset(_ context.Context, topic string, partition int32) (int64, error)

GetEarliestOffset returns the earliest (oldest) offset available for the given partition.

func (*KafkaTopicStatsClient) GetEarliestOffsets added in v1.9.0 

func (c *KafkaTopicStatsClient) GetEarliestOffsets(ctx context.Context, topic string, partitionCount int) (map[int32]int64, error)

GetEarliestOffsets returns earliest offsets for the first partitionCount partitions of the topic.

func (*KafkaTopicStatsClient) GetLatestOffset added in v1.9.0 

func (c *KafkaTopicStatsClient) GetLatestOffset(_ context.Context, topic string, partition int32) (int64, error)

GetLatestOffset returns the latest (head) offset for the given partition. This is the offset of the next message to be produced, not the last message.

func (*KafkaTopicStatsClient) GetLatestOffsets added in v1.9.0 

func (c *KafkaTopicStatsClient) GetLatestOffsets(ctx context.Context, topic string, partitionCount int) (map[int32]int64, error)

GetLatestOffsets returns latest offsets for the first partitionCount partitions of the topic.

func (*KafkaTopicStatsClient) GetOffsetForTimestamp added in v1.9.0 

func (c *KafkaTopicStatsClient) GetOffsetForTimestamp(_ context.Context, topic string, partition int32, timestampMs int64) (int64, bool, error)

GetOffsetForTimestamp returns the earliest offset whose record timestamp is greater than or equal to timestampMs. The bool return is false when no such offset exists (e.g. timestamp is past the head).

type KafkaTopicStatsClientConfig added in v1.9.0 

type KafkaTopicStatsClientConfig struct {
	BootstrapServers string
	SaramaConfig     *sarama.Config
}

KafkaTopicStatsClientConfig configures a KafkaTopicStatsClient.

type KeyExtractorFunc 

type KeyExtractorFunc func(*pubsub.Message) []byte

KeyExtractorFunc is a function that extracts an ordering key from a Message.

type MessageMetadata added in v0.7.0 

type MessageMetadata struct {
	// The topic partition the message was published to.
	Partition int

	// The offset the message was assigned.
	//
	// If this MessageMetadata was returned for a publish result and publish
	// idempotence was enabled, the offset may be -1 when the message was
	// identified as a duplicate of an already successfully published message,
	// but the server did not have sufficient information to return the message's
	// offset at publish time. Messages received by subscribers will always have
	// the correct offset.
	Offset int64
}

MessageMetadata holds properties of a message published to the Pub/Sub Lite service.

func ParseMessageMetadata added in v0.7.0 

func ParseMessageMetadata(id string) (*MessageMetadata, error)

ParseMessageMetadata creates MessageMetadata from the ID string of a pubsub.PublishResult returned by PublisherClient or pubsub.Message.ID received from SubscriberClient.

Example (Publisher) 
package main

import (
	"context"
	"fmt"

	"cloud.google.com/go/pubsub"
	"cloud.google.com/go/pubsublite/pscompat"
)

func main() {
	ctx := context.Background()
	const topic = "projects/my-project/locations/region-or-zone/topics/my-topic"
	publisher, err := pscompat.NewPublisherClient(ctx, topic)
	if err != nil {
		// TODO: Handle error.
	}
	defer publisher.Stop()

	result := publisher.Publish(ctx, &pubsub.Message{Data: []byte("payload")})
	id, err := result.Get(ctx)
	if err != nil {
		// TODO: Handle error.
	}
	metadata, err := pscompat.ParseMessageMetadata(id)
	if err != nil {
		// TODO: Handle error.
	}
	fmt.Printf("Published message to partition %d with offset %d\n", metadata.Partition, metadata.Offset)
}
Example (Subscriber) 
package main

import (
	"context"
	"fmt"

	"cloud.google.com/go/pubsub"
	"cloud.google.com/go/pubsublite/pscompat"
)

func main() {
	ctx := context.Background()
	const subscription = "projects/my-project/locations/region-or-zone/subscriptions/my-subscription"
	subscriber, err := pscompat.NewSubscriberClient(ctx, subscription)
	if err != nil {
		// TODO: Handle error.
	}
	err = subscriber.Receive(ctx, func(ctx context.Context, m *pubsub.Message) {
		// TODO: Handle message.
		m.Ack()
		metadata, err := pscompat.ParseMessageMetadata(m.ID)
		if err != nil {
			// TODO: Handle error.
		}
		fmt.Printf("Received message from partition %d with offset %d\n", metadata.Partition, metadata.Offset)
	})
	if err != nil {
		// TODO: Handle error.
	}
}

func (*MessageMetadata) String added in v0.7.0 

func (m *MessageMetadata) String() string

type MessageStats added in v1.9.0 

type MessageStats struct {
	// MessageCount is the number of messages in the offset range.
	MessageCount int64
	// MessageBytes is the estimated byte count of messages in the range
	// (MessageCount * avgBytesPerMessage).
	MessageBytes int64
}

MessageStats mirrors the subset of PSL ComputeMessageStatsResponse that Kafka can cheaply populate. MinPublishTime is left unset since Kafka lacks a low-latency mechanism to retrieve it without reading messages.

type MessagingBackend added in v1.9.0 

type MessagingBackend int

MessagingBackend specifies the messaging backend to use for Publisher and Subscriber clients. 

const (
	// PubSubLite uses Google Cloud Pub/Sub Lite as the backend (default).
	// This is the traditional backend with zonal storage and predictable pr
	PubSubLite MessagingBackend = iota

	// ManagedKafka uses Google Cloud Managed Service for Apache Kafka as th
	// backend. Provides Kafka-compatible API with Google Cloud management.
	ManagedKafka
)

type MigrationConfig added in v1.9.0 

type MigrationConfig struct {
	// PSL side (source).
	PSLCursorClient     *vkit.CursorClient
	PSLTopicStatsClient *vkit.TopicStatsClient
	PSLTopicPath        string // "projects/P/locations/L/topics/T"
	PSLSubscriptionPath string // "projects/P/locations/L/subscriptions/S"

	// Kafka side (destination).
	KafkaTopicStats *KafkaTopicStatsClient
	KafkaCursor     *KafkaCursorClient
	KafkaTopicName  string
	KafkaGroupID    string // consumer group (PSL subscription equivalent)
	Partitions      []int32

	// DryRun resolves offsets but skips the commit.
	DryRun bool
	// Validate re-reads committed offsets after commit to confirm accuracy.
	Validate bool
}

MigrationConfig captures the clients and identifiers needed to perform an offset migration from PSL to Managed Kafka.

type MigrationOrchestrator added in v1.9.0 

type MigrationOrchestrator struct {
	Config *MigrationConfig
}

MigrationOrchestrator coordinates a full migration lifecycle with dry-run gating, result summary, and a post-flight backlog check.

func NewMigrationOrchestrator added in v1.9.0 

func NewMigrationOrchestrator(cfg *MigrationConfig) *MigrationOrchestrator

NewMigrationOrchestrator creates a new orchestrator wrapping the given config.

func (*MigrationOrchestrator) CheckBacklog added in v1.9.0 

func (o *MigrationOrchestrator) CheckBacklog(ctx context.Context) (map[int32]int64, error)

CheckBacklog reports, per partition, how many messages are between the (now-migrated) committed offset and the current head.

func (*MigrationOrchestrator) Execute added in v1.9.0 

func (o *MigrationOrchestrator) Execute(ctx context.Context) (*MigrationSummary, error)

Execute runs the full migration and returns a summary of per-partition outcomes.

type MigrationResult added in v1.9.0 

type MigrationResult struct {
	Status      MigrationStatus
	PSLOffset   int64
	KafkaOffset int64 // -1 if unknown
	Message     string
	Validated   bool
}

MigrationResult records the outcome of migrating one partition's offset.

func (MigrationResult) String added in v1.9.0 

func (r MigrationResult) String() string

String renders the result in the same shape as the Java toString().

type MigrationStatus added in v1.9.0 

type MigrationStatus int

MigrationStatus enumerates the outcome of migrating a single partition. 

const (
	// MigrationSuccess indicates the offset was resolved and committed.
	MigrationSuccess MigrationStatus = iota
	// MigrationDryRun indicates the offset was resolved but not committed.
	MigrationDryRun
	// MigrationNoPSLOffset indicates no PSL cursor existed; reset to earliest
	// Kafka offset instead.
	MigrationNoPSLOffset
	// MigrationNoKafkaOffset indicates no Kafka offset at/after the target
	// timestamp.
	MigrationNoKafkaOffset
	// MigrationValidationFailed indicates post-commit validation saw an offset
	// different from what was committed.
	MigrationValidationFailed
	// MigrationError indicates execution failure. Inspect Message for details.
	MigrationError
)

func (MigrationStatus) String added in v1.9.0 

func (s MigrationStatus) String() string

String returns the uppercased status name (matches Java enum names).

type MigrationSummary added in v1.9.0 

type MigrationSummary struct {
	PartitionResults map[int32]MigrationResult
	Successful       bool
	// Summary is a formatted multi-line report suitable for direct logging.
	Summary string
}

MigrationSummary aggregates per-partition results.

type NackHandler 

type NackHandler func(*pubsub.Message) error

NackHandler is invoked when pubsub.Message.Nack() is called. Pub/Sub Lite does not have a concept of 'nack'. If the nack handler implementation returns nil, the message is acknowledged. If an error is returned, the SubscriberClient will consider this a fatal error and terminate.

In Pub/Sub Lite, only a single subscriber for a given subscription is connected to any partition at a time, and there is no other client that may be able to handle messages.

type PublishMessageTransformerFunc 

type PublishMessageTransformerFunc func(*pubsub.Message, *pb.PubSubMessage) error

PublishMessageTransformerFunc transforms a pubsub.Message to a Pub/Sub Lite PubSubMessage API proto. If this returns an error, the pubsub.PublishResult will be errored and the PublisherClient will consider this a fatal error and terminate.

type PublishSettings 

type PublishSettings struct {
	// Publish a non-empty batch after this delay has passed. If DelayThreshold is
	// 0, it will be treated as DefaultPublishSettings.DelayThreshold. Otherwise
	// must be > 0.
	DelayThreshold time.Duration

	// Publish a batch when it has this many messages. The maximum is
	// MaxPublishRequestCount. If CountThreshold is 0, it will be treated as
	// DefaultPublishSettings.CountThreshold. Otherwise must be > 0.
	CountThreshold int

	// Publish a batch when its size in bytes reaches this value. The maximum is
	// MaxPublishRequestBytes. If ByteThreshold is 0, it will be treated as
	// DefaultPublishSettings.ByteThreshold. Otherwise must be > 0.
	ByteThreshold int

	// The maximum time that the client will attempt to open a publish stream
	// to the server. If Timeout is 0, it will be treated as
	// DefaultPublishSettings.Timeout, otherwise will be clamped to 2 minutes. In
	// the future, setting Timeout to less than 2 minutes will result in an error.
	//
	// If your application has a low tolerance to backend unavailability, set
	// Timeout to a lower duration to detect and handle. When the timeout is
	// exceeded, the PublisherClient will terminate with ErrBackendUnavailable and
	// details of the last error that occurred while trying to reconnect to
	// backends. Note that if the timeout duration is long, ErrOverflow may occur
	// first.
	//
	// If no failover operations need to be performed by the application, it is
	// recommended to just use the default timeout value to avoid the
	// PublisherClient terminating during short periods of backend unavailability.
	Timeout time.Duration

	// The maximum number of bytes that the publisher will keep in memory before
	// returning ErrOverflow. If BufferedByteLimit is 0, it will be treated as
	// DefaultPublishSettings.BufferedByteLimit. Otherwise must be > 0.
	//
	// Note that this setting applies per partition. If BufferedByteLimit is being
	// used to bound memory usage, keep in mind the number of partitions in the
	// topic.
	//
	// Note that Pub/Sub Lite topics are provisioned a publishing throughput
	// capacity, per partition, shared by all publisher clients. Setting a large
	// buffer size can mitigate transient publish spikes. However, consistently
	// attempting to publish messages at a much higher rate than the publishing
	// throughput capacity can cause the buffers to overflow. For more
	// information, see https://cloud.google.com/pubsub/lite/docs/topics.
	BufferedByteLimit int

	// Whether idempotence is enabled, where the server will ensure that unique
	// messages within a single publisher session are stored only once. Default
	// true.
	EnableIdempotence optional.Bool

	// Optional custom function that extracts an ordering key from a Message. The
	// default implementation extracts the key from Message.OrderingKey.
	KeyExtractor KeyExtractorFunc

	// Optional custom function that transforms a pubsub.Message to a
	// PubSubMessage API proto.
	MessageTransformer PublishMessageTransformerFunc

	// Backend specifies the messaging backend to use. Default is PubSubLite.
	// Set to ManagedKafka to publish to Google Managed Kafka.
	Backend MessagingBackend

	// KafkaConfig holds configuration for connecting to Google Managed Kafka.
	// Required when Backend is ManagedKafka.
	KafkaConfig *KafkaPublishConfig
	// contains filtered or unexported fields
}

PublishSettings configure the PublisherClient. Batching settings (DelayThreshold, CountThreshold, ByteThreshold, BufferedByteLimit) apply per partition.

A zero PublishSettings will result in values equivalent to DefaultPublishSettings.

type PublisherClient 

type PublisherClient struct {
	// contains filtered or unexported fields
}

PublisherClient is a Pub/Sub Lite client to publish messages to a given topic. A PublisherClient is safe to use from multiple goroutines.

PublisherClients are expected to be long-lived and used for the duration of the application, rather than for publishing small batches of messages. Stop must be called to release resources when a PublisherClient is no longer required.

See https://cloud.google.com/pubsub/lite/docs/publishing for more information about publishing.

func NewPublisherClient 

func NewPublisherClient(ctx context.Context, topic string, opts ...option.ClientOption) (*PublisherClient, error)

NewPublisherClient creates a new Pub/Sub Lite publisher client to publish messages to a given topic, using DefaultPublishSettings. A valid topic path has the format: "projects/PROJECT_ID/locations/LOCATION/topics/TOPIC_ID".

Stop must be called to release resources when a PublisherClient is no longer required.

Example (EarlyTokenRefresh)

This example illustrates how to configure OAuth tokens to be refreshed 5 minutes before they expire, in order to mitigate delays which may occur during refresh. 

package main

import (
	"context"
	"fmt"
	"log"
	"time"

	"cloud.google.com/go/pubsub"
	api "cloud.google.com/go/pubsublite/apiv1"
	"cloud.google.com/go/pubsublite/pscompat"
	"golang.org/x/oauth2/google"
	"google.golang.org/api/option"
)

func main() {
	ctx := context.Background()
	const topic = "projects/my-project/locations/region-or-zone/topics/my-topic"
	params := google.CredentialsParams{
		Scopes:            api.DefaultAuthScopes(),
		EarlyTokenRefresh: 5 * time.Minute,
	}
	creds, err := google.FindDefaultCredentialsWithParams(ctx, params)
	if err != nil {
		log.Fatalf("No 'Application Default Credentials' found: %v.", err)
	}
	publisher, err := pscompat.NewPublisherClient(ctx, topic, option.WithTokenSource(creds.TokenSource))
	if err != nil {
		// TODO: Handle error.
	}
	defer publisher.Stop()

	var results []*pubsub.PublishResult
	r := publisher.Publish(ctx, &pubsub.Message{
		Data: []byte("hello world"),
	})
	results = append(results, r)
	// Publish more messages ...

	var publishFailed bool
	for _, r := range results {
		id, err := r.Get(ctx)
		if err != nil {
			// TODO: Handle error.
			publishFailed = true
			continue
		}
		fmt.Printf("Published a message with a message ID: %s\n", id)
	}

	// NOTE: A failed PublishResult indicates that the publisher client
	// encountered a fatal error and has permanently terminated. After the fatal
	// error has been resolved, a new publisher client instance must be created to
	// republish failed messages.
	if publishFailed {
		fmt.Printf("Publisher client terminated due to error: %v\n", publisher.Error())
	}
}
Example (Interface)

This example illustrates how to declare a common interface for publisher clients from Cloud Pub/Sub (cloud.google.com/go/pubsub) and Pub/Sub Lite (cloud.google.com/go/pubsublite/pscompat). 

package main

import (
	"context"

	"cloud.google.com/go/pubsub"
	"cloud.google.com/go/pubsublite/pscompat"
)

func main() {
	// publisherInterface is implemented by both pscompat.PublisherClient and
	// pubsub.Topic.
	type publisherInterface interface {
		Publish(context.Context, *pubsub.Message) *pubsub.PublishResult
		Stop()
	}

	publish := func(publisher publisherInterface) {
		defer publisher.Stop()
		// TODO: Publish messages.
	}

	// Create a Pub/Sub Lite publisher client.
	ctx := context.Background()
	publisher, err := pscompat.NewPublisherClient(ctx, "projects/my-project/locations/region-or-zone/topics/my-topic")
	if err != nil {
		// TODO: Handle error.
	}
	publish(publisher)

	// Create a Cloud Pub/Sub topic to publish.
	client, err := pubsub.NewClient(ctx, "my-project")
	if err != nil {
		// TODO: Handle error.
	}
	topic := client.Topic("my-topic")
	publish(topic)
}

func NewPublisherClientWithSettings 

func NewPublisherClientWithSettings(ctx context.Context, topic string, settings PublishSettings, opts ...option.ClientOption) (*PublisherClient, error)

NewPublisherClientWithSettings creates a new Pub/Sub Lite publisher client to publish messages to a given topic, using the specified PublishSettings. A valid topic path has the format: "projects/PROJECT_ID/locations/LOCATION/topics/TOPIC_ID".

Stop must be called to release resources when a PublisherClient is no longer required.

Example (BatchingSettings)

This example illustrates how to set batching settings for publishing. Note that batching settings apply per partition. If BufferedByteLimit is being used to bound memory usage, keep in mind the number of partitions in the topic. 

package main

import (
	"context"
	"fmt"
	"time"

	"cloud.google.com/go/pubsub"
	"cloud.google.com/go/pubsublite/pscompat"
)

func main() {
	ctx := context.Background()
	const topic = "projects/my-project/locations/region-or-zone/topics/my-topic"
	settings := pscompat.PublishSettings{
		DelayThreshold:    50 * time.Millisecond,
		CountThreshold:    200,
		BufferedByteLimit: 5e8,
	}
	publisher, err := pscompat.NewPublisherClientWithSettings(ctx, topic, settings)
	if err != nil {
		// TODO: Handle error.
	}
	defer publisher.Stop()

	var results []*pubsub.PublishResult
	r := publisher.Publish(ctx, &pubsub.Message{
		Data: []byte("hello world"),
	})
	results = append(results, r)
	// Publish more messages ...

	var publishFailed bool
	for _, r := range results {
		id, err := r.Get(ctx)
		if err != nil {
			// TODO: Handle error.
			publishFailed = true
			continue
		}
		fmt.Printf("Published a message with a message ID: %s\n", id)
	}

	// NOTE: A failed PublishResult indicates that the publisher client
	// encountered a fatal error and has permanently terminated. After the fatal
	// error has been resolved, a new publisher client instance must be created to
	// republish failed messages.
	if publishFailed {
		fmt.Printf("Publisher client terminated due to error: %v\n", publisher.Error())
	}
}

func (*PublisherClient) Error 

func (p *PublisherClient) Error() error

Error returns the error that caused the publisher client to terminate. The error returned here may contain more context than PublishResult errors. The return value may be nil if Stop() was called.

func (*PublisherClient) Publish 

func (p *PublisherClient) Publish(ctx context.Context, msg *pubsub.Message) *pubsub.PublishResult

Publish publishes `msg` to the topic asynchronously. Messages are batched and sent according to the client's PublishSettings. Publish never blocks.

Publish returns a non-nil PublishResult which will be ready when the message has been sent (or has failed to be sent) to the server. Retryable errors are automatically handled. If a PublishResult returns an error, this indicates that the publisher client encountered a fatal error and can no longer be used. Fatal errors should be manually inspected and the cause resolved. A new publisher client instance must be created to republish failed messages.

Once Stop() has been called or the publisher client has failed permanently due to an error, future calls to Publish will immediately return a PublishResult with error ErrPublisherStopped.

Error() returns the error that caused the publisher client to terminate and may contain more context than the error returned by PublishResult.

Example 
package main

import (
	"context"
	"fmt"

	"cloud.google.com/go/pubsub"
	"cloud.google.com/go/pubsublite/pscompat"
)

func main() {
	ctx := context.Background()
	const topic = "projects/my-project/locations/region-or-zone/topics/my-topic"
	publisher, err := pscompat.NewPublisherClient(ctx, topic)
	if err != nil {
		// TODO: Handle error.
	}
	defer publisher.Stop()

	var results []*pubsub.PublishResult
	r := publisher.Publish(ctx, &pubsub.Message{
		Data: []byte("hello world"),
	})
	results = append(results, r)
	// Publish more messages ...

	var publishFailed bool
	for _, r := range results {
		id, err := r.Get(ctx)
		if err != nil {
			// TODO: Handle error.
			publishFailed = true
			continue
		}
		fmt.Printf("Published a message with a message ID: %s\n", id)
	}

	// NOTE: A failed PublishResult indicates that the publisher client
	// encountered a fatal error and has permanently terminated. After the fatal
	// error has been resolved, a new publisher client instance must be created to
	// republish failed messages.
	if publishFailed {
		fmt.Printf("Publisher client terminated due to error: %v\n", publisher.Error())
	}
}
Example (ErrorHandling)

This example illustrates how to handle various publishing errors. Some errors can be automatically handled (e.g. backend unavailable and buffer overflow), while others are fatal errors that should be inspected. If the application has a low tolerance to backend unavailability, set a lower PublishSettings.Timeout value to detect and alert. 

package main

import (
	"context"
	"errors"
	"fmt"
	"sync"
	"time"

	"cloud.google.com/go/pubsub"
	"cloud.google.com/go/pubsublite/pscompat"
	"golang.org/x/sync/errgroup"
)

func main() {
	ctx := context.Background()
	const topic = "projects/my-project/locations/region-or-zone/topics/my-topic"
	settings := pscompat.PublishSettings{
		// The PublisherClient will terminate when it cannot connect to backends for
		// more than 10 minutes.
		Timeout: 10 * time.Minute,
		// Sets a conservative publish buffer byte limit, per partition.
		BufferedByteLimit: 1e8,
	}
	publisher, err := pscompat.NewPublisherClientWithSettings(ctx, topic, settings)
	if err != nil {
		// TODO: Handle error.
	}
	defer publisher.Stop()

	var toRepublish []*pubsub.Message
	var mu sync.Mutex
	g := new(errgroup.Group)

	for i := 0; i < 10; i++ {
		msg := &pubsub.Message{
			Data: []byte(fmt.Sprintf("message-%d", i)),
		}
		result := publisher.Publish(ctx, msg)

		g.Go(func() error {
			id, err := result.Get(ctx)
			if err != nil {
				// NOTE: A failed PublishResult indicates that the publisher client has
				// permanently terminated. A new publisher client instance must be
				// created to republish failed messages.
				fmt.Printf("Publish error: %v\n", err)
				// Oversized messages cannot be published.
				if !errors.Is(err, pscompat.ErrOversizedMessage) {
					mu.Lock()
					toRepublish = append(toRepublish, msg)
					mu.Unlock()
				}
				return err
			}
			fmt.Printf("Published a message with a message ID: %s\n", id)
			return nil
		})
	}
	if err := g.Wait(); err != nil {
		fmt.Printf("Publisher client terminated due to error: %v\n", publisher.Error())
		switch {
		case errors.Is(publisher.Error(), pscompat.ErrBackendUnavailable):
			// TODO: Create a new publisher client to republish failed messages.
		case errors.Is(publisher.Error(), pscompat.ErrOverflow):
			// TODO: Create a new publisher client to republish failed messages.
			// Throttle publishing. Note that backend unavailability can also cause
			// buffer overflow before the ErrBackendUnavailable error.
		default:
			// TODO: Inspect and handle fatal error.
		}
	}
}

func (*PublisherClient) Stop 

func (p *PublisherClient) Stop()

Stop sends all remaining published messages and closes publish streams. Returns once all outstanding messages have been sent or have failed to be sent. Stop should be called when the client is no longer required.

type ReassignmentHandlerFunc added in v1.2.0 

type ReassignmentHandlerFunc func(previousPartitions, nextPartitions []int) error

ReassignmentHandlerFunc is called any time a new partition assignment is received from the server. It will be called with both the previous and new partition numbers as decided by the server. Both slices of partition numbers are sorted in ascending order.

When this handler is called, partitions that are being assigned away are stopping and new partitions are starting. Acks and nacks for messages from partitions that are being assigned away will have no effect, but message deliveries may still be in flight.

The client library will not acknowledge the assignment until this handler returns. The server will not assign any of the partitions in `previousPartitions` to another client unless the assignment is acknowledged, or a client takes too long to acknowledge (currently 30 seconds from the time the assignment is sent from server's point of view).

Because of the above, as long as reassignment handling is processed quickly, it can be used to abort outstanding operations on partitions which are being assigned away from this client.

If this handler returns an error, the SubscriberClient will consider this a fatal error and terminate.

type ReceiveMessageTransformerFunc 

type ReceiveMessageTransformerFunc func(*pb.SequencedMessage, *pubsub.Message) error

ReceiveMessageTransformerFunc transforms a Pub/Sub Lite SequencedMessage API proto to a pubsub.Message. The implementation must not set pubsub.Message.ID.

If this returns an error, the SubscriberClient will consider this a fatal error and terminate.

type ReceiveSettings 

type ReceiveSettings struct {
	// MaxOutstandingMessages is the maximum number of unacknowledged messages.
	// If MaxOutstandingMessages is 0, it will be treated as
	// DefaultReceiveSettings.MaxOutstandingMessages. Otherwise must be > 0.
	MaxOutstandingMessages int

	// MaxOutstandingBytes is the maximum size (in quota bytes) of unacknowledged
	// messages. If MaxOutstandingBytes is 0, it will be treated as
	// DefaultReceiveSettings.MaxOutstandingBytes. Otherwise must be > 0.
	//
	// Note that this setting applies per partition. If MaxOutstandingBytes is
	// being used to bound memory usage, keep in mind the number of partitions in
	// the associated topic.
	MaxOutstandingBytes int

	// The maximum time that the client will attempt to open a subscribe stream
	// to the server. If Timeout is 0, it will be treated as
	// DefaultReceiveSettings.Timeout, otherwise will be clamped to 2 minutes. In
	// the future, setting Timeout to less than 2 minutes will result in an error.
	//
	// If your application has a low tolerance to backend unavailability, set
	// Timeout to a lower duration to detect and handle. When the timeout is
	// exceeded, the SubscriberClient will terminate with ErrBackendUnavailable
	// and details of the last error that occurred while trying to reconnect to
	// backends.
	//
	// If no failover operations need to be performed by the application, it is
	// recommended to just use the default timeout value to avoid the
	// SubscriberClient terminating during short periods of backend
	// unavailability.
	Timeout time.Duration

	// The topic partition numbers (zero-indexed) to receive messages from.
	// Values must be less than the number of partitions for the topic. If not
	// specified, the SubscriberClient will use the partition assignment service
	// to determine which partitions it should connect to.
	Partitions []int

	// Optional custom function to handle pubsub.Message.Nack() calls. If not set,
	// the default behavior is to terminate the SubscriberClient.
	NackHandler NackHandler

	// Optional custom function that transforms a SequencedMessage API proto to a
	// pubsub.Message.
	MessageTransformer ReceiveMessageTransformerFunc

	// Optional custom function that is called when a new partition assignment has
	// been delivered to the client.
	ReassignmentHandler ReassignmentHandlerFunc

	// Backend specifies the messaging backend to use. Default is PubSubLite.
	// Set to ManagedKafka to receive from Google Managed Kafka.
	Backend MessagingBackend

	// KafkaConfig holds configuration for connecting to Google Managed Kafka.
	// Required when Backend is ManagedKafka.
	KafkaConfig *KafkaSubscribeConfig
}

ReceiveSettings configure the SubscriberClient. Flow control settings (MaxOutstandingMessages, MaxOutstandingBytes) apply per partition.

A zero ReceiveSettings will result in values equivalent to DefaultReceiveSettings.

type SubscriberClient 

type SubscriberClient struct {
	// contains filtered or unexported fields
}

SubscriberClient is a Pub/Sub Lite client to receive messages for a given subscription.

See https://cloud.google.com/pubsub/lite/docs/subscribing for more information about receiving messages.

func NewSubscriberClient 

func NewSubscriberClient(ctx context.Context, subscription string, opts ...option.ClientOption) (*SubscriberClient, error)

NewSubscriberClient creates a new Pub/Sub Lite client to receive messages for a given subscription, using DefaultReceiveSettings. A valid subscription path has the format: "projects/PROJECT_ID/locations/LOCATION/subscriptions/SUBSCRIPTION_ID".

Example (Interface)

This example illustrates how to declare a common interface for subscriber clients from Cloud Pub/Sub (cloud.google.com/go/pubsub) and Pub/Sub Lite (cloud.google.com/go/pubsublite/pscompat). 

package main

import (
	"context"

	"cloud.google.com/go/pubsub"
	"cloud.google.com/go/pubsublite/pscompat"
)

func main() {
	// subscriberInterface is implemented by both pscompat.SubscriberClient and
	// pubsub.Subscription.
	type subscriberInterface interface {
		Receive(context.Context, func(context.Context, *pubsub.Message)) error
	}

	receive := func(subscriber subscriberInterface) {
		// TODO: Receive messages.
	}

	// Create a Pub/Sub Lite subscriber client.
	ctx := context.Background()
	subscriber, err := pscompat.NewSubscriberClient(ctx, "projects/my-project/locations/region-or-zone/subscriptions/my-subscription")
	if err != nil {
		// TODO: Handle error.
	}
	receive(subscriber)

	// Create a Cloud Pub/Sub subscription to receive.
	client, err := pubsub.NewClient(ctx, "my-project")
	if err != nil {
		// TODO: Handle error.
	}
	subscription := client.Subscription("my-subscription")
	receive(subscription)
}

func NewSubscriberClientWithSettings 

func NewSubscriberClientWithSettings(ctx context.Context, subscription string, settings ReceiveSettings, opts ...option.ClientOption) (*SubscriberClient, error)

NewSubscriberClientWithSettings creates a new Pub/Sub Lite client to receive messages for a given subscription, using the specified ReceiveSettings. A valid subscription path has the format: "projects/PROJECT_ID/locations/LOCATION/subscriptions/SUBSCRIPTION_ID".

Example (ManualPartitionAssignment)

This example shows how to manually assign which topic partitions a SubscriberClient should connect to. If not specified, the SubscriberClient will use Pub/Sub Lite's partition assignment service to automatically determine which partitions it should connect to. 

package main

import (
	"context"

	"cloud.google.com/go/pubsub"
	"cloud.google.com/go/pubsublite/pscompat"
)

func main() {
	ctx := context.Background()
	const subscription = "projects/my-project/locations/region-or-zone/subscriptions/my-subscription"
	settings := pscompat.ReceiveSettings{
		// NOTE: The corresponding topic must have 2 or more partitions.
		Partitions: []int{0, 1},
	}
	subscriber, err := pscompat.NewSubscriberClientWithSettings(ctx, subscription, settings)
	if err != nil {
		// TODO: Handle error.
	}
	cctx, cancel := context.WithCancel(ctx)
	err = subscriber.Receive(cctx, func(ctx context.Context, m *pubsub.Message) {
		// TODO: Handle message.
		// NOTE: May be called concurrently; synchronize access to shared memory.
		m.Ack()
	})
	if err != nil {
		// TODO: Handle error.
	}

	// Call cancel from the receiver callback or another goroutine to stop
	// receiving.
	cancel()
}
Example (MaxOutstanding)

This example shows how to throttle SubscriberClient.Receive, which aims for high throughput by default. By limiting the number of messages and/or bytes being processed at once, you can bound your program's resource consumption. Note that ReceiveSettings apply per partition, so keep in mind the number of partitions in the associated topic. 

package main

import (
	"context"

	"cloud.google.com/go/pubsub"
	"cloud.google.com/go/pubsublite/pscompat"
)

func main() {
	ctx := context.Background()
	const subscription = "projects/my-project/locations/region-or-zone/subscriptions/my-subscription"
	settings := pscompat.ReceiveSettings{
		MaxOutstandingMessages: 5,
		MaxOutstandingBytes:    10e6,
	}
	subscriber, err := pscompat.NewSubscriberClientWithSettings(ctx, subscription, settings)
	if err != nil {
		// TODO: Handle error.
	}
	cctx, cancel := context.WithCancel(ctx)
	err = subscriber.Receive(cctx, func(ctx context.Context, m *pubsub.Message) {
		// TODO: Handle message.
		// NOTE: May be called concurrently; synchronize access to shared memory.
		m.Ack()
	})
	if err != nil {
		// TODO: Handle error.
	}

	// Call cancel from the receiver callback or another goroutine to stop
	// receiving.
	cancel()
}

func (*SubscriberClient) Receive 

func (s *SubscriberClient) Receive(ctx context.Context, f func(context.Context, *pubsub.Message)) error

Receive calls f with the messages from the subscription. It blocks until ctx is done, or the service returns a non-retryable error.

The standard way to terminate a Receive is to cancel its context: 

cctx, cancel := context.WithCancel(ctx)
err := sub.Receive(cctx, callback)
// Call cancel from callback, or another goroutine.

If there is a fatal service error, Receive returns that error after all of the outstanding calls to f have returned. If ctx is done, Receive returns nil after all of the outstanding calls to f have returned and all messages have been acknowledged. The context passed to f will be canceled when ctx is Done or there is a fatal service error.

Receive calls f concurrently from multiple goroutines if the SubscriberClient is connected to multiple partitions. Only one call from any connected partition will be outstanding at a time, and blocking in the receiver callback f will block the delivery of subsequent messages for the partition.

All messages received by f must be ACKed or NACKed. Failure to do so can prevent Receive from returning. Messages may be processed by the client concurrently and ACKed asynchronously to increase throughput.

Each SubscriberClient may have only one invocation of Receive active at a time.

Example 
package main

import (
	"context"

	"cloud.google.com/go/pubsub"
	"cloud.google.com/go/pubsublite/pscompat"
)

func main() {
	ctx := context.Background()
	const subscription = "projects/my-project/locations/region-or-zone/subscriptions/my-subscription"
	subscriber, err := pscompat.NewSubscriberClient(ctx, subscription)
	if err != nil {
		// TODO: Handle error.
	}
	cctx, cancel := context.WithCancel(ctx)
	err = subscriber.Receive(cctx, func(ctx context.Context, m *pubsub.Message) {
		// TODO: Handle message.
		// NOTE: May be called concurrently; synchronize access to shared memory.
		m.Ack()
	})
	if err != nil {
		// TODO: Handle error.
	}

	// Call cancel from the receiver callback or another goroutine to stop
	// receiving.
	cancel()
}
Example (ErrorHandling)

If the application has a low tolerance to backend unavailability, set a lower ReceiveSettings.Timeout value to detect and alert. 

package main

import (
	"context"
	"errors"
	"fmt"
	"time"

	"cloud.google.com/go/pubsub"
	"cloud.google.com/go/pubsublite/pscompat"
)

func main() {
	ctx := context.Background()
	const subscription = "projects/my-project/locations/region-or-zone/subscriptions/my-subscription"
	settings := pscompat.ReceiveSettings{
		// The SubscriberClient will terminate when it cannot connect to backends
		// for more than 5 minutes.
		Timeout: 5 * time.Minute,
	}
	subscriber, err := pscompat.NewSubscriberClientWithSettings(ctx, subscription, settings)
	if err != nil {
		// TODO: Handle error.
	}

	for {
		cctx, cancel := context.WithCancel(ctx)
		err = subscriber.Receive(cctx, func(ctx context.Context, m *pubsub.Message) {
			// TODO: Handle message.
			// NOTE: May be called concurrently; synchronize access to shared memory.
			m.Ack()
		})
		if err != nil {
			cancel()
			fmt.Printf("Subscriber client stopped receiving due to error: %v\n", err)
			if errors.Is(err, pscompat.ErrBackendUnavailable) {
				// TODO: Alert if necessary. Receive can be retried.
			} else {
				// TODO: Handle fatal error.
				break
			}
		}

		// Call cancel from the receiver callback or another goroutine to stop
		// receiving.
		cancel()
	}
}

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.