kafka

type Client interface {
	NewProducer(Common) (Producer, error)
	NewConsumer(Common) (Consumer, error)
	Brokers() []*sarama.Broker
}

type Common interface {
	ValidateConf() error
	Start() error
	Conf() conf.KafkaConf
	Producer() Producer
}

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

type Factory interface {
	NewClient(Common, *sarama.Config) (Client, error)
}

type GoRoutines interface {
	ConsumerMessagesLoop(consumer Consumer, producer Producer, wg *sync.WaitGroup)
	ProducerErrorLoop(consumer Consumer, producer Producer, wg *sync.WaitGroup)
	ProducerSuccessLoop(consumer Consumer, producer Producer, wg *sync.WaitGroup)
}

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

type SaramaKafkaFactory struct{}