subscription

Documentation

Index

• func NormalizeSubscriptionName(projectID string, subscriptionID string) string
• type DefaultSubscriptionFactory
  • func NewDefaultSubscriptionFactory(client *pubsub.Client, registry schema.SchemaConfigRegistry, ...) *DefaultSubscriptionFactory
  • func (f *DefaultSubscriptionFactory) Create(ctx context.Context, subscriptionID string) (*Subscription, error)
• type DefaultSubscriptionRegistry
  • func NewDefaultSubscriptionRegistry() *DefaultSubscriptionRegistry
  • func (r *DefaultSubscriptionRegistry) Add(subscription *Subscription)
  • func (r *DefaultSubscriptionRegistry) All() map[string]*Subscription
  • func (r *DefaultSubscriptionRegistry) Get(subscriptionID string) (*Subscription, error)
  • func (r *DefaultSubscriptionRegistry) Has(subscriptionID string) bool
• type Options
  • func DefaultSubscribeOptions() *Options
• type SubscribeFunc
• type SubscribeOption
  • func WithLegacyFlowControl(c bool) SubscribeOption
  • func WithMaxExtension(t time.Duration) SubscribeOption
  • func WithMaxExtensionPeriod(t time.Duration) SubscribeOption
  • func WithMaxOutstandingBytes(n int) SubscribeOption
  • func WithMaxOutstandingMessages(n int) SubscribeOption
  • func WithMinExtensionPeriod(t time.Duration) SubscribeOption
  • func WithNumGoroutines(n int) SubscribeOption
• type Subscription
  • func NewSubscription(codec codec.Codec, subscription *pubsub.Subscription) *Subscription
  • func (s *Subscription) BaseSubscription() *pubsub.Subscription
  • func (s *Subscription) Codec() codec.Codec
  • func (s *Subscription) Subscribe(ctx context.Context, f SubscribeFunc) error
  • func (s *Subscription) WithOptions(options ...SubscribeOption) *Subscription
• type SubscriptionFactory
• type SubscriptionRegistry

Functions

func NormalizeSubscriptionName

func NormalizeSubscriptionName(projectID string, subscriptionID string) string

NormalizeSubscriptionName normalizes a subscription name for given projectID and subscriptionID to be compatible with the "projects/{projectID}/subscriptions/{subscriptionID}" format.

Types

type DefaultSubscriptionFactory

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

DefaultSubscriptionFactory is the default SubscriptionFactory implementation.

func NewDefaultSubscriptionFactory

func NewDefaultSubscriptionFactory(client *pubsub.Client, registry schema.SchemaConfigRegistry, factory codec.CodecFactory) *DefaultSubscriptionFactory

NewDefaultSubscriptionFactory returns a new DefaultSubscriptionFactory instance.

func (*DefaultSubscriptionFactory) Create

func (f *DefaultSubscriptionFactory) Create(ctx context.Context, subscriptionID string) (*Subscription, error)

Create creates a new Subscription.

type DefaultSubscriptionRegistry

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

DefaultSubscriptionRegistry is the default SubscriptionRegistry implementation.

func NewDefaultSubscriptionRegistry

func NewDefaultSubscriptionRegistry() *DefaultSubscriptionRegistry

NewDefaultSubscriptionRegistry returns a new DefaultSubscriptionRegistry instance.

func (*DefaultSubscriptionRegistry) Add

func (r *DefaultSubscriptionRegistry) Add(subscription *Subscription)

Add registers a Subscription.

func (*DefaultSubscriptionRegistry) All

func (r *DefaultSubscriptionRegistry) All() map[string]*Subscription

All returns all registered Subscription.

func (*DefaultSubscriptionRegistry) Get

func (r *DefaultSubscriptionRegistry) Get(subscriptionID string) (*Subscription, error)

Get returns a registered Subscription for the provided subscriptionID.

func (*DefaultSubscriptionRegistry) Has

func (r *DefaultSubscriptionRegistry) Has(subscriptionID string) bool

Has returns true if the registry contains a Subscription for the provided subscriptionID.

type Options

type Options struct {
	ReceiveSettings pubsub.ReceiveSettings
}

Options represents subscription options.

func DefaultSubscribeOptions

func DefaultSubscribeOptions() *Options

DefaultSubscribeOptions is the default subscription options.

type SubscribeFunc

type SubscribeFunc func(ctx context.Context, m *message.Message)

SubscribeFunc represents the Subscription execution callback.

type SubscribeOption

type SubscribeOption func(o *Options)

SubscribeOption represents subscription functional options.

func WithLegacyFlowControl

func WithLegacyFlowControl(c bool) SubscribeOption

WithLegacyFlowControl sets the legacy flow control usage.

func WithMaxExtension

func WithMaxExtension(t time.Duration) SubscribeOption

WithMaxExtension sets the max extension.

func WithMaxExtensionPeriod

func WithMaxExtensionPeriod(t time.Duration) SubscribeOption

WithMaxExtensionPeriod sets the max extension period.

func WithMaxOutstandingBytes

func WithMaxOutstandingBytes(n int) SubscribeOption

WithMaxOutstandingBytes sets the max outstanding bytes.

func WithMaxOutstandingMessages

func WithMaxOutstandingMessages(n int) SubscribeOption

WithMaxOutstandingMessages sets the max outstanding messages.

func WithMinExtensionPeriod

func WithMinExtensionPeriod(t time.Duration) SubscribeOption

WithMinExtensionPeriod sets the min extension period.

func WithNumGoroutines

func WithNumGoroutines(n int) SubscribeOption

WithNumGoroutines sets the num of goroutines.

type Subscription

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

Subscription represents a pub/sub subscription with an associated codec.Codec.

func NewSubscription

func NewSubscription(codec codec.Codec, subscription *pubsub.Subscription) *Subscription

NewSubscription returns a new Subscription instance.

func (*Subscription) BaseSubscription

func (s *Subscription) BaseSubscription() *pubsub.Subscription

BaseSubscription returns the base pubsub.Subscription.

func (*Subscription) Codec

func (s *Subscription) Codec() codec.Codec

Codec returns the subscription associated codec.Codec.

func (*Subscription) Subscribe

func (s *Subscription) Subscribe(ctx context.Context, f SubscribeFunc) error

Subscribe starts the subscription and runs the provided SubscribeFunc.

func (*Subscription) WithOptions

func (s *Subscription) WithOptions(options ...SubscribeOption) *Subscription

WithOptions configures the subscription with a list of SubscribeOption.

type SubscriptionFactory

type SubscriptionFactory interface {
	Create(ctx context.Context, subscriptionID string) (*Subscription, error)
}

SubscriptionFactory is the interface for Subscription factories.

type SubscriptionRegistry

type SubscriptionRegistry interface {
	Has(subscriptionID string) bool
	Get(subscriptionID string) (*Subscription, error)
	Add(subscription *Subscription)
	All() map[string]*Subscription
}

SubscriptionRegistry is the interface for Subscription registries.