kafka

type ByteEncoder = sarama.ByteEncoder

type Consumer interface {
	Close() error
	// blocking to consume the messages
	Consume(topics string, mh ConsumerMessageHandler, eh ConsumerErrorHandler) error
	ConsumeM(topics []string, mh ConsumerMessageHandler, eh ConsumerErrorHandler) error
}

type ConsumerConfig struct {
	Key          string
	Address      []string // kafka地址
	Group        string   // groupId
	Offset       int64
	Ack          int    // ack类型
	User         string //username and password for SASL/PLAIN  or SASL/SCRAM authentication
	Password     string
	DialTimeout  time.Duration // How long to wait for the initial connection.
	ReadTimeout  time.Duration // How long to wait for a response.
	WriteTimeout time.Duration // How long to wait for a transmit.
	KeepAlive    time.Duration
	Version      *sarama.KafkaVersion // kafka version
}

type ConsumerError = sarama.ConsumerError

type ConsumerErrorHandler func(err error)

type ConsumerMessage = sarama.ConsumerMessage

type ConsumerMessageHandler func(msg *ConsumerMessage) error

type Producer interface {
	Close() error
	Produce(msgs ...*ProducerMessage) error
	AsyncHandle(mh ProducerMessageHandler, eh ProducerErrorHandler) // 必须设置 asyncReturnSuccess 或 asyncReturnError
}

type ProducerConfig struct {
	Key           string
	Address       []string // kafka地址
	Async         bool
	ReturnSuccess bool
	ReturnError   bool
	User          string //username and password for SASL/PLAIN  or SASL/SCRAM authentication
	Password      string
	Version       *sarama.KafkaVersion // kafka version
}

type ProducerError = sarama.ProducerError

type ProducerErrorHandler func(err *ProducerError)

type ProducerMessage = sarama.ProducerMessage

type ProducerMessageHandler func(msg *ProducerMessage)

type StringEncoder = sarama.StringEncoder