kafka

type AcceptMessageFunc func(logrus.FieldLogger, []byte) (fault.TypedError, string)

type ConsumG struct {
	Log *logger.Logger

	AcceptMessageFunc AcceptMessageFunc
	KafkaClient       *KafkaClientImpl
	// contains filtered or unexported fields
}

type ConsumerConfiguration struct {
	Group             string
	Oldest            bool
	AcceptMessageFunc AcceptMessageFunc
}

type KafkaClient interface {
	Init(*logger.Logger, *KafkaConfig)
	KafkaConfig() *KafkaConfig
	Loop() fault.TypedError
	GroupLoop() fault.TypedError
	ConfigurationName() string
	Version() string
	Push([]byte) fault.TypedError
	PushText(string) fault.TypedError
	PushEvent(interface{}) fault.TypedError
	Ping() fault.TypedError
	Status() string
	IsWorked() bool
	GetMetrics() map[string]*Metrics
	GetHost() []string
	SetAcceptMessageFunc(AcceptMessageFunc)
}

type KafkaClientImpl struct {
	Log *logger.Logger

	Topic          string
	Partition      int32
	Offset         string
	MsgCount       int64
	IsWorkedStatus bool
	StatusError    error

	Metrics map[string]*Metrics
	Brokers []string
	// contains filtered or unexported fields
}

type KafkaConfig struct {
	ConfigurationName string
	Enable            bool
	ClientID          string
	Brokers           string
	Version           string
	Topic             string
	Verbose           bool
	Partition         int32
	ProducerEnable    bool
	ConsumerEnable    bool
	Consumer          ConsumerConfiguration
	ActionName        string
}

type Metrics struct {
	ProduceMsgCount         uint64
	ConsumeMsgCount         uint64
	ProduceFailedMsgCount   uint64
	ProduceSuccessMsgCount  uint64
	HealthCheckFailedCount  uint64
	ConsumeFailedMsgCount   uint64
	ConsumeSuccessMsgCount  uint64
	HealthCheckCount        uint64
	HealthCheckSuccessCount uint64
}