stream

type KafkaStreamConfig interface {
	NewProducer(km *kafka.ConfigMap) (*kafka.Producer, error)
	NewConsumer(km *kafka.ConfigMap) (*kafka.Consumer, error)
	SetDeliveryError(f func(*kafka.Message))
	SetTopic(topic string)
	SetPrefix(prefix string)
	SetBrokers(brokers string)
	SetFlags()
	ProducerDefaults() *kafka.ConfigMap
	Produce(topic *string, value []byte) error
	Flush(ms int) int
	FullTopic(t string) string
	ChannelProduce(topic *string, value []byte)
	GetConsumer() *kafka.Consumer
	GetProducer() *kafka.Producer
	GetBrokers() string
	GetPrefix() string
	Close() error
}

type StreamConfig struct {
	Brokers  string `json:"brokers" yaml:"brokers" default:"kafka-atsu-prod-01:9092,kafka-atsu-prod-02:9092,kafka-atsu-prod-03:9092"`
	Prefix   string `json:"prefix" yaml:"prefix" default:"atsu"` // defaults to atsu
	Topic    string `json:"topic" yaml:"topic" default:"unset"`  // defaults to unset
	Messages int    `json:"messages" yaml:"messages"`
	Bytes    int    `json:"bytes" yaml:"bytes"`

	Timeout  time.Duration `ignored:"true" json:"-"` // Must be set explicitly
	Interval time.Duration `ignored:"true" json:"-"` // Must be set explicitly

	Offset          string `default:"latest" json:"offset" yaml:"offset"`
	GroupId         string `default:"atsu-unset-group-id" split_words:"true" json:"group_id" yaml:"group_id"`
	Glob            bool   `default:"false" json:"glob" yaml:"glob"`
	DeliveryReports bool   `default:"false" json:"reports" yaml:"reports"`

	Codec string `default:"none" json:"codec" yaml:"codec"`
	// contains filtered or unexported fields
}

type StreamConsumer interface {
	Start(*StreamConfig, interface{}) error
	Message(*kafka.Message) error // error != nil, stop consumer
	Interval(time.Time) error     // error != nil, stop consumer
	Timeout(time.Time, bool) bool // bool != false, stop consumer
	Error(kafka.Error) bool       // bool != false, stop consumer
	Process() (bool, error)
	Finish() error // This value will be returned to the caller

	DoneCh() <-chan bool
}