kafka

package

v0.0.0-...-5c69eea Latest Latest Go to latest Published: Dec 18, 2020 License: MIT Imports: 9 Imported by: 4

Details

This section is empty.

This section is empty.

func WithAfterFailed(afterFailed AfterFailed) func(*Loadastic)

func WithAfterSend(afterSend AfterSend) func(*Loadastic)

func WithBeforeSend(beforeSend BeforeSend) func(*Loadastic)

func WithFailedChecker(checker FailedChecker) func(*Loadastic)

func WithInitialWorkers(initialWorkers uint) func(*Loadastic)

type AfterFailed func(request RecordPayload, err error, id uint64, uuid string)

type AfterSend func(request RecordPayload, response interface{}, id uint64, uuid string)

type BeforeSend func(request RecordPayload, tickerTimestamp time.Time, id uint64, uuid string)

type FailedChecker func(response interface{}) error

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

func (ks *KafkaSender) InitializeWorker() *KafkaWorker

func (ks *KafkaSender) Send(worker *KafkaWorker, payload RecordPayload) (interface{}, error)

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

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

func NewLoadastic(sender Sender, opts ...func(*Loadastic)) Loadastic

func (l *Loadastic) ExecutePace(requestFactory RequestFactory, pacer vegeta.Pacer, duration time.Duration)

func (l *Loadastic) StartSteps(requestFactory RequestFactory, steps ...common.Step)

type RecordPayload []byte

type RequestFactory func(tickerTimestamp time.Time, id uint64, uuid string) RecordPayload

func RandomRequestFactory(messageSize uint) RequestFactory

type Sender interface {
	InitializeWorker() *KafkaWorker
	Send(worker *KafkaWorker, request RecordPayload) (interface{}, error)
}

func NewKafkaSender(bootstrapUrl string, topicName string) (Sender, error)

func NewKafkaSenderFromSaramaClient(client sarama.Client, topicName string) (Sender, error)