package pubsub

import "go.chromium.org/luci/common/gcloud/pubsub"

Index ¶

• Constants
• Variables
• type ClientFactory
• type Publisher
• type Subscription
• func NewSubscription(project, name string) Subscription
• func (s *Subscription) Set(value string) error
• func (s Subscription) Split() (p, n string)
• func (s Subscription) SplitErr() (p, n string, err error)
• func (s *Subscription) String() string
• func (s Subscription) Validate() error
• type Topic
• func NewTopic(project, name string) Topic
• func (t *Topic) Set(value string) error
• func (t Topic) Split() (p, n string)
• func (t Topic) SplitErr() (p, n string, err error)
• func (t *Topic) String() string
• func (t Topic) Validate() error
• type UnbufferedPublisher
• func (up *UnbufferedPublisher) Close() error
• func (up *UnbufferedPublisher) Publish(c context.Context, msgs ...*pubsub.Message) ([]string, error)

Package Files ¶

publisher.go quota.go resource.go scopes.go subscription.go topic.go

Constants ¶

const (
    // MaxPublishRequestBytes is the maximum size of a single publish request in
    // bytes, as determined by the PubSub service.
    //
    // See: https://cloud.google.com/pubsub/publisher
    MaxPublishRequestBytes = 1e7

    // MaxPublishRequestCount is the maximum PubSub batch size.
    MaxPublishRequestCount = pubsub.MaxPublishRequestCount

    // MaxProjectMessagesPerSecond is the maximum number of requests per second,
    // across the entire project.
    MaxProjectMessagesPerSecond = 10000

    // MaxACKDeadline is the maximum acknowledgement deadline that can be applied
    // to a leased subscription Message.
    MaxACKDeadline = 600 * time.Second
)

Cloud PubSub quota is documented here: https://cloud.google.com/pubsub/quotas

Variables ¶

var (
    // PublisherScopes is the set of OAuth2 scopes needed for a publisher to
    // publish messages.
    PublisherScopes = []string{
        pubsub.ScopePubSub,
    }

    // SubscriberScopes is the set of OAuth2 scopes needed for a subscriber to
    // pull and acknowledge messages.
    SubscriberScopes = []string{
        pubsub.ScopePubSub,
    }
)

type ClientFactory ¶ Uses

type ClientFactory interface {
    // Client returns the Pub/Sub publisher client to use.
    // Client will be closed when this UnbufferedPublisher is closed.
    Client(context.Context) (*vkit.PublisherClient, error)

    // RecreateClient is called if any publish calls fail.
    // This is used to tell the underlying service to maybe generate a new client.
    RecreateClient()
}

ClientFactory is passed into an UnbufferedPublisher to create or reset a client.

type Publisher ¶ Uses

type Publisher interface {
    Publish(c context.Context, msgs ...*pubsub.Message) ([]string, error)
    Close() error
}

Publisher is a generic interface to something that can publish Pub/Sub messages.

A Publisher should be Closed when finished with it.

type Subscription ¶ Uses

type Subscription string

Subscription is a Pub/Sub subscription name.

func NewSubscription ¶ Uses

func NewSubscription(project, name string) Subscription

NewSubscription generates a new Subscription for a given project and subscription name.

func (*Subscription) Set ¶ Uses

func (s *Subscription) Set(value string) error

Set implements flag.Value.

func (Subscription) Split ¶ Uses

func (s Subscription) Split() (p, n string)

Split returns the Subscription's project component. If no project is defined (malformed), an empty string will be returned.

func (Subscription) SplitErr ¶ Uses

func (s Subscription) SplitErr() (p, n string, err error)

SplitErr returns the Subscription's project and name components.

func (*Subscription) String ¶ Uses

func (s *Subscription) String() string

func (Subscription) Validate ¶ Uses

func (s Subscription) Validate() error

Validate returns an error if the subscription name is invalid.

type Topic ¶ Uses

type Topic string

Topic is a fully-qualified Pub/Sub project/topic name.

func NewTopic ¶ Uses

func NewTopic(project, name string) Topic

NewTopic generates a new Topic for a given project and topic name.

func (*Topic) Set ¶ Uses

func (t *Topic) Set(value string) error

Set implements flag.Value.

func (Topic) Split ¶ Uses

func (t Topic) Split() (p, n string)

Split returns the Topic's project component. If no project is defined (malformed), an empty string will be returned.

func (Topic) SplitErr ¶ Uses

func (t Topic) SplitErr() (p, n string, err error)

SplitErr returns the Topic's project and name components.

func (*Topic) String ¶ Uses

func (t *Topic) String() string

func (Topic) Validate ¶ Uses

func (t Topic) Validate() error

Validate returns an error if the topic name is invalid.

type UnbufferedPublisher ¶ Uses

type UnbufferedPublisher struct {
    // AECtx is the AppEngine context used to create a pubsub client.
    AECtx context.Context

    // Topic is the name of the Topic to publish to.
    Topic Topic

    // ClientFactory produces a client for the publisher.  This is called on each
    // and every publish request.  If a publish request fails, then RecreateClient is called.
    ClientFactory ClientFactory

    // CallOpts are arbitrary call options that will be passed to the Publish
    // call.
    CallOpts []gax.CallOption
}

UnbufferedPublisher directly instantiates a Pub/Sub client and publishes a message to it.

The standard Pub/Sub library has several issues, especially when used from AppEngine:

- It uses an empty Context, discarding AppEngine context.
- It uses a buffer, which expects a lifecycle beyond that of a simple
  AppEngine Request.

func (*UnbufferedPublisher) Close ¶ Uses

func (up *UnbufferedPublisher) Close() error

Close closes the UnbufferedPublisher, notably its Client.

func (*UnbufferedPublisher) Publish ¶ Uses

func (up *UnbufferedPublisher) Publish(c context.Context, msgs ...*pubsub.Message) ([]string, error)

Publish publishes a message immediately, blocking until it completes.

"c" must be an AppEngine context.