Cloud PubSub Lite - Package cloud.google.com/go/pubsublite (v1.8.2)

Note: To get more information about this package, such as access to older versions, view this package on pkg.go.dev.

Package pubsublite provides an easy way to publish and receive messages usingthe Pub/Sub Lite service.

Google Pub/Sub services are designed to provide reliable, many-to-many,asynchronous messaging between applications. Publisher applications can sendmessages to a topic and other applications can subscribe to that topic toreceive the messages. By decoupling senders and receivers, Google Pub/Sub allowsdevelopers to communicate between independently written applications.

Compared to Cloud Pub/Sub, Pub/Sub Lite provides partitioned data storage withpredefined throughput and storage capacity. Guidance on how to choose betweenCloud Pub/Sub and Pub/Sub Lite is available athttps://cloud.google.com/pubsub/docs/choosing-pubsub-or-lite.

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

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

Introduction

Examples can be found athttps://pkg.go.dev/cloud.google.com/go/pubsublite#pkg-examplesandhttps://pkg.go.dev/cloud.google.com/go/pubsublite/pscompat#pkg-examples.

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

The cloud.google.com/go/pubsublite/pscompat subpackage contains clients forpublishing and receiving messages, which have similar interfaces to theirpubsub.Topic and pubsub.Subscription counterparts in cloud.google.com/go/pubsub.The following examples demonstrate how to declare common interfaces:https://pkg.go.dev/cloud.google.com/go/pubsublite/pscompat#example-NewPublisherClient-Interfaceandhttps://pkg.go.dev/cloud.google.com/go/pubsublite/pscompat#example-NewSubscriberClient-Interface.

The following imports are required for code snippets below:

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

Creating Topics

Messages are published to topics. Pub/Sub Lite topics may be created like so:

