kafkax

package

v0.0.0-...-322c5b9 Latest Latest Go to latest Published: Mar 4, 2025 License: MIT Imports: 8 Imported by: 0

Details

This section is empty.

View Source

var (
	ErrReaderClosed = errors.New("kafka reader closed")
)

This section is empty.

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

func NewBulkHandler[T any](logger *slog.Logger,
	fn func(ctx context.Context, msg []kafka.Message, t []T) error) *BulkHandler[T]

func (h *BulkHandler[T]) Close()

func (h *BulkHandler[T]) ReadMsg(ctx context.Context, r *kafka.Reader) error

func (h *BulkHandler[T]) SetBatchSize(batchSize int) *BulkHandler[T]

func (h *BulkHandler[T]) SetDecodeFn(fn func([]byte) (T, error)) *BulkHandler[T]

func (h *BulkHandler[T]) SetInterval(interval time.Duration) *BulkHandler[T]

type Consumer interface {
	Start() error
}

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

func NewHandler[T any](logger *slog.Logger,
	fn func(ctx context.Context, msg kafka.Message, t T) error) *Handler[T]

func (h *Handler[T]) Close()

func (h *Handler[T]) ReadMsg(ctx context.Context, r *kafka.Reader) error

func (h *Handler[T]) SetDecodeFn(fn func([]byte) (T, error)) *Handler[T]