collector

Documentation

Index

• func NewCollectors(config *config.Config, newRecord func() interface{}, ...) ([]*Service, []Collector, error)
• type Accumulator
  • func NewAccumulator() *Accumulator
  • func (a *Accumulator) Get(key interface{}) (interface{}, bool)
  • func (a *Accumulator) Len() int
  • func (a *Accumulator) Put(key, value interface{})
• type Batch
  • func NewBatch(stream *tconfig.Stream, fs afs.Service, options ...Option) (*Batch, error)
  • func (b Batch) IsActive(batch *config.Batch) bool
• type Collector
• type Option
  • func WithInstanceId(instanceId string) Option
  • func WithStreamURLSymLinkTrg(streamURLSymLinkTrg string) Option
• type Options
  • func NewOptions(options ...Option) *Options
  • func (o *Options) Apply(opts ...Option)
  • func (o *Options) GetInstanceId() string
  • func (o *Options) GetStreamURLSymLinkTrg() string
• type Service
  • func New(config *config.Config, newRecord func() interface{}, ...) (*Service, error)
  • func (s *Service) Close() error
  • func (s *Service) Collect(record interface{}) error
  • func (s *Service) FlushInBackground(batch *Batch) error
  • func (s *Service) NotifyWatcher()
  • func (s *Service) RetryFailed(ctx context.Context, onStartUp bool) error

Functions

func NewCollectors

func NewCollectors(config *config.Config,
	newRecord func() interface{},
	key func(record interface{}) interface{},
	reducer func(key, source interface{}),
	mapper func(accumulator *Accumulator) interface{},
	loader loader2.Loader, metrics *gmetric.Service, count int) ([]*Service, []Collector, error)

Types

type Accumulator

type Accumulator struct {
	Map map[interface{}]interface{}
	sync.RWMutex
}

func NewAccumulator

func NewAccumulator() *Accumulator

func (*Accumulator) Get

func (a *Accumulator) Get(key interface{}) (interface{}, bool)

func (*Accumulator) Len

func (a *Accumulator) Len() int

func (*Accumulator) Put

func (a *Accumulator) Put(key, value interface{})

type Batch

type Batch struct {
	ID string
	*tconfig.Stream
	Accumulator *Accumulator
	Started     time.Time
	Count       int32

	PendingURL string
	sync.Mutex
	// contains filtered or unexported fields
}

func NewBatch

func NewBatch(stream *tconfig.Stream, fs afs.Service, options ...Option) (*Batch, error)

func (Batch) IsActive

func (b Batch) IsActive(batch *config.Batch) bool

type Collector

type Collector interface {
	Collect(record interface{}) error
}

Collector represents a collector

type Option

type Option func(o *Options)

func WithInstanceId

func WithInstanceId(instanceId string) Option

func WithStreamURLSymLinkTrg

func WithStreamURLSymLinkTrg(streamURLSymLinkTrg string) Option

type Options

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

func NewOptions

func NewOptions(options ...Option) *Options

func (*Options) Apply

func (o *Options) Apply(opts ...Option)

func (*Options) GetInstanceId

func (o *Options) GetInstanceId() string

func (*Options) GetStreamURLSymLinkTrg

func (o *Options) GetStreamURLSymLinkTrg() string

type Service

type Service struct {
	*msg.Provider
	// contains filtered or unexported fields
}

func New

func New(config *config.Config,
	newRecord func() interface{},
	key func(record interface{}) interface{},
	reducer func(key, source interface{}),
	mapper func(accumulator *Accumulator) interface{},
	loader loader2.Loader, metrics *gmetric.Service, options ...Option) (*Service, error)

func (*Service) Close

func (s *Service) Close() error

func (*Service) Collect

func (s *Service) Collect(record interface{}) error

func (*Service) FlushInBackground

func (s *Service) FlushInBackground(batch *Batch) error

func (*Service) NotifyWatcher

func (s *Service) NotifyWatcher()

func (*Service) RetryFailed

func (s *Service) RetryFailed(ctx context.Context, onStartUp bool) error