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
()
error
Close 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
)
error
DeleteSubscription 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
)
error
DeleteTopic 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
)
*
SubscriptionIterator
Subscriptions 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
func
(
ac
*
AdminClient
)
TopicPartitions
(
ctx
context
.
Context
,
topic
TopicPath
)
(
int
,
error
)
TopicPartitions returns the number of partitions for a topic.
func (*AdminClient) TopicSubscriptions
func
(
ac
*
AdminClient
)
TopicSubscriptions
(
ctx
context
.
Context
,
topic
TopicPath
)
*
SubscriptionPathIterator
TopicSubscriptions 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
)
*
TopicIterator
Topics 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
int32
DeliveryRequirement 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
()
string
SubscriptionConfig
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
()
string
SubscriptionPathIterator
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.
