kafka

Documentation

Index

• func HeadersFromContext(ctx context.Context) ([]kafka.Header, bool)
• func NewHeadersContext(ctx context.Context, headers ...kafka.Header) context.Context
• func WithSync() queue.CallOptions
• type Balancer
• type BalancerFunc
• type CRC32Balancer
• type Conf
• type Hash
• type Headers
  • func (h *Headers) Get(key string) string
  • func (h *Headers) Keys() []string
  • func (h *Headers) Set(key string, value string)
• type LeastBytes
• type Murmur2Balancer
• type PushOption
  • func WithBalancer(balancer Balancer) PushOption
  • func WithChunkSize(chunkSize int) PushOption
  • func WithCompletion(completion func(messages []kafka.Message, err error)) PushOption
  • func WithDisableAutoTopicCreation() PushOption
  • func WithFlushInterval(interval time.Duration) PushOption
• type Pusher
  • 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)
  • func (p *Pusher) Start()
  • func (p *Pusher) Stop()
• type QueueOption
  • 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
  • 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()
• type ReferenceHash
• type RoundRobin
• type XXHashBalancer
  • func (x *XXHashBalancer) Balance(msg kafka.Message, partitions ...int) (partition int)

Functions

func HeadersFromContext

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

func NewHeadersContext

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

func WithSync

func WithSync() queue.CallOptions

Types

type Balancer

type Balancer = kafka.Balancer

type BalancerFunc

type BalancerFunc = kafka.BalancerFunc

type CRC32Balancer

type CRC32Balancer = kafka.CRC32Balancer

type Conf

type Conf struct {
	Brokers    []string
	Group      string
	Topic      string
	Offset     string `json:",options=first|last,default=last"`
	Conns      int    `json:",default=1"`
	Consumers  int    `json:",default=8"`
	Processors int    `json:",default=8"`
	MinBytes   int    `json:",default=10240"`    // 10K
	MaxBytes   int    `json:",default=10485760"` // 10M
}

type Hash

type Hash = kafka.Hash

type Headers

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

func (*Headers) Get

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

func (*Headers) Keys

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

func (*Headers) Set

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

type LeastBytes

type LeastBytes = kafka.LeastBytes

type Murmur2Balancer

type Murmur2Balancer = kafka.Murmur2Balancer

type PushOption

type PushOption func(options *pushOptions)

func WithBalancer

func WithBalancer(balancer Balancer) PushOption

func WithChunkSize

func WithChunkSize(chunkSize int) PushOption

func WithCompletion

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

func WithDisableAutoTopicCreation

func WithDisableAutoTopicCreation() PushOption

func WithFlushInterval

func WithFlushInterval(interval time.Duration) PushOption

type Pusher

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

func NewPusher

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

func (*Pusher) Close

func (p *Pusher) Close() error

func (*Pusher) Name

func (p *Pusher) Name() string

func (*Pusher) Push

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

func (*Pusher) Start

func (p *Pusher) Start()

func (*Pusher) Stop

func (p *Pusher) Stop()

type QueueOption

type QueueOption func(*queueOptions)

func WithCommitInterval

func WithCommitInterval(interval time.Duration) QueueOption

func WithMaxWait

func WithMaxWait(wait time.Duration) QueueOption

func WithMetrics

func WithMetrics(metrics *stat.Metrics) QueueOption

func WithQueueCapacity

func WithQueueCapacity(queueCapacity int) QueueOption

type Queues

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

func MustNewQueue

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

func NewQueue

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

func (Queues) Start

func (q Queues) Start()

func (Queues) Stop

func (q Queues) Stop()

type ReferenceHash

type ReferenceHash = kafka.ReferenceHash

type RoundRobin

type RoundRobin = kafka.RoundRobin

type XXHashBalancer

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

func (*XXHashBalancer) Balance

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