ctx:=context.Background()consttopicPath="projects/my-project/locations/us-central1-c/topics/my-topic"topicConfig:=pubsublite.TopicConfig{Name:topicPath,PartitionCount:1,PublishCapacityMiBPerSec:4,SubscribeCapacityMiBPerSec:4,PerPartitionBytes:30*1024*1024*1024,// 30 GiBRetentionDuration:pubsublite.InfiniteRetention,}adminClient,err:=pubsublite.NewAdminClient(ctx,"us-central1")iferr!=nil{// TODO: Handle error.}if_,err=adminClient.CreateTopic(ctx,topicConfig);err!=nil{// TODO: Handle error.}

Close must be called to release resources when an AdminClient is no longerrequired.

Seehttps://cloud.google.com/pubsub/lite/docs/topics for more information abouthow Pub/Sub Lite topics are configured.

Seehttps://cloud.google.com/pubsub/lite/docs/locations for the list oflocations where Pub/Sub Lite is available.

Publishing

Pub/Sub Lite uses gRPC streams extensively for high throughput. For moredifferences, seehttps://pkg.go.dev/cloud.google.com/go/pubsublite/pscompat.

To publish messages to a topic, first create a PublisherClient:

publisher,err:=pscompat.NewPublisherClient(ctx,topicPath)iferr!=nil{// TODO: Handle error.}

Then call Publish:

result:=publisher.Publish(ctx,&pubsub.Message{Data:[]byte("payload")})

Publish queues the message for publishing and returns immediately. When enoughmessages have accumulated, or enough time has elapsed, the batch of messages issent to the Pub/Sub Lite service. Thresholds for batching can be configured inPublishSettings.

Publish returns a PublishResult, which behaves like a future; its Get methodblocks until the message has been sent (or has failed to be sent) to theservice:

id,err:=result.Get(ctx)iferr!=nil{// TODO: Handle error.}

Once you've finishing publishing all messages, call Stop to flush all messagesto the service and close gRPC streams. The PublisherClient can no longer be usedafter it has been stopped or has terminated due to a permanent error.

publisher.Stop()

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

Seehttps://cloud.google.com/pubsub/lite/docs/publishing for more informationabout publishing.

Creating Subscriptions

To receive messages published to a topic, create a subscription to the topic.There may be more than one subscription per topic; each message that ispublished to the topic will be delivered to all of its subscriptions.

Pub/Sub Lite subscriptions may be created like so:

constsubscriptionPath="projects/my-project/locations/us-central1-c/subscriptions/my-subscription"subscriptionConfig:=pubsublite.SubscriptionConfig{Name:subscriptionPath,Topic:topicPath,DeliveryRequirement:pubsublite.DeliverImmediately,}if_,err=adminClient.CreateSubscription(ctx,subscriptionConfig);err!=nil{// TODO: Handle error.}

Seehttps://cloud.google.com/pubsub/lite/docs/subscriptions for more informationabout how subscriptions are configured.

Receiving

To receive messages for a subscription, first create a SubscriberClient:

subscriber,err:=pscompat.NewSubscriberClient(ctx,subscriptionPath)

Messages are then consumed from a subscription via callback. The callback may beinvoked concurrently by multiple goroutines (one per partition that thesubscriber client is connected to).

cctx,cancel:=context.WithCancel(ctx)err=subscriber.Receive(cctx,func(ctxcontext.Context,m*pubsub.Message){log.Printf("Got message: %s",m.Data)m.Ack()})iferr!=nil{// TODO: Handle error.}

Receive blocks until either the context is canceled or a permanent error occurs.To terminate a call to Receive, cancel its context:

cancel()

Clients must call pubsub.Message.Ack() or pubsub.Message.Nack() for everymessage received. Pub/Sub Lite does not have ACK deadlines. Pub/Sub Lite alsodoes not actually have the concept of NACK. The default behavior terminates theSubscriberClient. In Pub/Sub Lite, only a single subscriber for a givensubscription is connected to any partition at a time, and there is no otherclient that may be able to handle messages.

Seehttps://cloud.google.com/pubsub/lite/docs/subscribing for more informationabout receiving messages.

gRPC Connection Pools

Pub/Sub Lite utilizes gRPC streams extensively. gRPC allows a maximum of 100streams per connection. Internally, the library uses a default connection poolsize of 8, which supports up to 800 topic partitions. To alter the connectionpool size, pass a ClientOption to pscompat.NewPublisherClient andpscompat.NewSubscriberClient:

pub,err:=pscompat.NewPublisherClient(ctx,topicPath,option.WithGRPCConnectionPool(10))

Constants

InfiniteRetention

constInfiniteRetention=time.Duration(-1)

InfiniteRetention is a sentinel used in topic configs to denote an infiniteretention duration (i.e. retain messages as long as there is availablestorage).

AdminClient

typeAdminClientstruct{// contains filtered or unexported fields}

AdminClient provides admin operations for Pub/Sub Lite resources within aGoogle Cloud region. The location (region or zone) component of resourcepaths must be within this region.Seehttps://cloud.google.com/pubsub/lite/docs/locations for the list ofregions and zones where Pub/Sub Lite is available.

An AdminClient may be shared by multiple goroutines.

Close must be called to release resources when an AdminClient is no longerrequired. If the client is available for the lifetime of the program, thenClose need not be called at exit.

func NewAdminClient

funcNewAdminClient(ctxcontext.Context,regionstring,opts...option.ClientOption)(*AdminClient,error)

NewAdminClient creates a new Pub/Sub Lite client to perform admin operationsfor resources within a given region.

func (*AdminClient) Close

func(ac*AdminClient)Close()error

Close releases any resources held by the client when it is no longerrequired. If the client is available for the lifetime of the program, thenClose need not be called at exit.

func (*AdminClient) CreateReservation

func(ac*AdminClient)CreateReservation(ctxcontext.Context,configReservationConfig)(*ReservationConfig,error)

CreateReservation creates a new reservation from the given config. If thereservation already exists an error will be returned.

Example

packagemainimport("context""cloud.google.com/go/pubsublite")funcmain(){ctx:=context.Background()admin,err:=pubsublite.NewAdminClient(ctx,"region")iferr!=nil{// TODO: Handle error.}deferadmin.Close()reservationConfig:=pubsublite.ReservationConfig{Name:"projects/my-project/locations/region/reservations/my-reservation",ThroughputCapacity:10,}_,err=admin.CreateReservation(ctx,reservationConfig)iferr!=nil{// TODO: Handle error.}}

func (*AdminClient) CreateSubscription

func(ac*AdminClient)CreateSubscription(ctxcontext.Context,configSubscriptionConfig,opts...CreateSubscriptionOption)(*SubscriptionConfig,error)

CreateSubscription creates a new subscription from the given config. If thesubscription already exists an error will be returned.

By default, a new subscription will only receive messages published afterthe subscription was created. Use AtTargetLocation to initialize thesubscription to another location within the message backlog.

Examples

packagemainimport("context""cloud.google.com/go/pubsublite")funcmain(){ctx:=context.Background()// NOTE: resources must be located within this region.admin,err:=pubsublite.NewAdminClient(ctx,"region")iferr!=nil{// TODO: Handle error.}deferadmin.Close()subscriptionConfig:=pubsublite.SubscriptionConfig{Name:"projects/my-project/locations/region-or-zone/subscriptions/my-subscription",Topic:"projects/my-project/locations/region-or-zone/topics/my-topic",// Do not wait for a published message to be successfully written to storage// before delivering it to subscribers.DeliveryRequirement:pubsublite.DeliverImmediately,}_,err=admin.CreateSubscription(ctx,subscriptionConfig)iferr!=nil{// TODO: Handle error.}}

atTargetLocation

packagemainimport("context""cloud.google.com/go/pubsublite")funcmain(){ctx:=context.Background()// NOTE: resources must be located within this region.admin,err:=pubsublite.NewAdminClient(ctx,"region")iferr!=nil{// TODO: Handle error.}deferadmin.Close()subscriptionConfig:=pubsublite.SubscriptionConfig{Name:"projects/my-project/locations/region-or-zone/subscriptions/my-subscription",Topic:"projects/my-project/locations/region-or-zone/topics/my-topic",// Do not wait for a published message to be successfully written to storage// before delivering it to subscribers.DeliveryRequirement:pubsublite.DeliverImmediately,}// Initialize the subscription to the oldest retained messages for each// partition.targetLocation:=pubsublite.AtTargetLocation(pubsublite.Beginning)_,err=admin.CreateSubscription(ctx,subscriptionConfig,targetLocation)iferr!=nil{// TODO: Handle error.}}

exportToPubSub

packagemainimport("context""cloud.google.com/go/pubsublite")funcmain(){ctx:=context.Background()// NOTE: resources must be located within this region.admin,err:=pubsublite.NewAdminClient(ctx,"region")iferr!=nil{// TODO: Handle error.}deferadmin.Close()subscriptionConfig:=pubsublite.SubscriptionConfig{Name:"projects/my-project/locations/region-or-zone/subscriptions/my-subscription",Topic:"projects/my-project/locations/region-or-zone/topics/my-topic",// Deliver a published message to subscribers after it has been successfully// written to storage.DeliveryRequirement:pubsublite.DeliverAfterStored,ExportConfig:&pubsublite.ExportConfig{DesiredState:pubsublite.ExportActive,// Configure an export subscription to a Pub/Sub topic.Destination:&pubsublite.PubSubDestinationConfig{Topic:"projects/my-project/topics/destination-pubsub-topic",},// Optional Lite topic to receive messages that cannot be exported to the// destination.DeadLetterTopic:"projects/my-project/locations/region-or-zone/topics/dead-letter-topic",},}_,err=admin.CreateSubscription(ctx,subscriptionConfig)iferr!=nil{// TODO: Handle error.}}

func (*AdminClient) CreateTopic

func(ac*AdminClient)CreateTopic(ctxcontext.Context,configTopicConfig)(*TopicConfig,error)

CreateTopic creates a new topic from the given config. If the topic alreadyexists an error will be returned.

Example

packagemainimport("context""cloud.google.com/go/pubsublite")funcmain(){ctx:=context.Background()// NOTE: resources must be located within this region.admin,err:=pubsublite.NewAdminClient(ctx,"region")iferr!=nil{// TODO: Handle error.}deferadmin.Close()constgib=1 <<30topicConfig:=pubsublite.TopicConfig{Name:"projects/my-project/locations/region-or-zone/topics/my-topic",PartitionCount:2,// Must be at least 1.PublishCapacityMiBPerSec:4,// Must be 4-16 MiB/s.SubscribeCapacityMiBPerSec:8,// Must be 4-32 MiB/s.PerPartitionBytes:30*gib,// Must be 30 GiB-10 TiB.// Retain messages indefinitely as long as there is available storage.RetentionDuration:pubsublite.InfiniteRetention,}_,err=admin.CreateTopic(ctx,topicConfig)iferr!=nil{// TODO: Handle error.}}

func (*AdminClient) DeleteReservation

func(ac*AdminClient)DeleteReservation(ctxcontext.Context,reservationstring)error

DeleteReservation deletes a reservation. A valid reservation path has theformat: "projects/PROJECT_ID/locations/REGION/reservations/RESERVATION_ID".

Example

packagemainimport("context""cloud.google.com/go/pubsublite")funcmain(){ctx:=context.Background()admin,err:=pubsublite.NewAdminClient(ctx,"region")iferr!=nil{// TODO: Handle error.}deferadmin.Close()constreservation="projects/my-project/locations/region/reservations/my-reservation"iferr:=admin.DeleteReservation(ctx,reservation);err!=nil{// TODO: Handle error.}}

func (*AdminClient) DeleteSubscription

func(ac*AdminClient)DeleteSubscription(ctxcontext.Context,subscriptionstring)error

DeleteSubscription deletes a subscription. A valid subscription path has theformat: "projects/PROJECT_ID/locations/LOCATION/subscriptions/SUBSCRIPTION_ID".

Example

packagemainimport("context""cloud.google.com/go/pubsublite")funcmain(){ctx:=context.Background()// NOTE: resources must be located within this region.admin,err:=pubsublite.NewAdminClient(ctx,"region")iferr!=nil{// TODO: Handle error.}deferadmin.Close()constsubscription="projects/my-project/locations/region-or-zone/subscriptions/my-subscription"iferr:=admin.DeleteSubscription(ctx,subscription);err!=nil{// TODO: Handle error.}}

func (*AdminClient) DeleteTopic

func(ac*AdminClient)DeleteTopic(ctxcontext.Context,topicstring)error

DeleteTopic deletes a topic. A valid topic path has the format:"projects/PROJECT_ID/locations/LOCATION/topics/TOPIC_ID".

Example

packagemainimport("context""cloud.google.com/go/pubsublite")funcmain(){ctx:=context.Background()// NOTE: resources must be located within this region.admin,err:=pubsublite.NewAdminClient(ctx,"region")iferr!=nil{// TODO: Handle error.}deferadmin.Close()consttopic="projects/my-project/locations/region-or-zone/topics/my-topic"iferr:=admin.DeleteTopic(ctx,topic);err!=nil{// TODO: Handle error.}}

func (*AdminClient) Reservation

func(ac*AdminClient)Reservation(ctxcontext.Context,reservationstring)(*ReservationConfig,error)

Reservation retrieves the configuration of a reservation. A valid reservationname has the format:"projects/PROJECT_ID/locations/REGION/reservations/RESERVATION_ID".

func (*AdminClient) ReservationTopics

func(ac*AdminClient)ReservationTopics(ctxcontext.Context,reservationstring)*TopicPathIterator

ReservationTopics retrieves the list of topic paths for a reservation.A valid reservation path has the format:"projects/PROJECT_ID/locations/REGION/reservations/RESERVATION_ID".

Example

packagemainimport("context""fmt""cloud.google.com/go/pubsublite""google.golang.org/api/iterator")funcmain(){ctx:=context.Background()admin,err:=pubsublite.NewAdminClient(ctx,"region")iferr!=nil{// TODO: Handle error.}deferadmin.Close()// List the paths of all topics using a reservation.constreservation="projects/my-project/locations/region/reservations/my-reservation"it:=admin.ReservationTopics(ctx,reservation)for{topicPath,err:=it.Next()iferr==iterator.Done{break}iferr!=nil{// TODO: Handle error.}fmt.Println(topicPath)}}

func (*AdminClient) Reservations

func(ac*AdminClient)Reservations(ctxcontext.Context,parentstring)*ReservationIterator

Reservations retrieves the list of reservation configs for a given projectand region. A valid parent path has the format:"projects/PROJECT_ID/locations/REGION".

Example

packagemainimport("context""fmt""cloud.google.com/go/pubsublite""google.golang.org/api/iterator")funcmain(){ctx:=context.Background()admin,err:=pubsublite.NewAdminClient(ctx,"region")iferr!=nil{// TODO: Handle error.}deferadmin.Close()// List the configs of all reservations in the given region for the project.it:=admin.Reservations(ctx,"projects/my-project/locations/region")for{reservation,err:=it.Next()iferr==iterator.Done{break}iferr!=nil{// TODO: Handle error.}fmt.Println(reservation)}}

func (*AdminClient) SeekSubscription

func(ac*AdminClient)SeekSubscription(ctxcontext.Context,subscriptionstring,targetSeekTarget,opts...SeekSubscriptionOption)(*SeekSubscriptionOperation,error)

SeekSubscription initiates an out-of-band seek for a subscription to aspecified target, which may be timestamps or named positions within themessage backlog. A valid subscription path has the format:"projects/PROJECT_ID/locations/LOCATION/subscriptions/SUBSCRIPTION_ID".

Seehttps://cloud.google.com/pubsub/lite/docs/seek for more information.

Example

packagemainimport("context""fmt""cloud.google.com/go/pubsublite")funcmain(){ctx:=context.Background()// NOTE: resources must be located within this region.admin,err:=pubsublite.NewAdminClient(ctx,"region")iferr!=nil{// TODO: Handle error.}deferadmin.Close()constsubscription="projects/my-project/locations/region-or-zone/subscriptions/my-subscription"seekOp,err:=admin.SeekSubscription(ctx,subscription,pubsublite.Beginning)iferr!=nil{// TODO: Handle error.}// Optional: Wait for the seek operation to complete, which indicates when// subscribers for all partitions are receiving messages from the seek target._,err=seekOp.Wait(ctx)iferr!=nil{// TODO: Handle error.}metadata,err:=seekOp.Metadata()iferr!=nil{// TODO: Handle error.}fmt.Println(metadata)}

func (*AdminClient) Subscription

func(ac*AdminClient)Subscription(ctxcontext.Context,subscriptionstring)(*SubscriptionConfig,error)

Subscription retrieves the configuration of a subscription. A validsubscription name has the format:"projects/PROJECT_ID/locations/LOCATION/subscriptions/SUBSCRIPTION_ID".

func (*AdminClient) Subscriptions

func(ac*AdminClient)Subscriptions(ctxcontext.Context,parentstring)*SubscriptionIterator

Subscriptions retrieves the list of subscription configs for a given projectand location (region or zone). A valid parent path has the format:"projects/PROJECT_ID/locations/LOCATION".

Example

packagemainimport("context""fmt""cloud.google.com/go/pubsublite""google.golang.org/api/iterator")funcmain(){ctx:=context.Background()// NOTE: resources must be located within this region.admin,err:=pubsublite.NewAdminClient(ctx,"region")iferr!=nil{// TODO: Handle error.}deferadmin.Close()// List the configs of all subscriptions in the given region or zone for the project.it:=admin.Subscriptions(ctx,"projects/my-project/locations/region-or-zone")for{subscriptionConfig,err:=it.Next()iferr==iterator.Done{break}iferr!=nil{// TODO: Handle error.}fmt.Println(subscriptionConfig)}}

func (*AdminClient) Topic

func(ac*AdminClient)Topic(ctxcontext.Context,topicstring)(*TopicConfig,error)

Topic retrieves the configuration of a topic. A valid topic path has theformat: "projects/PROJECT_ID/locations/LOCATION/topics/TOPIC_ID".

func (*AdminClient) TopicPartitionCount

func(ac*AdminClient)TopicPartitionCount(ctxcontext.Context,topicstring)(int,error)

TopicPartitionCount returns the number of partitions for a topic. A validtopic path has the format:"projects/PROJECT_ID/locations/LOCATION/topics/TOPIC_ID".

func (*AdminClient) TopicSubscriptions

func(ac*AdminClient)TopicSubscriptions(ctxcontext.Context,topicstring)*SubscriptionPathIterator

TopicSubscriptions retrieves the list of subscription paths for a topic.A valid topic path has the format:"projects/PROJECT_ID/locations/LOCATION/topics/TOPIC_ID".

Example

packagemainimport("context""fmt""cloud.google.com/go/pubsublite""google.golang.org/api/iterator")funcmain(){ctx:=context.Background()// NOTE: resources must be located within this region.admin,err:=pubsublite.NewAdminClient(ctx,"region")iferr!=nil{// TODO: Handle error.}deferadmin.Close()// List the paths of all subscriptions of a topic.consttopic="projects/my-project/locations/region-or-zone/topics/my-topic"it:=admin.TopicSubscriptions(ctx,topic)for{subscriptionPath,err:=it.Next()iferr==iterator.Done{break}iferr!=nil{// TODO: Handle error.}fmt.Println(subscriptionPath)}}

func (*AdminClient) Topics

func(ac*AdminClient)Topics(ctxcontext.Context,parentstring)*TopicIterator

Topics retrieves the list of topic configs for a given project and location(region or zone). A valid parent path has the format:"projects/PROJECT_ID/locations/LOCATION".

Example

packagemainimport("context""fmt""cloud.google.com/go/pubsublite""google.golang.org/api/iterator")funcmain(){ctx:=context.Background()// NOTE: resources must be located within this region.admin,err:=pubsublite.NewAdminClient(ctx,"region")iferr!=nil{// TODO: Handle error.}deferadmin.Close()// List the configs of all topics in the given region or zone for the project.it:=admin.Topics(ctx,"projects/my-project/locations/region-or-zone")for{topicConfig,err:=it.Next()iferr==iterator.Done{break}iferr!=nil{// TODO: Handle error.}fmt.Println(topicConfig)}}

func (*AdminClient) UpdateReservation

func(ac*AdminClient)UpdateReservation(ctxcontext.Context,configReservationConfigToUpdate)(*ReservationConfig,error)

UpdateReservation updates an existing reservation from the given config andreturns the new reservation config. UpdateReservation returns an error if nofields were modified.

Example

packagemainimport("context""cloud.google.com/go/pubsublite")funcmain(){ctx:=context.Background()admin,err:=pubsublite.NewAdminClient(ctx,"region")iferr!=nil{// TODO: Handle error.}deferadmin.Close()updateConfig:=pubsublite.ReservationConfigToUpdate{Name:"projects/my-project/locations/region/reservations/my-reservation",ThroughputCapacity:20,}_,err=admin.UpdateReservation(ctx,updateConfig)iferr!=nil{// TODO: Handle error.}}

func (*AdminClient) UpdateSubscription

func(ac*AdminClient)UpdateSubscription(ctxcontext.Context,configSubscriptionConfigToUpdate)(*SubscriptionConfig,error)

UpdateSubscription updates an existing subscription from the given config andreturns the new subscription config. UpdateSubscription returns an error ifno fields were modified.

Example

packagemainimport("context""cloud.google.com/go/pubsublite")funcmain(){ctx:=context.Background()// NOTE: resources must be located within this region.admin,err:=pubsublite.NewAdminClient(ctx,"region")iferr!=nil{// TODO: Handle error.}deferadmin.Close()updateConfig:=pubsublite.SubscriptionConfigToUpdate{Name:"projects/my-project/locations/region-or-zone/subscriptions/my-subscription",// Deliver a published message to subscribers after it has been successfully// written to storage.DeliveryRequirement:pubsublite.DeliverAfterStored,}_,err=admin.UpdateSubscription(ctx,updateConfig)iferr!=nil{// TODO: Handle error.}}

func (*AdminClient) UpdateTopic

func(ac*AdminClient)UpdateTopic(ctxcontext.Context,configTopicConfigToUpdate)(*TopicConfig,error)

UpdateTopic updates an existing topic from the given config and returns thenew topic config. UpdateTopic returns an error if no fields were modified.

Example

packagemainimport("context""time""cloud.google.com/go/pubsublite")funcmain(){ctx:=context.Background()// NOTE: resources must be located within this region.admin,err:=pubsublite.NewAdminClient(ctx,"region")iferr!=nil{// TODO: Handle error.}deferadmin.Close()updateConfig:=pubsublite.TopicConfigToUpdate{Name:"projects/my-project/locations/region-or-zone/topics/my-topic",PartitionCount:3,// Only increases currently supported.PublishCapacityMiBPerSec:8,SubscribeCapacityMiBPerSec:16,RetentionDuration:24*time.Hour,// Garbage collect messages older than 24 hours.}_,err=admin.UpdateTopic(ctx,updateConfig)iferr!=nil{// TODO: Handle error.}}

BacklogLocation

typeBacklogLocationint

BacklogLocation refers to a location with respect to the message backlog.It implements the SeekTarget interface.

End, Beginning

const(// End refers to the location past all currently published messages. End// skips the entire message backlog.EndBacklogLocation=iota+1// Beginning refers to the location of the oldest retained message.Beginning)

CreateSubscriptionOption

typeCreateSubscriptionOptioninterface{// contains filtered or unexported methods}

CreateSubscriptionOption is an option for AdminClient.CreateSubscription.

func AtTargetLocation

funcAtTargetLocation(targetSeekTarget)CreateSubscriptionOption

AtTargetLocation specifies the target location within the message backlogthat a new subscription should be initialized to.

An additional seek request is initiated if the target location is a publishor event time. If the seek fails, the created subscription is not deleted.

func StartingOffset

funcStartingOffset(locationBacklogLocation)CreateSubscriptionOption

StartingOffset specifies the offset at which a newly created subscriptionwill start receiving messages.

Deprecated. This is equivalent to calling AtTargetLocation with aBacklogLocation and will be removed in the next major version.

DeliveryRequirement

typeDeliveryRequirementint

DeliveryRequirement specifies when a subscription should send messages tosubscribers relative to persistence in storage.

UnspecifiedDeliveryRequirement, DeliverImmediately, DeliverAfterStored

const(// UnspecifiedDeliveryRequirement represents an unset delivery requirement.UnspecifiedDeliveryRequirementDeliveryRequirement=iota// DeliverImmediately means the server will not wait for a published message// to be successfully written to storage before delivering it to subscribers.DeliverImmediately// DeliverAfterStored means the server will not deliver a published message to// subscribers until the message has been successfully written to storage.// This will result in higher end-to-end latency, but consistent delivery.DeliverAfterStored)

EventTime

typeEventTimetime.Time

EventTime is a message event timestamp. It implements the SeekTargetinterface.

ExportConfig

typeExportConfigstruct{// The desired state of this export subscription. This should only be set to// ExportActive or ExportPaused.DesiredStateExportState// This is an output only field that reports the current export state. It is// ignored if set in any requests.CurrentStateExportState// The path of an optional Pub/Sub Lite topic to receive messages that cannot// be exported to the destination, in the format:// "projects/PROJECT_ID/locations/LOCATION/topics/TOPIC_ID".// Must be within the same project and location as the subscription.DeadLetterTopicstring// The destination to export messages to.DestinationExportDestinationConfig}

ExportConfig describes the properties of a Pub/Sub Lite export subscription,which configures the service to write messages to a destination.

ExportConfigToUpdate

typeExportConfigToUpdatestruct{// If non-zero, updates the desired state. This should only be set to// ExportActive or ExportPaused.DesiredStateExportState// The path of an optional Pub/Sub Lite topic to receive messages that cannot// be exported to the destination, in the format:// "projects/PROJECT_ID/locations/LOCATION/topics/TOPIC_ID".// Must be within the same project and location as the subscription.DeadLetterTopicoptional.String// If non-nil, updates the export destination configuration.DestinationExportDestinationConfig}

ExportConfigToUpdate specifies the properties to update for an exportsubscription.

ExportDestinationConfig

typeExportDestinationConfiginterface{// contains filtered or unexported methods}

ExportDestinationConfig is the configuration for exporting to a destination.Implemented by *PubSubDestinationConfig.

ExportState

typeExportStateint

ExportState specifies the desired state of an export subscription.

UnspecifiedExportState, ExportActive, ExportPaused, ExportPermissionDenied, ExportResourceNotFound

const(// UnspecifiedExportState represents an unset export state.UnspecifiedExportStateExportState=iota// ExportActive specifies that export processing should be enabled.ExportActive// ExportPaused specifies that export processing should be suspended.ExportPaused// ExportPermissionDenied specifies that messages cannot be exported due to// permission denied errors. Output only.ExportPermissionDenied// ExportResourceNotFound specifies that messages cannot be exported due to// missing resources. Output only.ExportResourceNotFound)

OperationMetadata

typeOperationMetadatastruct{// The target of the operation. For example, targets of seeks are// subscriptions, structured like:// "projects/PROJECT_ID/locations/LOCATION/subscriptions/SUBSCRIPTION_ID"Targetstring// The verb describing the kind of operation.Verbstring// The time the operation was created.CreateTimetime.Time// The time the operation finished running. Is zero if the operation has not// completed.EndTimetime.Time}

OperationMetadata stores metadata for long-running operations.

PubSubDestinationConfig

typePubSubDestinationConfigstruct{// The path of a Pub/Sub topic, in the format:// "projects/PROJECT_ID/topics/TOPIC_ID".Topicstring}

PubSubDestinationConfig configures messages to be exported to a Pub/Subtopic. Implements the ExportDestinationConfig interface.

Seehttps://cloud.google.com/pubsub/lite/docs/export-pubsub for moreinformation about how export subscriptions to Pub/Sub are configured.

PublishTime

typePublishTimetime.Time

PublishTime is a message publish timestamp. It implements the SeekTargetinterface.

ReservationConfig

typeReservationConfigstruct{// The full path of the reservation, in the format:// "projects/PROJECT_ID/locations/REGION/reservations/RESERVATION_ID".//// - PROJECT_ID: The project ID (e.g. "my-project") or the project number//   (e.g. "987654321") can be provided.// - REGION: The Google Cloud region (e.g. "us-central1") for the reservation.//   See https://cloud.google.com/pubsub/lite/docs/locations for the list of//   regions where Pub/Sub Lite is available.// - RESERVATION_ID: The ID of the reservation (e.g. "my-reservation"). See//   https://cloud.google.com/pubsub/docs/admin#resource_names for information//   about valid reservation IDs.Namestring// The reserved throughput capacity. Every unit of throughput capacity is// equivalent to 1 MiB/s of published messages or 2 MiB/s of subscribed// messages.//// Any topics which are declared as using capacity from a reservation will// consume resources from this reservation instead of being charged// individually.ThroughputCapacityint}

ReservationConfig describes the properties of a Pub/Sub Lite reservation.

ReservationConfigToUpdate

typeReservationConfigToUpdatestruct{// The full path of the reservation to update, in the format:// "projects/PROJECT_ID/locations/REGION/reservations/RESERVATION_ID".// Required.Namestring// If non-zero, updates the throughput capacity.ThroughputCapacityint}

ReservationConfigToUpdate specifies the properties to update for areservation.

ReservationIterator

typeReservationIteratorstruct{// contains filtered or unexported fields}

ReservationIterator is an iterator that returns a list of reservationconfigs.

func (*ReservationIterator) Next

func(r*ReservationIterator)Next()(*ReservationConfig,error)

Next returns the next reservation config. The second return value will beiterator.Done if there are no more reservation configs.

SeekSubscriptionOperation

typeSeekSubscriptionOperationstruct{// contains filtered or unexported fields}

SeekSubscriptionOperation manages a long-running seek operation fromAdminClient.SeekSubscription.

func (*SeekSubscriptionOperation) Done

func(s*SeekSubscriptionOperation)Done()bool

Done returns whether the seek operation has completed.

func (*SeekSubscriptionOperation) Metadata

func(s*SeekSubscriptionOperation)Metadata()(*OperationMetadata,error)

Metadata returns metadata associated with the seek operation. To get thelatest metadata, call this method after a successful call to Wait.

func (*SeekSubscriptionOperation) Name

func(s*SeekSubscriptionOperation)Name()string

Name returns the path of the seek operation, in the format:"projects/PROJECT_ID/locations/LOCATION/operations/OPERATION_ID".

func (*SeekSubscriptionOperation) Wait

func(s*SeekSubscriptionOperation)Wait(ctxcontext.Context)(*SeekSubscriptionResult,error)

Wait polls until the seek operation is complete and returns one of thefollowing:

A SeekSubscriptionResult and nil error if the operation is complete andsucceeded.
Error containing failure reason if the operation is complete and failed.
Error if polling the operation status failed due to a non-retryable error.

SeekSubscriptionOption

typeSeekSubscriptionOptioninterface{}

SeekSubscriptionOption is reserved for future options.

SeekSubscriptionResult

typeSeekSubscriptionResultstruct{}

SeekSubscriptionResult is the result of a seek subscription operation.Currently empty.

SeekTarget

typeSeekTargetinterface{// contains filtered or unexported methods}

SeekTarget is the target location to seek a subscription to. Implemented byBacklogLocation, PublishTime, EventTime.

SubscriptionConfig

typeSubscriptionConfigstruct{// The full path of the subscription, in the format:// "projects/PROJECT_ID/locations/LOCATION/subscriptions/SUBSCRIPTION_ID".//// - PROJECT_ID: The project ID (e.g. "my-project") or the project number//   (e.g. "987654321") can be provided.// - LOCATION: The Google Cloud region (e.g. "us-central1") or zone//   (e.g. "us-central1-a") of the corresponding topic.// - SUBSCRIPTION_ID: The ID of the subscription (e.g. "my-subscription"). See//   https://cloud.google.com/pubsub/docs/admin#resource_names for information//   about valid subscription IDs.Namestring// The path of the topic that this subscription is attached to, in the format:// "projects/PROJECT_ID/locations/LOCATION/topics/TOPIC_ID". This cannot be// changed after creation.Topicstring// Whether a message should be delivered to subscribers immediately after it// has been published or after it has been successfully written to storage.DeliveryRequirementDeliveryRequirement// If non-nil, configures this subscription to export messages from the// associated topic to a destination. The ExportConfig cannot be removed after// creation of the subscription, however its properties can be changed.ExportConfig*ExportConfig}

SubscriptionConfig describes the properties of a Pub/Sub Lite subscription,which is attached to a Pub/Sub Lite topic.Seehttps://cloud.google.com/pubsub/lite/docs/subscriptions for moreinformation about how subscriptions are configured.

SubscriptionConfigToUpdate

typeSubscriptionConfigToUpdatestruct{// The full path of the subscription to update, in the format:// "projects/PROJECT_ID/locations/LOCATION/subscriptions/SUBSCRIPTION_ID".// Required.Namestring// If non-zero, updates the message delivery requirement.DeliveryRequirementDeliveryRequirement// If non-nil, updates export config properties.ExportConfig*ExportConfigToUpdate}

SubscriptionConfigToUpdate specifies the properties to update for asubscription.

SubscriptionIterator

typeSubscriptionIteratorstruct{// contains filtered or unexported fields}

SubscriptionIterator is an iterator that returns a list of subscriptionconfigs.

func (*SubscriptionIterator) Next

func(s*SubscriptionIterator)Next()(*SubscriptionConfig,error)

Next returns the next subscription config. The second return value will beiterator.Done if there are no more subscription configs.

SubscriptionPathIterator

typeSubscriptionPathIteratorstruct{// contains filtered or unexported fields}

SubscriptionPathIterator is an iterator that returns a list of subscriptionpaths.

func (*SubscriptionPathIterator) Next

func(sp*SubscriptionPathIterator)Next()(string,error)

Next returns the next subscription path, which has format:"projects/PROJECT_ID/locations/LOCATION/subscriptions/SUBSCRIPTION_ID". Thesecond return value will be iterator.Done if there are no more subscriptionpaths.

TopicConfig

typeTopicConfigstruct{// The full path of the topic, in the format:// "projects/PROJECT_ID/locations/LOCATION/topics/TOPIC_ID".//// - PROJECT_ID: The project ID (e.g. "my-project") or the project number//   (e.g. "987654321") can be provided.// - LOCATION: The Google Cloud region (e.g. "us-central1") or zone//   (e.g. "us-central1-a") where the topic is located.//   See https://cloud.google.com/pubsub/lite/docs/locations for the list of//   regions and zones where Pub/Sub Lite is available.// - TOPIC_ID: The ID of the topic (e.g. "my-topic"). See//   https://cloud.google.com/pubsub/docs/admin#resource_names for information//   about valid topic IDs.Namestring// The number of partitions in the topic. Must be at least 1. Can be increased// after creation, but not decreased.PartitionCountint// Publish throughput capacity per partition in MiB/s.// Must be >= 4 and <= 16.="" publishcapacitymibpersec="">int// Subscribe throughput capacity per partition in MiB/s.// Must be >= 4 and <= 32.="" subscribecapacitymibpersec="">int// The provisioned storage, in bytes, per partition. If the number of bytes// stored in any of the topic's partitions grows beyond this value, older// messages will be dropped to make room for newer ones, regardless of the// value of `RetentionDuration`. Must be >= 30 GiB.PerPartitionBytesint64// How long a published message is retained. If set to `InfiniteRetention`,// messages will be retained as long as the bytes retained for each partition// is below `PerPartitionBytes`. Otherwise, must be > 0.RetentionDurationtime.Duration// The path of the reservation to use for this topic's throughput capacity, in// the format:// "projects/PROJECT_ID/locations/REGION/reservations/RESERVATION_ID".ThroughputReservationstring}

TopicConfig describes the properties of a Pub/Sub Lite topic.Seehttps://cloud.google.com/pubsub/lite/docs/topics for more informationabout how topics are configured.

TopicConfigToUpdate

typeTopicConfigToUpdatestruct{// The full path of the topic to update, in the format:// "projects/PROJECT_ID/locations/LOCATION/topics/TOPIC_ID". Required.Namestring// If non-zero, will update the number of partitions in the topic.// Set value must be >= 1. The number of partitions can only be increased, not// decreased.PartitionCountint// If non-zero, will update the publish throughput capacity per partition.// Set value must be >= 4 and <= 16.="" publishcapacitymibpersec="">int// If non-zero, will update the subscribe throughput capacity per partition.// Set value must be >= 4 and <= 32.="" subscribecapacitymibpersec="">int// If non-zero, will update the provisioned storage per partition.// Set value must be >= 30 GiB.PerPartitionBytesint64// If specified, will update how long a published message is retained. To// clear a retention duration (i.e. retain messages as long as there is// available storage), set this to `InfiniteRetention`.RetentionDurationoptional.Duration// The path of the reservation to use for this topic's throughput capacity, in// the format:// "projects/PROJECT_ID/locations/REGION/reservations/RESERVATION_ID".ThroughputReservationoptional.String}

TopicConfigToUpdate specifies the properties to update for a topic.

TopicIterator

typeTopicIteratorstruct{// contains filtered or unexported fields}

TopicIterator is an iterator that returns a list of topic configs.

func (*TopicIterator) Next

func(t*TopicIterator)Next()(*TopicConfig,error)

Next returns the next topic config. The second return value will beiterator.Done if there are no more topic configs.

TopicPathIterator

typeTopicPathIteratorstruct{// contains filtered or unexported fields}

TopicPathIterator is an iterator that returns a list of topic paths.

func (*TopicPathIterator) Next

func(sp*TopicPathIterator)Next()(string,error)

Next returns the next topic path, which has format:"projects/PROJECT_ID/locations/LOCATION/topics/TOPIC_ID". The second returnvalue will be iterator.Done if there are no more topic paths.

Except as otherwise noted, the content of this page is licensed under theCreative Commons Attribution 4.0 License, and code samples are licensed under theApache 2.0 License. For details, see theGoogle Developers Site Policies. Java is a registered trademark of Oracle and/or its affiliates.

Last updated 2025-10-30 UTC.

Movatterモバイル変換

Cloud PubSub Lite - Package cloud.google.com/go/pubsublite (v1.8.2) Stay organized with collections Save and categorize content based on your preferences.

Introduction

Creating Topics

Publishing

Creating Subscriptions

Receiving

gRPC Connection Pools

Constants

InfiniteRetention

AdminClient

func NewAdminClient

func (*AdminClient) Close

func (*AdminClient) CreateReservation

Example

func (*AdminClient) CreateSubscription

Examples

atTargetLocation

exportToPubSub

func (*AdminClient) CreateTopic

Example

func (*AdminClient) DeleteReservation

Example

func (*AdminClient) DeleteSubscription

Example

func (*AdminClient) DeleteTopic

Example

func (*AdminClient) Reservation

func (*AdminClient) ReservationTopics

Example

func (*AdminClient) Reservations

Example

func (*AdminClient) SeekSubscription

Example

func (*AdminClient) Subscription

func (*AdminClient) Subscriptions

Example

func (*AdminClient) Topic

func (*AdminClient) TopicPartitionCount

func (*AdminClient) TopicSubscriptions

Example

func (*AdminClient) Topics

Example

func (*AdminClient) UpdateReservation

Example

func (*AdminClient) UpdateSubscription

Example

func (*AdminClient) UpdateTopic

Example

BacklogLocation

End, Beginning

CreateSubscriptionOption

func AtTargetLocation

func StartingOffset

DeliveryRequirement

UnspecifiedDeliveryRequirement, DeliverImmediately, DeliverAfterStored

EventTime

ExportConfig

ExportConfigToUpdate

ExportDestinationConfig

ExportState

UnspecifiedExportState, ExportActive, ExportPaused, ExportPermissionDenied, ExportResourceNotFound

OperationMetadata

PubSubDestinationConfig

PublishTime

ReservationConfig

ReservationConfigToUpdate

ReservationIterator

func (*ReservationIterator) Next

SeekSubscriptionOperation

func (*SeekSubscriptionOperation) Done

func (*SeekSubscriptionOperation) Metadata

func (*SeekSubscriptionOperation) Name

func (*SeekSubscriptionOperation) Wait

SeekSubscriptionOption

SeekSubscriptionResult

SeekTarget

SubscriptionConfig

SubscriptionConfigToUpdate

SubscriptionIterator

Cloud PubSub Lite - Package cloud.google.com/go/pubsublite (v1.8.2)