wkafka

Jan 27, 2025

Changes in this version

type Record

+ func CtxRecordWithIndex(ctx context.Context, index int) *Record

Jan 21, 2025

Changes in this version

type DLQError

+ func (e *DLQError) IsZero() bool

+ func (e *DLQError) Unwrap() error

Jan 17, 2025

Changes in this version

+ type DLQTriggerSpec struct

+ Offset int64

+ Partition int32

+ Topic string

type OptionDLQTrigger

+ Spec *DLQTriggerSpec

+ SpecPartitions map[string][]int32

type OptionDLQTriggerFn

+ func WithDLQTriggerForce() OptionDLQTriggerFn

+ func WithDLQTriggerSpec(specs *DLQTriggerSpec) OptionDLQTriggerFn

+ func WithDLQTriggerSpecPartitions(partitions map[string][]int32) OptionDLQTriggerFn

Jan 16, 2025

Changes in this version

type Client

+ func (c *Client) DLQRetry(opts ...OptionDLQTriggerFn)

type DLQRecord

+ Err error

+ type DLQTriggerSpecs struct

+ Offset int64

+ Partition int32

+ Topic string

+ type OptionDLQTrigger struct

+ Force bool

+ Specs *DLQTriggerSpecs

+ func (o *OptionDLQTrigger) ToOption() OptionDLQTriggerFn

+ type OptionDLQTriggerFn func(*OptionDLQTrigger)

+ func WithDLQTriggerSpecs(specs *DLQTriggerSpecs) OptionDLQTriggerFn

+ func WithForceDLQTrigger() OptionDLQTriggerFn

Jan 15, 2025

Changes in this version

type Client

+ func (c *Client) AddTrigger(t func(context.Context))

+ func (c *Client) AppName() string

+ func (c *Client) DLQRetryAt() time.Time

+ func (c *Client) Logger() Logger

type Config

+ Plugins map[string]interface{}

+ type DLQRecord struct

+ Record *kgo.Record

+ RetryAt time.Time

Jan 10, 2025

Changes in this version

+ func ToPtr[T any](v T) *T

type Client

+ func (c *Client) GetLogger() Logger

+ func (c *Client) GroupID() string

type Config

+ Plugin map[string]interface{}

type Option

+ func WithPlugin(name string, fn PluginFunc) Option

+ type PluginFunc func(ctx context.Context, client *Client, cfg interface{}) error

Dec 19, 2024

Changes in this version

+ var DefaultRetryMaxInterval = 15 * time.Minute

type Client

+ Topics []string

+ Trigger []func()

+ func (c *Client) ClientID() []byte

+ func (c *Client) DLQRecord() *kgo.Record

+ func (c *Client) SkipCheck() SkipMap

type DLQConfig

+ RetryMaxInterval time.Duration

type DLQError

+ func IsDQLError(err error) (*DLQError, bool)

Oct 22, 2024

Changes in this version

type DLQConfig

+ Disabled bool

Jul 15, 2024

Changes in this version

+ const HeaderServiceKey

type Client

+ func (c *Client) GetClientID() []byte

type Producer

+ func (p *Producer[T]) GetTopic() string

Jun 25, 2024

Changes in this version

+ var ErrConnection = errors.New("connect to kafka brokers failed")

type Client

+ Brokers []string

+ func (c *Client) DLQTopics() []string

+ type LogNoop struct

+ func (LogNoop) Debug(_ string, _ ...interface{})

+ func (LogNoop) Error(_ string, _ ...interface{})

+ func (LogNoop) Info(_ string, _ ...interface{})

+ func (LogNoop) Warn(_ string, _ ...interface{})

+ type Logger interface

+ Debug func(msg string, keysAndValues ...interface{})

+ Error func(msg string, keysAndValues ...interface{})

+ Info func(msg string, keysAndValues ...interface{})

+ Warn func(msg string, keysAndValues ...interface{})

type Option

+ func WithPing(v bool) Option

+ func WithPingBackoff(b backoff.BackOff) Option

+ func WithPingRetry(v bool) Option

Jun 12, 2024

Changes in this version

+ func SkipAppend(skip SkipMap) func(SkipMap) SkipMap

+ func SkipReplace(skip SkipMap) func(SkipMap) SkipMap

type Client

+ func (c *Client) Skip(modify func(SkipMap) SkipMap)

+ type SkipMap = map[string]map[int32]OffsetConfig

Apr 19, 2024

