topic

Documentation

Index

• type DefaultTopicFactory
  • func NewDefaultTopicFactory(client *pubsub.Client, registry schema.SchemaConfigRegistry, ...) *DefaultTopicFactory
  • func (f *DefaultTopicFactory) Create(ctx context.Context, topicID string) (*Topic, error)
• type DefaultTopicRegistry
  • func NewDefaultTopicRegistry() *DefaultTopicRegistry
  • func (r *DefaultTopicRegistry) Add(topic *Topic)
  • func (r *DefaultTopicRegistry) All() map[string]*Topic
  • func (r *DefaultTopicRegistry) Get(topicID string) (*Topic, error)
  • func (r *DefaultTopicRegistry) Has(topicID string) bool
• type MessageSettings
• type Options
  • func DefaultPublishOptions() *Options
• type PublishOption
  • func WithByteThreshold(n int) PublishOption
  • func WithCompression(c bool) PublishOption
  • func WithCompressionBytesThreshold(n int) PublishOption
  • func WithCountThreshold(n int) PublishOption
  • func WithDelayThreshold(t time.Duration) PublishOption
  • func WithFlowControlSettings(s pubsub.FlowControlSettings) PublishOption
  • func WithMessageAttributes(a map[string]string) PublishOption
  • func WithMessageOrderingKey(k string) PublishOption
  • func WithNumGoroutines(n int) PublishOption
  • func WithTimeout(t time.Duration) PublishOption
• type Topic
  • func NewTopic(codec codec.Codec, topic *pubsub.Topic) *Topic
  • func (t *Topic) BaseTopic() *pubsub.Topic
  • func (t *Topic) Codec() codec.Codec
  • func (t *Topic) Publish(ctx context.Context, data any) (*pubsub.PublishResult, error)
  • func (t *Topic) WithOptions(options ...PublishOption) *Topic
• type TopicFactory
• type TopicRegistry

Types

type DefaultTopicFactory

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

DefaultTopicFactory is the default TopicFactory implementation.

func NewDefaultTopicFactory

func NewDefaultTopicFactory(client *pubsub.Client, registry schema.SchemaConfigRegistry, factory codec.CodecFactory) *DefaultTopicFactory

NewDefaultTopicFactory returns a new DefaultTopicFactory instance.

func (*DefaultTopicFactory) Create

func (f *DefaultTopicFactory) Create(ctx context.Context, topicID string) (*Topic, error)

Create creates a new Topic.

type DefaultTopicRegistry

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

DefaultTopicRegistry is the default TopicRegistry implementation.

func NewDefaultTopicRegistry

func NewDefaultTopicRegistry() *DefaultTopicRegistry

NewDefaultTopicRegistry returns a new DefaultTopicRegistry instance.

func (*DefaultTopicRegistry) Add

func (r *DefaultTopicRegistry) Add(topic *Topic)

Add registers a Topic.

func (*DefaultTopicRegistry) All

func (r *DefaultTopicRegistry) All() map[string]*Topic

All returns all registered Topic.

func (*DefaultTopicRegistry) Get

func (r *DefaultTopicRegistry) Get(topicID string) (*Topic, error)

Get returns a registered Topic for the provided topicID.

func (*DefaultTopicRegistry) Has

func (r *DefaultTopicRegistry) Has(topicID string) bool

Has returns true if the registry contains a Topic for the provided topicID.

type MessageSettings

type MessageSettings struct {
	OrderingKey string
	Attributes  map[string]string
}

MessageSettings represents message publish options.

type Options

type Options struct {
	PublishSettings pubsub.PublishSettings
	MessageSettings MessageSettings
}

Options represents publish options.

func DefaultPublishOptions

func DefaultPublishOptions() *Options

DefaultPublishOptions is the default publish options.

type PublishOption

type PublishOption func(o *Options)

PublishOption represents publish functional options.

func WithByteThreshold

func WithByteThreshold(n int) PublishOption

WithByteThreshold sets the byte threshold.

func WithCompression

func WithCompression(c bool) PublishOption

WithCompression sets the compression usage.

func WithCompressionBytesThreshold

func WithCompressionBytesThreshold(n int) PublishOption

WithCompressionBytesThreshold sets the compression bytes threshold.

func WithCountThreshold

func WithCountThreshold(n int) PublishOption

WithCountThreshold sets the count threshold.

func WithDelayThreshold

func WithDelayThreshold(t time.Duration) PublishOption

WithDelayThreshold sets the delay threshold.

func WithFlowControlSettings

func WithFlowControlSettings(s pubsub.FlowControlSettings) PublishOption

WithFlowControlSettings sets the flow control settings.

func WithMessageAttributes

func WithMessageAttributes(a map[string]string) PublishOption

WithMessageAttributes sets the message attributes.

func WithMessageOrderingKey

func WithMessageOrderingKey(k string) PublishOption

WithMessageOrderingKey sets the message ordering key.

func WithNumGoroutines

func WithNumGoroutines(n int) PublishOption

WithNumGoroutines sets the num of goroutines.

func WithTimeout

func WithTimeout(t time.Duration) PublishOption

WithTimeout sets the timeout.

type Topic

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

Topic represents a pub/sub topic with an associated codec.Codec.

func NewTopic

func NewTopic(codec codec.Codec, topic *pubsub.Topic) *Topic

NewTopic returns a new Topic instance.

func (*Topic) BaseTopic

func (t *Topic) BaseTopic() *pubsub.Topic

BaseTopic returns the base pubsub.Topic.

func (*Topic) Codec

func (t *Topic) Codec() codec.Codec

Codec returns the topic associated codec.Codec.

func (*Topic) Publish

func (t *Topic) Publish(ctx context.Context, data any) (*pubsub.PublishResult, error)

Publish publishes the provided data.

func (*Topic) WithOptions

func (t *Topic) WithOptions(options ...PublishOption) *Topic

WithOptions configures the topic with a list of PublishOption.

type TopicFactory

type TopicFactory interface {
	Create(ctx context.Context, topicID string) (*Topic, error)
}

TopicFactory is the interface for Topic factories.

type TopicRegistry

type TopicRegistry interface {
	Has(topicID string) bool
	Get(topicID string) (*Topic, error)
	Add(topic *Topic)
	All() map[string]*Topic
}

TopicRegistry is the interface for Topic registries.