consumer

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

type Handler interface {
	codec.Meta
	Handle(context.Context, Message)
	Offset(partition int32) int64
}

type HandlerBatch[T any] struct {
	// contains filtered or unexported fields
}

type HandlerBatchFunc[T any] func(ctx context.Context, msg []T, raw []Message) error

type HandlerOption func(*HandlerOptions)

type HandlerOptions struct {
	Logger *slog.Logger

	Timeout time.Duration

	BatchSize         int
	BatchFlushTimeout time.Duration

	Retry                bool
	RetryInitialInterval time.Duration
	RetryMaxInterval     time.Duration
	RetryMaxTime         time.Duration
}

type HandlerProto[T any] struct {
	// contains filtered or unexported fields
}

type HandlerProtoFunc[T any] func(ctx context.Context, msg T, raw Message) error

type Message struct {
	Key       []byte
	Value     []byte
	ID        string
	Group     string
	Topic     string
	TraceID   string
	Partition int32
	Offset    int64
	Timestamp time.Time
	Headers   []codec.Header
}

type Offset map[int32]int64

type OffsetSeq = iter.Seq2[int32, int64]

type Option func(*Options)

type Options struct {
	InitialOffset int64 // Determine when the group should to start consuming (default: "latest")

	FetchMaxWait  time.Duration // Waiting time before returning messages to the client (relevant if FetchMinBytes > 1) (default 500)
	FetchMinBytes int32         // Determine how many bytes kafka library have to receive before returning to the client (latency << FetchMinBytes << throughput) (default 1)

	CommitCount    int           // After how many messages do client have to call the commit (default 20)
	CommitDuration time.Duration // After how much time do client have to call the commit (default 1 second)

	DialTimeout  time.Duration // How long to wait for the initial connection
	ReadTimeout  time.Duration // How long to wait for a response
	WriteTimeout time.Duration // How long to wait for a transmit

	Logger     *slog.Logger   // Logger (default: slog.Default)
	TraceIDCtx TraceIDCtxFunc // Func which inserts trace id into the context
	SASL       *sasl.SASL     // SASL based authentication with broker
}

type TraceIDCtxFunc func(context.Context, string) context.Context