Changes in this version

type Client

+ Meter Meter

+ type Meter interface

+ Meter func(start time.Time, batchSize int64, topic string, err error, isDLQ bool)

+ func EmptyMeter() Meter

+ func NewMeter(meterFunc func(time.Time, int64, string, error, bool)) Meter

type Option

+ func WithMeter(m Meter) Option

Feb 9, 2024

Changes in this version

+ const KeyIsDLQEnabled

+ const KeyIsDLQProcess

+ func CtxIsDLQEnabled(ctx context.Context) bool

+ func CtxIsDLQProcess(ctx context.Context) bool

Jan 7, 2024

Changes in this version

+ func WrapErrDLQ(err error) error

+ type DLQConfig struct

+ Disable bool

+ RetryInterval time.Duration

+ Skip map[int32]OffsetConfig

+ SkipExtra map[string]map[int32]OffsetConfig

+ StartOffset int64

+ Topic string

+ TopicsExtra []string

+ type DLQError struct

+ Err error

+ Indexes map[int]error

+ func (e *DLQError) Error() string

+ type OffsetConfig struct

+ Before int64

+ Offsets []int64

Jan 7, 2024

Changes in this version

+ const KeyIsDLQ

+ var DefaultRetryInterval = 10 * time.Second

+ var ErrDLQ = fmt.Errorf("send to DLQ")

+ var ScramSha256 = "SCRAM-SHA-256"

+ var ScramSha512 = "SCRAM-SHA-512"

+ func CtxIsDLQ(ctx context.Context) bool

+ type CallBackFunc func(*optionConsumer) error

type Client

+ KafkaDLQ *kgo.Client

type ConsumerPreConfig

+ FormatDLQTopic string

type DLQ

+ Disable bool

+ RetryInterval time.Duration

+ Skip map[int32]Offsets

+ SkipExtra map[string]map[int32]Offsets

+ StartOffset int64

+ TopicsExtra []string

+ type DLQIndexedError struct

+ Err error

+ Indexes map[int]error

+ func (e *DLQIndexedError) Error() string

type Option

+ func WithAppName(appName string) Option

+ func WithKGOOptionsDLQ(opts ...kgo.Opt) Option

+ func WithLogger(logger logz.Adapter) Option

+ func WithNoLogger(v bool) Option

type OptionConsumer

+ func WithCallbackDLQ[T any](fn func(ctx context.Context, msg T) error) OptionConsumer

+ type OptionProducer func(producerConfigInf) error

+ func WithEncoder[T any](fn func(T) ([]byte, error)) OptionProducer

+ func WithHeaders[T any](headers ...Header) OptionProducer

+ func WithHook[T any](fn func(T, *Record) error) OptionProducer

type Record

+ func CtxRecord(ctx context.Context) *Record

+ func CtxRecordBatch(ctx context.Context) []*Record

Jan 3, 2024

Changes in this version

type Client

+ func New(ctx context.Context, cfg Config, opts ...Option) (*Client, error)

type Config

+ Consumer ConsumerPreConfig

+ type ConsumerConfig struct

+ BatchCount int

+ DLQ DLQ

+ GroupID string

+ MaxPollRecords int

+ Skip map[string]map[int32]Offsets

+ StartOffset int64

+ Topics []string

+ type ConsumerPreConfig struct

+ PrefixGroupID string

+ Validation Validation

+ func (c ConsumerPreConfig) Apply(consumerConfig ConsumerConfig) (ConsumerConfig, error)

+ type DLQ struct

+ Enabled bool

+ Topic string

+ type GroupIDValidation struct

+ DisableWordBoundary bool

+ Enabled bool

+ RgxGroupID string

+ func (v GroupIDValidation) Validate(groupID string) error

+ type OptionConsumer func(*optionConsumer) error

+ func WithCallbackBatch[T any](fn func(ctx context.Context, msg []T) error) OptionConsumer

+ func WithCallback[T any](fn func(ctx context.Context, msg T) error) OptionConsumer

+ func WithDecode[T any](fn func(raw []byte, r *kgo.Record) (T, error)) OptionConsumer

+ func WithDisableCommit() OptionConsumer

+ func WithPreCheck(fn func(ctx context.Context, r *kgo.Record) error) OptionConsumer

type Validation

+ func (v Validation) Validate(consumerConfig ConsumerConfig) error

Dec 28, 2023

Changes in this version

+ var DefaultBatchCount = 100

type ConsumeConfig

