Package pubsublite provides an easy way to publish and receive messages using Google Cloud Pub/Sub Lite.
Google Cloud Pub/Sub is designed to provide reliable, many-to-many, asynchronous messaging between applications. Publisher applications can send messages to a topic and other applications can subscribe to that topic to receive the messages. By decoupling senders and receivers, Google Cloud Pub/Sub allows developers to communicate between independently written applications.
Compared to Google Cloud Pub/Sub, Pub/Sub Lite provides partitioned zonal data storage with predefined throughput and storage capacity. Guidance on how to choose between Google Cloud Pub/Sub and Pub/Sub Lite is available at https://cloud.google.com/pubsub/docs/choosing-pubsub-or-lite.
More information about Google Cloud Pub/Sub Lite is available at https://cloud.google.com/pubsub/lite.
See https://godoc.org/cloud.google.com/go for authentication, timeouts, connection pooling and similar aspects of this package.
Note: This library is in ALPHA. Backwards-incompatible changes may be made before stable v1.0.0 is released.
Creating Topics
Messages are published to topics. Cloud Pub/Sub Lite topics may be created like so (error handling omitted for brevity):
topicPath := pubsublite.TopicPath{
Project: "project-id",
Zone: "zone",
TopicID: "topic-id",
}
topicConfig := pubsublite.TopicConfig{
Name: topicPath,
PartitionCount: 1,
PublishCapacityMiBPerSec: 4,
SubscribeCapacityMiBPerSec: 4,
PerPartitionBytes: 30 * 1024 * 1024 * 1024, // 30 GiB
RetentionDuration: pubsublite.InfiniteRetention,
}
region, err := pubsublite.ZoneToRegion(topicPath.Zone)
adminClient, err := pubsublite.NewAdminClient(ctx, region)
_, err = adminClient.CreateTopic(ctx, topicConfig)
See https://cloud.google.com/pubsub/lite/docs/topics for more information about how Cloud Pub/Sub Lite topics are configured.
Publishing
The pubsublite/ps subpackage contains clients for publishing and receiving messages, which have similar interfaces to their Topic and Subscription counterparts in the pubsub package. Cloud Pub/Sub Lite uses gRPC streams extensively for high throughput. For more differences, see https://godoc.org/cloud.google.com/go/pubsublite/ps.
To publish messages to a topic, first create a PublisherClient:
publisher, err := ps.NewPublisherClient(ctx, ps.DefaultPublishSettings, topicPath)
Then call Publish:
// Note: result is a pubsub.PublishResult
result := publisher.Publish(ctx, &pubsub.Message{Data: []byte("payload")})
Publish queues the message for publishing and returns immediately. When enough messages have accumulated, or enough time has elapsed, the batch of messages is sent to the Cloud Pub/Sub Lite service. Thresholds for batching can be configured in PublishSettings.
Publish returns a PublishResult, which behaves like a future: its Get method blocks until the message has been sent (or has failed to be sent) to the service. Once you've finishing publishing, call Stop to flush all messages to the service and close gRPC streams:
publisher.Stop()
See https://cloud.google.com/pubsub/lite/docs/publishing for more information about publishing.
Creating Subscriptions
To receive messages published to a topic, clients create subscriptions to the topic. There may be more than one subscription per topic; each message that is published to the topic will be delivered to all of its subscriptions.
Cloud Pub/Sub Lite subscriptions may be created like so:
subscriptionPath := pubsublite.SubscriptionPath{
Project: "project-id",
Zone: "zone",
SubscriptionID: "subscription-id",
}
subscriptionConfig := pubsublite.SubscriptionConfig{
Name: subscriptionPath,
Topic: topicPath,
DeliveryRequirement: pubsublite.DeliverImmediately,
}
_, err = admin.CreateSubscription(ctx, subscriptionConfig)
See https://cloud.google.com/pubsub/lite/docs/subscriptions for more information about how subscriptions are configured.
Receiving
To receive messages for a subscription, first create a SubscriberClient:
subscriber, err := ps.NewSubscriberClient(ctx, ps.DefaultReceiveSettings, subscriptionPath)
Messages are then consumed from a subscription via callback.
cctx, cancel := context.WithCancel(ctx)
err = subscriber.Receive(cctx, func(ctx context.Context, m *pubsub.Message) {
log.Printf("Got message: %s", m.Data)
m.Ack()
})
if err != nil {
// Handle error.
}
The callback may be invoked concurrently by multiple goroutines (one per partition that the subscriber client is connected to). To terminate a call to Receive, cancel its context:
cancel()
Clients must call pubsub.Message.Ack() or pubsub.Message.Nack() for every message received. Cloud Pub/Sub Lite does not have ACK deadlines. Cloud Pub/Sub Lite also does not actually have the concept of NACK. The default behavior terminates the SubscriberClient. In Cloud 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.
See https://cloud.google.com/pubsub/lite/docs/subscribing for more information about receiving messages.
Constants
UnspecifiedDeliveryRequirement, DeliverImmediately, DeliverAfterStored
const (
// UnspecifiedDeliveryRequirement represents and unset delivery requirement.
UnspecifiedDeliveryRequirement = DeliveryRequirement(pb.Subscription_DeliveryConfig_DELIVERY_REQUIREMENT_UNSPECIFIED)
// DeliverImmediately means the server will not not wait for a published
// message to be successfully written to storage before delivering it to
// subscribers.
DeliverImmediately = DeliveryRequirement(pb.Subscription_DeliveryConfig_DELIVER_IMMEDIATELY)
// 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 = DeliveryRequirement(pb.Subscription_DeliveryConfig_DELIVER_AFTER_STORED)
)InfiniteRetention
InfiniteRetention is a sentinel used in topic configs to denote an infinite retention duration (i.e. retain messages as long as there is available storage).
Functions
func ZoneToRegion
ZoneToRegion returns the region that the given zone is in.
AdminClient
type AdminClient struct {
// contains filtered or unexported fields
}AdminClient provides admin operations for Cloud Pub/Sub Lite resources within a Google Cloud region. An AdminClient may be shared by multiple goroutines.
func NewAdminClient
func NewAdminClient(ctx context.Context, region string, opts ...option.ClientOption) (*AdminClient, error)NewAdminClient creates a new Cloud Pub/Sub Lite client to perform admin operations for resources within a given region. See https://cloud.google.com/pubsub/lite/docs/locations for the list of regions and zones where Cloud Pub/Sub Lite is available.
func (*AdminClient) Close
func (ac *AdminClient) Close() errorClose releases any resources held by the client when it is no longer required. If the client is available for the lifetime of the program, then Close need not be called at exit.
func (*AdminClient) CreateSubscription
func (ac *AdminClient) CreateSubscription(ctx context.Context, config SubscriptionConfig) (*SubscriptionConfig, error)CreateSubscription creates a new subscription from the given config. If the subscription already exists an error will be returned.
Example
package main
import (
"context"
"cloud.google.com/go/pubsublite"
)
func main() {
ctx := context.Background()
admin, err := pubsublite.NewAdminClient(ctx, "region")
if err != nil {
// TODO: Handle error.
}
subscriptionConfig := pubsublite.SubscriptionConfig{
Name: pubsublite.SubscriptionPath{
Project: "project-id",
Zone: "zone",
SubscriptionID: "subscription-id",
},
Topic: pubsublite.TopicPath{
Project: "project-id",
Zone: "zone",
TopicID: "topic-id",
},
// 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)
if err != nil {
// TODO: Handle error.
}
}
func (*AdminClient) CreateTopic
func (ac *AdminClient) CreateTopic(ctx context.Context, config TopicConfig) (*TopicConfig, error)CreateTopic creates a new topic from the given config. If the topic already exists an error will be returned.
Example
package main
import (
"context"
"cloud.google.com/go/pubsublite"
)
func main() {
ctx := context.Background()
admin, err := pubsublite.NewAdminClient(ctx, "region")
if err != nil {
// TODO: Handle error.
}
const gib = 1 << 30
topicConfig := pubsublite.TopicConfig{
Name: pubsublite.TopicPath{
Project: "project-id",
Zone: "zone",
TopicID: "topic-id",
},
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)
if err != nil {
// TODO: Handle error.
}
}
func (*AdminClient) DeleteSubscription
func (ac *AdminClient) DeleteSubscription(ctx context.Context, subscription SubscriptionPath) errorDeleteSubscription deletes a subscription.
Example
package main
import (
"context"
"cloud.google.com/go/pubsublite"
)
func main() {
ctx := context.Background()
admin, err := pubsublite.NewAdminClient(ctx, "region")
if err != nil {
// TODO: Handle error.
}
subscription := pubsublite.SubscriptionPath{
Project: "project-id",
Zone: "zone",
SubscriptionID: "subscription-id",
}
if err := admin.DeleteSubscription(ctx, subscription); err != nil {
// TODO: Handle error.
}
}
func (*AdminClient) DeleteTopic
func (ac *AdminClient) DeleteTopic(ctx context.Context, topic TopicPath) errorDeleteTopic deletes a topic.
Example
package main
import (
"context"
"cloud.google.com/go/pubsublite"
)
func main() {
ctx := context.Background()
admin, err := pubsublite.NewAdminClient(ctx, "region")
if err != nil {
// TODO: Handle error.
}
topic := pubsublite.TopicPath{
Project: "project-id",
Zone: "zone",
TopicID: "topic-id",
}
if err := admin.DeleteTopic(ctx, topic); err != nil {
// TODO: Handle error.
}
}
func (*AdminClient) Subscription
func (ac *AdminClient) Subscription(ctx context.Context, subscription SubscriptionPath) (*SubscriptionConfig, error)Subscription retrieves the configuration of a subscription.
func (*AdminClient) Subscriptions
func (ac *AdminClient) Subscriptions(ctx context.Context, location LocationPath) *SubscriptionIteratorSubscriptions retrieves the list of subscription configs for a given project and zone.
Example
package main
import (
"context"
"fmt"
"cloud.google.com/go/pubsublite"
"google.golang.org/api/iterator"
)
func main() {
ctx := context.Background()
admin, err := pubsublite.NewAdminClient(ctx, "region")
if err != nil {
// TODO: Handle error.
}
// List the configs of all subscriptions in the given zone for the project.
location := pubsublite.LocationPath{Project: "project-id", Zone: "zone"}
it := admin.Subscriptions(ctx, location)
for {
subscriptionConfig, err := it.Next()
if err == iterator.Done {
break
}
if err != nil {
// TODO: Handle error.
}
fmt.Println(subscriptionConfig)
}
}
func (*AdminClient) Topic
func (ac *AdminClient) Topic(ctx context.Context, topic TopicPath) (*TopicConfig, error)Topic retrieves the configuration of a topic.
func (*AdminClient) TopicPartitions
TopicPartitions returns the number of partitions for a topic.
func (*AdminClient) TopicSubscriptions
func (ac *AdminClient) TopicSubscriptions(ctx context.Context, topic TopicPath) *SubscriptionPathIteratorTopicSubscriptions retrieves the list of subscription paths for a topic.
Example
package main
import (
"context"
"fmt"
"cloud.google.com/go/pubsublite"
"google.golang.org/api/iterator"
)
func main() {
ctx := context.Background()
admin, err := pubsublite.NewAdminClient(ctx, "region")
if err != nil {
// TODO: Handle error.
}
// List the paths of all subscriptions of a topic.
topic := pubsublite.TopicPath{
Project: "project-id",
Zone: "zone",
TopicID: "topic-id",
}
it := admin.TopicSubscriptions(ctx, topic)
for {
subscriptionPath, err := it.Next()
if err == iterator.Done {
break
}
if err != nil {
// TODO: Handle error.
}
fmt.Println(subscriptionPath)
}
}
func (*AdminClient) Topics
func (ac *AdminClient) Topics(ctx context.Context, location LocationPath) *TopicIteratorTopics retrieves the list of topic configs for a given project and zone.
Example
package main
import (
"context"
"fmt"
"cloud.google.com/go/pubsublite"
"google.golang.org/api/iterator"
)
func main() {
ctx := context.Background()
admin, err := pubsublite.NewAdminClient(ctx, "region")
if err != nil {
// TODO: Handle error.
}
// List the configs of all topics in the given zone for the project.
location := pubsublite.LocationPath{Project: "project-id", Zone: "zone"}
it := admin.Topics(ctx, location)
for {
topicConfig, err := it.Next()
if err == iterator.Done {
break
}
if err != nil {
// TODO: Handle error.
}
fmt.Println(topicConfig)
}
}
func (*AdminClient) UpdateSubscription
func (ac *AdminClient) UpdateSubscription(ctx context.Context, config SubscriptionConfigToUpdate) (*SubscriptionConfig, error)UpdateSubscription updates an existing subscription from the given config and returns the new subscription config. UpdateSubscription returns an error if no fields were modified.
Example
package main
import (
"context"
"cloud.google.com/go/pubsublite"
)
func main() {
ctx := context.Background()
admin, err := pubsublite.NewAdminClient(ctx, "region")
if err != nil {
// TODO: Handle error.
}
updateConfig := pubsublite.SubscriptionConfigToUpdate{
Name: pubsublite.SubscriptionPath{
Project: "project-id",
Zone: "zone",
SubscriptionID: "subscription-id",
},
// Deliver a published message to subscribers after it has been successfully
// written to storage.
DeliveryRequirement: pubsublite.DeliverAfterStored,
}
_, err = admin.UpdateSubscription(ctx, updateConfig)
if err != nil {
// TODO: Handle error.
}
}
func (*AdminClient) UpdateTopic
func (ac *AdminClient) UpdateTopic(ctx context.Context, config TopicConfigToUpdate) (*TopicConfig, error)UpdateTopic updates an existing topic from the given config and returns the new topic config. UpdateTopic returns an error if no fields were modified.
Example
package main
import (
"context"
"time"
"cloud.google.com/go/pubsublite"
)
func main() {
ctx := context.Background()
admin, err := pubsublite.NewAdminClient(ctx, "region")
if err != nil {
// TODO: Handle error.
}
updateConfig := pubsublite.TopicConfigToUpdate{
Name: pubsublite.TopicPath{
Project: "project-id",
Zone: "zone",
TopicID: "topic-id",
},
PublishCapacityMiBPerSec: 8,
SubscribeCapacityMiBPerSec: 16,
// Garbage collect messages older than 24 hours.
RetentionDuration: 24 * time.Hour,
}
_, err = admin.UpdateTopic(ctx, updateConfig)
if err != nil {
// TODO: Handle error.
}
}
DeliveryRequirement
type DeliveryRequirement int32DeliveryRequirement specifies when a subscription should send messages to subscribers relative to persistence in storage.
LocationPath
type LocationPath struct {
// A Google Cloud project. The project ID (e.g. "my-project") or the project
// number (e.g. "987654321") can be provided.
Project string
// A Google Cloud zone, for example "us-central1-a".
// See https://cloud.google.com/pubsub/lite/docs/locations for the list of
// zones where Cloud Pub/Sub Lite is available.
Zone string
}LocationPath stores a path consisting of a project and zone.
func (LocationPath) String
func (l LocationPath) String() stringSubscriptionConfig
type SubscriptionConfig struct {
// The full path of a subscription.
Name SubscriptionPath
// The name of the topic this subscription is attached to. This cannot be
// changed after creation.
Topic TopicPath
// Whether a message should be delivered to subscribers immediately after it
// has been published or after it has been successfully written to storage.
DeliveryRequirement DeliveryRequirement
}SubscriptionConfig describes the properties of a Google Pub/Sub Lite subscription, which is attached to a topic. See https://cloud.google.com/pubsub/lite/docs/subscriptions for more information about how subscriptions are configured.
SubscriptionConfigToUpdate
type SubscriptionConfigToUpdate struct {
// The full path of the subscription to update. Required.
Name SubscriptionPath
// If non-zero, updates the message delivery requirement.
DeliveryRequirement DeliveryRequirement
}SubscriptionConfigToUpdate specifies the properties to update for a subscription.
SubscriptionIterator
type SubscriptionIterator struct {
// contains filtered or unexported fields
}SubscriptionIterator is an iterator that returns a list of subscription configs.
func (*SubscriptionIterator) Next
func (s *SubscriptionIterator) Next() (*SubscriptionConfig, error)Next returns the next subscription config. The second return value will be iterator.Done if there are no more subscription configs.
SubscriptionPath
type SubscriptionPath struct {
// A Google Cloud project. The project ID (e.g. "my-project") or the project
// number (e.g. "987654321") can be provided.
Project string
// A Google Cloud zone. An example zone is "us-central1-a".
// See https://cloud.google.com/pubsub/lite/docs/locations for the list of
// zones where Cloud Pub/Sub Lite is available.
Zone string
// The ID of the Cloud Pub/Sub Lite subscription, for example
// "my-subscription-name".
// See https://cloud.google.com/pubsub/docs/admin#resource_names for more
// information.
SubscriptionID string
}SubscriptionPath stores the full path of a Cloud Pub/Sub Lite subscription. See https://cloud.google.com/pubsub/lite/docs/subscriptions for more information.
func (SubscriptionPath) String
func (s SubscriptionPath) String() stringSubscriptionPathIterator
type SubscriptionPathIterator struct {
// contains filtered or unexported fields
}SubscriptionPathIterator is an iterator that returns a list of subscription paths.
func (*SubscriptionPathIterator) Next
func (sp *SubscriptionPathIterator) Next() (SubscriptionPath, error)Next returns the next subscription path. The second return value will be iterator.Done if there are no more subscription paths.
TopicConfig
type TopicConfig struct {
// The full path of a topic.
Name TopicPath
// The number of partitions in the topic. Must be at least 1. Cannot be
// changed after creation.
PartitionCount int
// 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 > 0.
PerPartitionBytes int64
// 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.
RetentionDuration time.Duration
}TopicConfig describes the properties of a Google Pub/Sub Lite topic. See https://cloud.google.com/pubsub/lite/docs/topics for more information about how topics are configured.
TopicConfigToUpdate
type TopicConfigToUpdate struct {
// The full path of the topic to update. Required.
Name TopicPath
// If non-zero, will update the publish throughput capacity per partition.
PublishCapacityMiBPerSec int
// If non-zero, will update the subscribe throughput capacity per partition.
SubscribeCapacityMiBPerSec int
// If non-zero, will update the provisioned storage per partition.
PerPartitionBytes int64
// 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`.
RetentionDuration optional.Duration
}TopicConfigToUpdate specifies the properties to update for a topic.
TopicIterator
type TopicIterator struct {
// 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 be iterator.Done if there are no more topic configs.
TopicPath
type TopicPath struct {
// A Google Cloud project. The project ID (e.g. "my-project") or the project
// number (e.g. "987654321") can be provided.
Project string
// A Google Cloud zone, for example "us-central1-a".
// See https://cloud.google.com/pubsub/lite/docs/locations for the list of
// zones where Cloud Pub/Sub Lite is available.
Zone string
// The ID of the Cloud Pub/Sub Lite topic, for example "my-topic-name".
// See https://cloud.google.com/pubsub/docs/admin#resource_names for more
// information.
TopicID string
}TopicPath stores the full path of a Cloud Pub/Sub Lite topic. See https://cloud.google.com/pubsub/lite/docs/topics for more information.