kafka

Documentation

Overview

Package kafka implements tools to work with kafka Producers and Consumers.

Index

• func ConsumeLogChannel(emitter LogEmitter, logger *slog.Logger)
• func LogProcessor(logEmitter LogEmitter, logger *slog.Logger) (execute func() error, interrupt func(error))
• type BrokerAddressFamily
  • func (s BrokerAddressFamily) String() string
  • func (s *BrokerAddressFamily) UnmarshalJSON(data []byte) error
  • func (s *BrokerAddressFamily) UnmarshalText(text []byte) error
• type LogEmitter
• type TimeDurationMilliSeconds
  • func (d TimeDurationMilliSeconds) Duration() time.Duration
  • func (d TimeDurationMilliSeconds) String() string
  • func (d *TimeDurationMilliSeconds) UnmarshalJSON(data []byte) error
  • func (d *TimeDurationMilliSeconds) UnmarshalText(text []byte) error

Functions

func ConsumeLogChannel

func ConsumeLogChannel(emitter LogEmitter, logger *slog.Logger)

ConsumeLogChannel is supposed to be called in a goroutine. It consumes a log channel returned by a LogEmitter.

func LogProcessor

func LogProcessor(logEmitter LogEmitter, logger *slog.Logger) (execute func() error, interrupt func(error))

LogProcessor consumes logs from a LogEmitter and passes them to an slog.Logger.

Types

type BrokerAddressFamily

type BrokerAddressFamily string

const (
	BrokerAddressFamilyAny  BrokerAddressFamily = "any"
	BrokerAddressFamilyIPv4 BrokerAddressFamily = "v4"
	BrokerAddressFamilyIPv6 BrokerAddressFamily = "v6"
)

func (BrokerAddressFamily) String

func (s BrokerAddressFamily) String() string

func (*BrokerAddressFamily) UnmarshalJSON

func (s *BrokerAddressFamily) UnmarshalJSON(data []byte) error

func (*BrokerAddressFamily) UnmarshalText

func (s *BrokerAddressFamily) UnmarshalText(text []byte) error

type LogEmitter

type LogEmitter interface {
	Logs() chan kafka.LogEvent
}

LogEmitter emits logs from a kafka.Consumer or kafka.Producer.

Requires `go.logs.channel.enable` option set to true.

This feature was implemented in this PR.

type TimeDurationMilliSeconds

type TimeDurationMilliSeconds time.Duration

func (TimeDurationMilliSeconds) Duration

func (d TimeDurationMilliSeconds) Duration() time.Duration

func (TimeDurationMilliSeconds) String

func (d TimeDurationMilliSeconds) String() string

func (*TimeDurationMilliSeconds) UnmarshalJSON

func (d *TimeDurationMilliSeconds) UnmarshalJSON(data []byte) error

func (*TimeDurationMilliSeconds) UnmarshalText

func (d *TimeDurationMilliSeconds) UnmarshalText(text []byte) error