+ Validation Validation

+ func (c ConsumeConfig) Validate() error

+ type OptionBatch func(*optionBatch)

+ func WithBatchConcurrent() OptionBatch

+ func WithBatchDisableCommit() OptionBatch

+ func WithSingleConcurrent() OptionBatch

+ func WithSingleDisableCommit() OptionBatch

+ type OptionSingle func(*optionSingle)

+ type Validation struct

+ GroupID string

Dec 14, 2023

Changes in this version

type Option

+ func WithClientInfo(name, version string) Option

Nov 7, 2023

Changes in this version

type Config

+ Compressions []string

type ConsumeConfig

+ BatchCount int

+ func (c ConsumeConfig) WithConcurrent(v bool) ConsumeConfig

type Option

+ func WithConsumerBatch[T any](cfg ConsumeConfig, processor Processor[[]T]) Option

+ type Processor interface

+ Process func(ctx context.Context, msg T) error

+ type ProcessorDecode interface

+ Decode func(raw []byte) (T, error)

+ type ProcessorDecodeWithRecord interface

+ DecodeWithRecord func(raw []byte, r *kgo.Record) (T, error)

+ type ProcessorPreCheck interface

+ PreCheck func(ctx context.Context, r *kgo.Record) error

type ProducerHook

+ ProduceHook func(r *Record)

type Record

+ func GetRecord(ctx context.Context) *Record

Oct 20, 2023

Changes in this version

+ var ErrInvalidCompression = fmt.Errorf("invalid compression")

type Config

+ Compression []string

Oct 19, 2023

Changes in this version

+ const KeyRecord

+ var DefaultClientID string

+ var ErrClientClosed = fmt.Errorf("client closed")

+ var ErrNilData = fmt.Errorf("nil data")

+ var ErrNotImplemented = fmt.Errorf("not implemented")

+ var ErrSkip = fmt.Errorf("skip message")

+ type Client struct

+ Consumer consumer

+ Kafka *kgo.Client

+ func NewClient(cfg Config, opts ...Option) (*Client, error)

+ func (c *Client) Admin() *kadm.Client

+ func (c *Client) Close()

+ func (c *Client) Consume(ctx context.Context) error

+ func (c *Client) ProduceRaw(ctx context.Context, records []*kgo.Record) error

+ type Config struct

+ Brokers []string

+ Security SecurityConfig

+ type ConsumeConfig struct

+ Concurrent bool

+ GroupID string

+ MaxPollRecords int

+ Skip map[string]map[int32]Offsets

+ StartOffset int64

+ Topics []string

+ type Header = kgo.RecordHeader

+ type Offsets struct

+ Before int64

+ Offsets []int64

+ type Option func(*options)

+ func WithAutoTopicCreation(v bool) Option

+ func WithClientID(clientID string) Option

+ func WithConsumer(cfg ConsumeConfig, callback func(ctx context.Context, msg T) error, ...) Option

+ func WithKGOOptions(opts ...kgo.Opt) Option

+ type Producer interface

+ Produce func(ctx context.Context, data ...T) error

+ func NewProducer(client *Client, cfg ProducerConfig[T]) (Producer[T], error)

+ type ProducerConfig struct

+ Encode func(T) ([]byte, error)

+ Headers []Header

+ Topic string

+ type ProducerHook interface

+ ProducerHook func(r *Record) *Record

+ type Record = kgo.Record

+ func ContextRecord(ctx context.Context) *Record

+ type SalsConfig struct

+ Plain SaslPlain

+ SCRAM SaslSCRAM

+ func (c SalsConfig) Generate() (sasl.Mechanism, error)

+ type SaslConfigs []SalsConfig

+ func (c SaslConfigs) Generate() ([]sasl.Mechanism, error)

+ type SaslPlain struct

+ Enabled bool

+ Pass string

+ User string

+ Zid string

+ func (s SaslPlain) Generate() (sasl.Mechanism, error)

+ type SaslSCRAM struct

+ Algorithm string

+ Enabled bool

+ IsToken bool

+ Pass string

+ User string

+ Zid string

+ func (s SaslSCRAM) Generate() (sasl.Mechanism, error)

+ type SecurityConfig struct

+ SASL SaslConfigs

+ TLS TLSConfig

+ type TLSConfig struct

+ CAFile string

+ CertFile string

+ Enabled bool

+ KeyFile string

+ func (t TLSConfig) Generate() (*tls.Config, error)