kafetrain

type Config struct {
	// Base configuration.
	Brokers []string `envconfig:"KAFKA_BROKERS" required:"true"`
	Version string   `envconfig:"KAFKA_VERSION" required:"true"`
	//Topic    string   `envconfig:"KAFKA_TOPIC" required:"true"`
	GroupID  string `envconfig:"KAFKA_GROUP_ID" required:"true"`
	ClientID string `envconfig:"KAFKA_CLIENT_ID" default:"kafetrain-consumer"`

	// Security configuration.
	Username string `envconfig:"KAFKA_USERNAME"`
	Password string `envconfig:"KAFKA_PASSWORD"`
	CACert   string `envconfig:"KAFKA_CA_CERT"`

	// Consumer configuration.
	MaxProcessingTime uint16 `envconfig:"KAFKA_MAX_PROCESSING_TIME_MS" default:"100"`
	InitialOffset     int64  `envconfig:"KAFKA_CONSUMER_INITIAL_OFFSET" default:"-2"`
	Silent            bool   `envconfig:"KAFKA_CONSUMER_SILENT" default:"false"`

	// Streamer configuration.
	BuffSize uint16 `envconfig:"KAFKA_MAX_PROCESSING_TIME_MS" default:"256"`

	// Retry configuration.
	RetryTopicPrefix    string `envconfig:"KAFKA_RETRY_TOPIC_PREFIX" default:"retry"`       // topic for messages to Retry
	RedirectTopicPrefix string `envconfig:"KAFKA_REDIRECT_TOPIC_PREFIX" default:"redirect"` // topic with message ids that should be retried
	RetrySegmentSize    int64  `envconfig:"KAFKA_REDIRECT_TOPIC_PREFIX" default:"redirect"` // topic with message ids that should be retried
}

type Consumer interface {
	Consume(ctx context.Context, topic string, messageHandler MessageHandler) error
	Close() error
}

type ErrorTracker struct {
	sync.Mutex
	// contains filtered or unexported fields
}

type ErrorTrackerConfig struct {
}

type HandlerRegistry struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

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

type HeaderList []*Header

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

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

type KeyTracker struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

type Message struct {
	Key     []byte
	Payload []byte
	Headers HeaderList

	Timestamp time.Time
	// contains filtered or unexported fields
}

type MessageChainTracker interface {
	IsRelated(ctx context.Context, msg Message) bool             // IsRelated returns true if message is related to error chain
	AddMessage(ctx context.Context, msg Message) (string, error) // AddMessage adds message to error chain
	ReleaseMessage(ctx context.Context, msg Message) error       // ReleaseMessage removes message from error chain
}

type MessageHandleFunc func(ctx context.Context, message Message) error

type MessageHandler interface {
	Handle(ctx context.Context, msg Message) error
}

type Middleware func(next MessageHandleFunc) MessageHandleFunc

type Option interface {
	Apply(cfg *consumerOptionConfig)
}

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

type RetriableError struct {
	Origin error
	Retry  bool
}

type Streamer interface {
	Stream(ctx context.Context, topic string) (<-chan Message, <-chan error)
	Close() error
}