kafka

May 13, 2023

Changes in this version

type Conf

+ Password string

+ Username string

type PushOption

+ func WithAuth(username, password string) PushOption

Aug 5, 2022

Changes in this version

+ func WithSyncCall() queue.CallOptions

Jul 20, 2022

Changes in this version

+ const RequireAll

+ const RequireNone

+ const RequireOne

type PushOption

+ func WithBatchBytes(batchBytes int64) PushOption

+ func WithBatchSize(batchSize int) PushOption

+ func WithRequiredAcks(requiredAcks RequiredAcks) PushOption

+ type RequiredAcks = kafka.RequiredAcks

Jul 19, 2022

Changes in this version

type PushOption

+ func WithDisableAutoTopicCreation() PushOption

Jul 19, 2022

Changes in this version

type Pusher

+ func (p *Pusher) Start()

+ func (p *Pusher) Stop()

+ type XXHashBalancer struct

+ func (x *XXHashBalancer) Balance(msg kafka.Message, partitions ...int) (partition int)

Jul 12, 2022

Changes in this version

+ type Balancer = kafka.Balancer

+ type BalancerFunc = kafka.BalancerFunc

+ type CRC32Balancer = kafka.CRC32Balancer

+ type Hash = kafka.Hash

+ type LeastBytes = kafka.LeastBytes

+ type Murmur2Balancer = kafka.Murmur2Balancer

type PushOption

+ func WithBalancer(balancer Balancer) PushOption

+ type ReferenceHash = kafka.ReferenceHash

+ type RoundRobin = kafka.RoundRobin

Jul 4, 2022

Changes in this version

type PushOption

+ func WithCompletion(completion func(messages []kafka.Message, err error)) PushOption

Jun 28, 2022

Changes in this version

+ type Headers struct

+ func (h *Headers) Get(key string) string

+ func (h *Headers) Keys() []string

+ func (h *Headers) Set(key string, value string)

Jun 27, 2022

Changes in this version

+ func HeadersFromContext(ctx context.Context, headers ...kafka.Header) ([]kafka.Header, bool)

+ func NewHeadersContext(ctx context.Context, headers ...kafka.Header) context.Context

+ func WithSync() queue.CallOptions

+ type Conf struct

+ Brokers []string

+ Conns int

+ Consumers int

+ Group string

+ MaxBytes int

+ MinBytes int

+ Offset string

+ Processors int

+ Topic string

+ type HeaderKey struct

+ type PushOption func(options *chunkOptions)

+ func WithChunkSize(chunkSize int) PushOption

+ func WithFlushInterval(interval time.Duration) PushOption

+ type Pusher struct

+ func NewPusher(addrs []string, topic string, opts ...PushOption) *Pusher

+ func (p *Pusher) Close() error

+ func (p *Pusher) Name() string

+ func (p *Pusher) Push(ctx context.Context, k, v []byte, opts ...queue.CallOptions) (interface{}, error)

+ type QueueOption func(*queueOptions)

+ func WithCommitInterval(interval time.Duration) QueueOption

+ func WithMaxWait(wait time.Duration) QueueOption

+ func WithMetrics(metrics *stat.Metrics) QueueOption

+ func WithQueueCapacity(queueCapacity int) QueueOption

+ type Queues struct

+ func MustNewQueue(c Conf, handler queue.Consumer, opts ...QueueOption) *Queues

+ func NewQueue(c Conf, handler queue.Consumer, opts ...QueueOption) (*Queues, error)

+ func (q Queues) Start()

+ func (q Queues) Stop()