maidenlanedkafka

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

type KafkaBridgeConf struct {
	Kafka       KafkaCommonConf `json:"kafka"`
	MaxInFlight int             `json:"maxInFlight"`
	maidenlanedtx.TxnProcessorConf
	maidenlanedeth.RPCConf
}

type KafkaClient interface {
	NewProducer(KafkaCommon) (KafkaProducer, error)
	NewConsumer(KafkaCommon) (KafkaConsumer, error)
	Brokers() []*sarama.Broker
}

type KafkaCommon interface {
	ValidateConf() error
	CobraInit(cmd *cobra.Command)
	Start() error
	Conf() *KafkaCommonConf
	Producer() KafkaProducer
}

type KafkaCommonConf struct {
	Brokers       []string `json:"brokers"`
	ClientID      string   `json:"clientID"`
	ConsumerGroup string   `json:"consumerGroup"`
	TopicIn       string   `json:"topicIn"`
	TopicOut      string   `json:"topicOut"`
	ProducerFlush struct {
		Frequency int `json:"frequency"`
		Messages  int `json:"messages"`
		Bytes     int `json:"bytes"`
	} `json:"producerFlush"`
	SASL struct {
		Username string
		Password string
	} `json:"sasl"`
	TLS maidenlanedutils.TLSConfig `json:"tls"`
}

type KafkaConsumer interface {
	Close() error
	Messages() <-chan *sarama.ConsumerMessage
	Errors() <-chan error
	MarkOffset(*sarama.ConsumerMessage, string)
}

type KafkaFactory interface {
	NewClient(KafkaCommon, *sarama.Config) (KafkaClient, error)
}

type KafkaGoRoutines interface {
	ConsumerMessagesLoop(consumer KafkaConsumer, producer KafkaProducer, wg *sync.WaitGroup)
	ProducerErrorLoop(consumer KafkaConsumer, producer KafkaProducer, wg *sync.WaitGroup)
	ProducerSuccessLoop(consumer KafkaConsumer, producer KafkaProducer, wg *sync.WaitGroup)
}

type KafkaProducer interface {
	AsyncClose()
	Input() chan<- *sarama.ProducerMessage
	Successes() <-chan *sarama.ProducerMessage
	Errors() <-chan *sarama.ProducerError
}

type MockKafkaConsumer struct {
	MockMessages       chan *sarama.ConsumerMessage
	MockErrors         chan error
	OffsetsByPartition map[int32]int64
}

type MockKafkaFactory struct {
	ClientConf         *sarama.Config
	ErrorOnNewClient   error
	ErrorOnNewProducer error
	ErrorOnNewConsumer error
	Producer           *MockKafkaProducer
	Consumer           *MockKafkaConsumer
}

type MockKafkaProducer struct {
	MockInput     chan *sarama.ProducerMessage
	MockSuccesses chan *sarama.ProducerMessage
	MockErrors    chan *sarama.ProducerError
	Closed        bool
	CloseSync     sync.Mutex
}

type SaramaKafkaFactory struct{}