kafka

type ConsumerMessage struct {
	Key, Value []byte
	Topic      string
	Partition  int32
	Offset     int64

	Headers        []*RecordHeader // only set if kafka is version 0.11+
	Timestamp      time.Time       // only set if kafka is version 0.10+, inner message timestamp
	BlockTimestamp time.Time       // only set if kafka is version 0.10+, outer (compressed) block timestamp
}

type Handler func(ctx context.Context, session sarama.ConsumerGroupSession, message *ConsumerMessage) error

type ProducerError struct {
	Msg *ProducerMessage
	Err error
}

type ProducerMessage struct {
	Topic string // The Kafka topic for this message.
	// The partitioning key for this message. Pre-existing Encoders include
	// StringEncoder and ByteEncoder.
	Key string
	// The actual message to store in Kafka. Pre-existing Encoders include
	// StringEncoder and ByteEncoder.
	Value []byte

	// This field is used to hold arbitrary data you wish to include so it
	// will be available when receiving on the Successes and Errors channels.
	// Sarama completely ignores this field and is only to be used for
	// pass-through data.
	Metadata interface{}

	// Offset is the offset of the message stored on the broker. This is only
	// guaranteed to be defined if the message was successfully delivered and
	// RequiredAcks is not NoResponse.
	Offset int64
	// Partition is the partition that the message was sent to. This is only
	// guaranteed to be defined if the message was successfully delivered.
	Partition int32
	// Timestamp is the timestamp assigned to the message by the broker. This
	// is only guaranteed to be defined if the message was successfully
	// delivered, RequiredAcks is not NoResponse, and the Kafka broker is at
	// least version 0.10.0.
	Timestamp time.Time

	// MessageID
	MessageID string
}

type Reader interface {
	FetchMessage(ctx context.Context, handler Handler) error
	CommitMessage(ctx context.Context, session sarama.ConsumerGroupSession, message *ConsumerMessage) error
	Close() error
}

type ReaderOpt func(o *ReaderOpts)

type ReaderOpts struct {
	ServiceName string
	// The list of broker addresses used to connect to the kafka cluster.
	Brokers []string

	// The topic to read messages from.
	Topic string

	// GroupID holds the optional consumer group id.  If GroupID is specified, then
	// Partition should NOT be specified e.g. 0
	GroupID string

	// StartOffset determines from whence the consumer group should begin
	// consuming when it finds a partition without a committed offset.  If
	// non-zero, it must be set to one of FirstOffset or LastOffset.
	//
	// Default: FirstOffset
	//
	// Only used when Group is set
	StartOffset StartOffset

	CommitInterval int

	Logger logger.Logger
}

type RecordHeader struct {
	Key   []byte
	Value []byte
}

type RequiredAck int16

type StartOffset int

type Writer interface {
	SendMessage(ctx context.Context, message *ProducerMessage) error
	Errors() <-chan *ProducerError
	Messages() <-chan *ProducerMessage
	Close() (err error)
}

type WriterOpt func(o *WriterOpts)

type WriterOpts struct {
	ServiceName string
	Brokers     []string
	// Limit on how many attempts will be made to deliver a message.
	//
	// The default is to try at most 10 times.
	MaxAttempts int

	// Number of acknowledges from partition replicas required before receiving
	// a response to a produce request. The default is -1, which means to wait for
	// all replicas, and a value above 0 is required to indicate how many replicas
	// should acknowledge a message to be considered successful.
	//
	// This version of kafka-go (v0.3) does not support 0 required acks, due to
	// some internal complexity implementing this with the Kafka protocol. If you
	// need that functionality specifically, you'll need to upgrade to v0.4.
	RequiredAck RequiredAck

	ReadTimeout int

	Async bool

	Logger logger.Logger
}