pubsub

Documentation

Overview

Package pubsub is the implementation of the receiver interface and the publisher interface for pubsub.

Index

• Variables
• type PublishSettings
• type Publisher
  • func NewPublisher(ctx context.Context, config PublisherConfig, settings PublishSettings) (*Publisher, error)
  • func (p *Publisher) Close() error
  • func (p *Publisher) Topic(topicID string) (broker.Topic, error)
• type PublisherConfig
• type ReceiveSettings
• type Receiver
  • func NewReceiver(ctx context.Context, config ReceiverConfig, settings ReceiveSettings) (*Receiver, error)
  • func (p *Receiver) Close() error
  • func (p *Receiver) Receive(ctx context.Context, f func(context.Context, broker.Message)) error
• type ReceiverConfig
• type Topic
  • func (t *Topic) BatchPublish(ctx context.Context, messages ...broker.OutboundMessage) error
  • func (t *Topic) Publish(ctx context.Context, message broker.OutboundMessage) error

Variables

var DefaultPublishSettings = PublishSettings{
	DelayThreshold:         50 * time.Millisecond,
	CountThreshold:         50,
	ByteThreshold:          50 * 1024 * 1024,
	NumGoroutines:          5,
	Timeout:                15 * time.Second,
	MaxOutstandingMessages: 800,
	MaxOutstandingBytes:    1000 * 1024 * 1024,
	EnableMessageOrdering:  false,
}

var DefaultReceiveSettings = ReceiveSettings{
	MaxExtension:           30 * time.Minute,
	MaxExtensionPeriod:     3 * time.Minute,
	MaxOutstandingMessages: 1000,
	MaxOutstandingBytes:    400 * 1024 * 1024,
	NumGoroutines:          10,
}

Types

type PublishSettings

type PublishSettings struct {
	DelayThreshold         time.Duration
	CountThreshold         int
	ByteThreshold          int
	NumGoroutines          int
	Timeout                time.Duration
	MaxOutstandingMessages int
	MaxOutstandingBytes    int
	EnableMessageOrdering  bool
}

type Publisher

type Publisher struct {
	Settings pubsub.PublishSettings
	// contains filtered or unexported fields
}

func NewPublisher

func NewPublisher(ctx context.Context, config PublisherConfig, settings PublishSettings) (*Publisher, error)

func (*Publisher) Close

func (p *Publisher) Close() error

func (*Publisher) Topic

func (p *Publisher) Topic(topicID string) (broker.Topic, error)

type PublisherConfig

type PublisherConfig struct {
	// ProjectID is the unique identifier of the GCP project.
	ProjectID string

	// GrpcClientConn optionally sets the grpc.ClientConn which is used for communicating with Pub/Sub.
	// See GitHub issue on this link for more information:
	// https://github.com/googleapis/google-cloud-go/issues/1410
	GrpcClientConn *grpc.ClientConn
}

PublisherConfig defines configuration properties needed for initializing a Pub/Sub producer.

type ReceiveSettings

type ReceiveSettings struct {
	MaxExtension           time.Duration
	MaxExtensionPeriod     time.Duration
	MaxOutstandingMessages int
	MaxOutstandingBytes    int
	NumGoroutines          int
}

type Receiver

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

func NewReceiver

func NewReceiver(ctx context.Context, config ReceiverConfig, settings ReceiveSettings) (*Receiver, error)

func (*Receiver) Close

func (p *Receiver) Close() error

func (*Receiver) Receive

func (p *Receiver) Receive(ctx context.Context, f func(context.Context, broker.Message)) error

type ReceiverConfig

type ReceiverConfig struct {
	// ProjectID is the unique identifier of the GCP project.
	ProjectID string

	// SubscriptionID is the unique identifier of the Pub/Sub subscription.
	SubscriptionID string

	// GrpcClientConn optionally sets the grpc.ClientConn which is used for communicating with Pub/Sub.
	// See GitHub issue on this link for more information:
	// https://github.com/googleapis/google-cloud-go/issues/1410
	GrpcClientConn *grpc.ClientConn
}

ReceiverConfig defines configuration properties needed for initializing a Pub/Sub receiver.

type Topic

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

func (*Topic) BatchPublish

func (t *Topic) BatchPublish(ctx context.Context, messages ...broker.OutboundMessage) error

func (*Topic) Publish

func (t *Topic) Publish(ctx context.Context, message broker.OutboundMessage) error