worker

Documentation

Index

• Variables
• type Consumer
• type KafkaScaleConsumer
  • func NewKafkaScaleConsumer(taskName string, name string, brokers []string, topic []string, ...) *KafkaScaleConsumer
  • func (c KafkaScaleConsumer) Consume()
• type KafkaScaleProducer
  • func NewKafkaScaleProducer(taskName string, name string, brokers []string, topic string, ...) *KafkaScaleProducer
  • func (p KafkaScaleProducer) Produce()
• type Producer
• type Worker
  • func NewWorker(name string, inputBrokers []string, inputTopics []string, ...) *Worker
  • func (w Worker) Fire()
  • func (w Worker) GetTaskName() string
  • func (w Worker) StopAndClean()

Variables

var AllTasks map[string]Worker

Types

type Consumer

type Consumer interface {
	Consume()
}

type KafkaScaleConsumer

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

func NewKafkaScaleConsumer

func NewKafkaScaleConsumer(
	taskName string,
	name string,
	brokers []string,
	topic []string,
	res chan<- model.KafkaLog,
	ctx context.Context,
	numConsumeChan chan<- int64,
	factor int) *KafkaScaleConsumer

func (KafkaScaleConsumer) Consume

func (c KafkaScaleConsumer) Consume()

type KafkaScaleProducer

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

func NewKafkaScaleProducer

func NewKafkaScaleProducer(
	taskName string,
	name string,
	brokers []string,
	topic string,
	res <-chan model.KafkaLog,
	ctx context.Context,
	factor int,
	numProduceSucceedChan chan<- int64,
	numProduceFailedChan chan<- int64) *KafkaScaleProducer

func (KafkaScaleProducer) Produce

func (p KafkaScaleProducer) Produce()

type Producer

type Producer interface {
	Produce()
}

type Worker

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

func NewWorker

func NewWorker(
	name string,
	inputBrokers []string,
	inputTopics []string,
	outputBrokers []string,
	outputTopic string,
	factor int,
	duration time.Duration,
	ms *metric.MetricSystem,
	workerAddress string) *Worker

func (Worker) Fire

func (w Worker) Fire()

func (Worker) GetTaskName

func (w Worker) GetTaskName() string

func (Worker) StopAndClean

func (w Worker) StopAndClean()