subscriber

Documentation

Index

• type Configuration
• type MultiSubscriber
  • func NewMultiSubscriber(opts ...Option) *MultiSubscriber
  • func (m *MultiSubscriber) AddSubscriber(subscriber *Subscriber)
  • func (m *MultiSubscriber) Start(ctx context.Context)
  • func (m *MultiSubscriber) Stop()
• type Option
  • func WithLogger(logger *zap.Logger) Option
  • func WithMetrics(r metrics.Registerer) Option
• type Processor
• type Subscriber
  • func NewSubscriber(name string, configuration Configuration, processor Processor, opts ...Option) *Subscriber
  • func (s Subscriber) Run(ctx context.Context)
  • func (s Subscriber) Stop()

Types

type Configuration

type Configuration struct {
	Project                      string `json:"project"`
	Subscription                 string `json:"subscription"`
	Topic                        string `json:"topic"`
	PullerNumGoroutines          int    `json:"pullerNumGoroutines"`
	PullerMaxOutstandingMessages int    `json:"pullerMaxOutstandingMessages"`
	PullerMaxOutstandingBytes    int    `json:"pullerMaxOutstandingBytes"`
	MaxMPS                       int    `json:"maxMPS"`
	WorkerNum                    int    `json:"workerNum"`
}

type MultiSubscriber

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

func NewMultiSubscriber

func NewMultiSubscriber(opts ...Option) *MultiSubscriber

func (*MultiSubscriber) AddSubscriber

func (m *MultiSubscriber) AddSubscriber(subscriber *Subscriber)

func (*MultiSubscriber) Start

func (m *MultiSubscriber) Start(ctx context.Context)

func (*MultiSubscriber) Stop

func (m *MultiSubscriber) Stop()

type Option

type Option func(*options)

func WithLogger

func WithLogger(logger *zap.Logger) Option

func WithMetrics

func WithMetrics(r metrics.Registerer) Option

type Processor

type Processor interface {
	Process(ctx context.Context, msgChan <-chan *puller.Message) error
}

type Subscriber

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

func NewSubscriber

func NewSubscriber(
	name string,
	configuration Configuration,
	processor Processor,
	opts ...Option,
) *Subscriber

func (Subscriber) Run

func (s Subscriber) Run(ctx context.Context)

func (Subscriber) Stop

func (s Subscriber) Stop()