processor

Documentation

Index

• type KafkaMetricProcessor
  • func NewMetricProcessor(params *MetricProcessorParams) *KafkaMetricProcessor
  • func (s *KafkaMetricProcessor) Process(message Message) error
• type Message
• type MetricProcessor
  • func NewDecoratedProcessor(f metrics.Factory, processor MetricProcessor) MetricProcessor
• type MetricProcessorParams
• type ParallelProcessor
  • func NewParallelProcessor(processor MetricProcessor, parallelism int, logger *zap.Logger) *ParallelProcessor
  • func (k *ParallelProcessor) Close() error
  • func (k *ParallelProcessor) Process(message Message) error
  • func (k *ParallelProcessor) Start()

Types

type KafkaMetricProcessor

type KafkaMetricProcessor struct {
	io.Closer
	// contains filtered or unexported fields
}

func NewMetricProcessor

func NewMetricProcessor(params *MetricProcessorParams) *KafkaMetricProcessor

func (*KafkaMetricProcessor) Process

func (s *KafkaMetricProcessor) Process(message Message) error

Process unmarshals and writes a single kafka message

type Message

type Message interface {
	Value() []byte
}

type MetricProcessor

type MetricProcessor interface {
	Process(input Message) error
	io.Closer
}

func NewDecoratedProcessor

func NewDecoratedProcessor(f metrics.Factory, processor MetricProcessor) MetricProcessor

NewDecoratedProcessor returns a processor with metrics

type MetricProcessorParams

type MetricProcessorParams struct {
	Writer       metricstore.Writer
	Unmarshaller kafka.Unmarshaller
}

type ParallelProcessor

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

ParallelProcessor is a processor that processes in parallel using a pool of goroutines

func NewParallelProcessor

func NewParallelProcessor(
	processor MetricProcessor,
	parallelism int,
	logger *zap.Logger) *ParallelProcessor

NewParallelProcessor creates a new parallel processor

func (*ParallelProcessor) Close

func (k *ParallelProcessor) Close() error

Close terminates all running goroutines

func (*ParallelProcessor) Process

func (k *ParallelProcessor) Process(message Message) error

Process queues a message for processing

func (*ParallelProcessor) Start

func (k *ParallelProcessor) Start()

Start begins processing queued messages