go: cloud.google.com/go/pubsub/internal/scheduler Index | Files

package scheduler

import "cloud.google.com/go/pubsub/internal/scheduler"

Index

Package Files

publish_scheduler.go receive_scheduler.go

type PublishScheduler Uses

type PublishScheduler struct {
    // Settings passed down to each bundler that gets created.
    DelayThreshold       time.Duration
    BundleCountThreshold int
    BundleByteThreshold  int
    BundleByteLimit      int
    BufferedByteLimit    int
    // contains filtered or unexported fields
}

PublishScheduler is a scheduler which is designed for Pub/Sub's Publish flow. It bundles items before handling them. All items in this PublishScheduler use the same handler.

Each item is added with a given key. Items added to the empty string key are handled in random order. Items added to any other key are handled sequentially.

func NewPublishScheduler Uses

func NewPublishScheduler(workers int, handle func(bundle interface{})) *PublishScheduler

NewPublishScheduler returns a new PublishScheduler.

The workers arg is the number of workers that will operate on the queue of work. A reasonably large number of workers is highly recommended. If the workers arg is 0, then a healthy default of 10 workers is used.

The scheduler does not use a parent context. If it did, canceling that context would immediately stop the scheduler without waiting for undelivered messages.

The scheduler should be stopped only with FlushAndStop.

func (*PublishScheduler) Add Uses

func (s *PublishScheduler) Add(key string, item interface{}, size int) error

Add adds an item to the scheduler at a given key.

Add never blocks. Buffering happens in the scheduler's publishers. There is no flow control.

Since ordered keys require only a single outstanding RPC at once, it is possible to send ordered key messages to Topic.Publish (and subsequently to PublishScheduler.Add) faster than the bundler can publish them to the Pub/Sub service, resulting in a backed up queue of Pub/Sub bundles. Each item in the bundler queue is a goroutine.

func (*PublishScheduler) FlushAndStop Uses

func (s *PublishScheduler) FlushAndStop()

FlushAndStop begins flushing items from bundlers and from the scheduler. It blocks until all items have been flushed.

func (*PublishScheduler) IsPaused Uses

func (s *PublishScheduler) IsPaused(orderingKey string) bool

IsPaused checks if the bundler associated with an ordering keys is paused.

func (*PublishScheduler) Pause Uses

func (s *PublishScheduler) Pause(orderingKey string)

Pause pauses the bundler associated with the provided ordering key, preventing it from accepting new messages. Any outstanding messages that haven't been published will error. If orderingKey is empty, this is a no-op.

func (*PublishScheduler) Resume Uses

func (s *PublishScheduler) Resume(orderingKey string)

Resume resumes accepting message with the provided ordering key.

type ReceiveScheduler Uses

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

ReceiveScheduler is a scheduler which is designed for Pub/Sub's Receive flow.

Each item is added with a given key. Items added to the empty string key are handled in random order. Items added to any other key are handled sequentially.

func NewReceiveScheduler Uses

func NewReceiveScheduler(workers int) *ReceiveScheduler

NewReceiveScheduler creates a new ReceiveScheduler.

The workers arg is the number of concurrent calls to handle. If the workers arg is 0, then a healthy default of 10 workers is used. If less than 0, this will be set to an large number, similar to PublishScheduler's handler limit.

func (*ReceiveScheduler) Add Uses

func (s *ReceiveScheduler) Add(key string, item interface{}, handle func(item interface{})) error

Add adds the item to be handled. Add may block.

Buffering happens above the ReceiveScheduler in the form of a flow controller that requests batches of messages to pull. A backed up ReceiveScheduler.Add call causes pushback to the pubsub service (less Receive calls on the long-lived stream), which keeps memory footprint stable.

func (*ReceiveScheduler) Shutdown Uses

func (s *ReceiveScheduler) Shutdown()

Shutdown begins flushing messages and stops accepting new Add calls. Shutdown does not block, or wait for all messages to be flushed.

Package scheduler imports 5 packages (graph) and is imported by 1 packages. Updated 2020-06-16. Refresh now. Tools for package owners.