kafka

type API interface {
	Start()
	Stop()
}

type APIConfiguration struct {
	// Port default is 8090
	Port *int

	// HealthCheckPath default is /healthcheck
	HealthCheckPath *string
}

type Balancer kafka.Balancer

type BatchConfiguration struct {
	BatchConsumeFn    BatchConsumeFn
	PreBatchFn        PreBatchFn
	MessageGroupLimit int
}

type BatchConsumeFn func([]*Message) error

type ConsumeFn func(*Message) error

type Consumer interface {
	// Consume starts consuming
	Consume()

	// Pause function pauses consumer, it is stop consuming new messages
	// It works idempotent under the hood
	// Calling with multiple goroutines is safe
	Pause()

	// Resume function resumes consumer, it is start to working
	// It works idempotent under the hood
	// Calling with multiple goroutines is safe
	Resume()

	// GetMetricCollectors for the purpose of making metric collectors available.
	// You can register these collectors on your own http server.
	// Please look at the examples/with-metric-collector directory.
	GetMetricCollectors() []prometheus.Collector

	// WithLogger for injecting custom log implementation
	WithLogger(logger LoggerInterface)

	// Stop for graceful shutdown. In order to avoid data loss, you have to call it!
	Stop() error
}

type ConsumerConfig struct {
	DistributedTracingConfiguration DistributedTracingConfiguration
	Logger                          LoggerInterface
	APIConfiguration                APIConfiguration
	MetricConfiguration             MetricConfiguration
	SASL                            *SASLConfig
	TLS                             *TLSConfig
	Dial                            *DialConfig
	BatchConfiguration              *BatchConfiguration
	ConsumeFn                       ConsumeFn
	SkipMessageByHeaderFn           SkipMessageByHeaderFn
	TransactionalRetry              *bool
	RetryConfiguration              RetryConfiguration
	LogLevel                        LogLevel
	Rack                            string
	ClientID                        string
	Reader                          ReaderConfig
	CommitInterval                  time.Duration
	MessageGroupDuration            time.Duration
	Concurrency                     int
	DistributedTracingEnabled       bool
	RetryEnabled                    bool
	APIEnabled                      bool

	// MetricPrefix is used for prometheus fq name prefix.
	// If not provided, default metric prefix value is `kafka_konsumer`.
	// Currently, there are two exposed prometheus metrics. `processed_messages_total_current` and `unprocessed_messages_total_current`.
	// So, if default metric prefix used, metrics names are `kafka_konsumer_processed_messages_total_current` and
	// `kafka_konsumer_unprocessed_messages_total_current`.
	MetricPrefix string
}

type ConsumerMetric struct {
	TotalUnprocessedMessagesCounter int64
	TotalProcessedMessagesCounter   int64
}

type DialConfig struct {
	Timeout   time.Duration
	KeepAlive time.Duration
}

type Dialer struct {
	*kafka.Dialer
}

type DistributedTracingConfiguration struct {
	TracerProvider trace.TracerProvider
	Propagator     propagation.TextMapPropagator
}

type Header = protocol.Header

type IncomingMessage struct {
	// contains filtered or unexported fields
}

type Layer interface {
	SetTLSConfig(config *tls.Config)
	SetSASL(mechanism sasl.Mechanism)
}

type LogLevel string

type LoggerInterface interface {
	// With returns a logger based off the root logger and decorates it with the given context and arguments.
	With(args ...interface{}) LoggerInterface

	// Debug uses fmt.Sprint to construct and log a message at DEBUG level
	Debug(args ...interface{})
	// Info uses fmt.Sprint to construct and log a message at INFO level
	Info(args ...interface{})
	// Warn uses fmt.Sprint to construct and log a message at ERROR level
	Warn(args ...interface{})
	// Error uses fmt.Sprint to construct and log a message at ERROR level
	Error(args ...interface{})

	// Debugf uses fmt.Sprintf to construct and log a message at DEBUG level
	Debugf(format string, args ...interface{})
	// Infof uses fmt.Sprintf to construct and log a message at INFO level
	Infof(format string, args ...interface{})
	// Warnf uses fmt.Sprintf to construct and log a message at WARN level
	Warnf(format string, args ...interface{})
	// Errorf uses fmt.Sprintf to construct and log a message at ERROR level
	Errorf(format string, args ...interface{})

	Infow(msg string, keysAndValues ...interface{})
	Errorw(msg string, keysAndValues ...interface{})
	Warnw(msg string, keysAndValues ...interface{})
}

type Mechanism string

type Message struct {
	Time       time.Time
	WriterData interface{}

	// Context To enable distributed tracing support
	Context       context.Context
	Topic         string
	Key           []byte
	Value         []byte
	Headers       []Header
	Partition     int
	Offset        int64
	HighWaterMark int64

	// IsFailed Is only used on transactional retry disabled
	IsFailed bool

	// If available, kafka-konsumer writes this description into the failed message's
	// headers as `x-error-message` key when producing retry topic
	ErrDescription string
}

type MetricCollector struct {
	// contains filtered or unexported fields
}

type MetricConfiguration struct {
	// Path default is /metrics
	Path *string
}

type OtelKafkaKonsumerWriter interface {
	WriteMessage(ctx context.Context, msg segmentio.Message) error
	WriteMessages(ctx context.Context, msgs []segmentio.Message) error
	Close() error
}

type PreBatchFn func([]*Message) []*Message

type Producer interface {
	Produce(ctx context.Context, message Message) error
	ProduceBatch(ctx context.Context, messages []Message) error
	Close() error
}

type ProducerConfig struct {
	DistributedTracingConfiguration DistributedTracingConfiguration
	Transport                       *TransportConfig
	SASL                            *SASLConfig
	TLS                             *TLSConfig
	ClientID                        string
	Writer                          WriterConfig
	DistributedTracingEnabled       bool
}

type Reader interface {
	FetchMessage(ctx context.Context, msg *kafka.Message) error
	Close() error
	CommitMessages(messages []kafka.Message) error
}

type ReaderConfig kafka.ReaderConfig

type RetryConfiguration struct {
	// MetricPrefix is used for prometheus fq name prefix.
	// If not provided, default metric prefix value is `kafka_cronsumer`.
	// Currently, there are two exposed prometheus metrics. `retried_messages_total_current` and `discarded_messages_total_current`.
	// So, if default metric prefix used, metrics names are `kafka_cronsumer_retried_messages_total_current` and
	// `kafka_cronsumer_discarded_messages_total_current`.
	MetricPrefix string

	SASL                  *SASLConfig
	TLS                   *TLSConfig
	ClientID              string
	StartTimeCron         string
	Topic                 string
	DeadLetterTopic       string
	Rack                  string
	LogLevel              LogLevel
	Brokers               []string
	Balancer              Balancer
	MaxRetry              int
	WorkDuration          time.Duration
	SkipMessageByHeaderFn SkipMessageByHeaderFn
}

type SASLConfig struct {
	Type     Mechanism
	Username string
	Password string
}

type SkipMessageByHeaderFn func(header []kafka.Header) bool

type TLSConfig struct {
	RootCAPath         string
	IntermediateCAPath string
}

type Transport struct {
	*kafka.Transport
}

type TransportConfig struct {
	MetadataTopics []string
	DialTimeout    time.Duration
	IdleTimeout    time.Duration
	MetadataTTL    time.Duration
}

type Writer interface {
	WriteMessages(context.Context, ...kafka.Message) error
	Close() error
}

type WriterConfig struct {
	ErrorLogger            kafka.Logger
	Logger                 kafka.Logger
	Balancer               kafka.Balancer
	Completion             func(messages []kafka.Message, err error)
	Topic                  string
	Brokers                []string
	ReadTimeout            time.Duration
	BatchTimeout           time.Duration
	BatchBytes             int64
	WriteTimeout           time.Duration
	RequiredAcks           kafka.RequiredAcks
	BatchSize              int
	WriteBackoffMax        time.Duration
	WriteBackoffMin        time.Duration
	MaxAttempts            int
	Async                  bool
	Compression            kafka.Compression
	AllowAutoTopicCreation bool
}