kafka

Documentation

Index

• Constants
• func HeadersFromContext(ctx context.Context) ([]kafka.Header, bool)
• func NewHeadersContext(ctx context.Context, headers ...kafka.Header) context.Context
• func WithSyncCall() 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 WithAuth(username, password string) PushOption
  • func WithBalancer(balancer Balancer) PushOption
  • func WithBatchBytes(batchBytes int64) PushOption
  • func WithBatchSize(batchSize int) PushOption
  • func WithChunkSize(chunkSize int) PushOption
  • func WithCompletion(completion func(messages []kafka.Message, err error)) PushOption
  • func WithDisableAutoTopicCreation() PushOption
  • func WithFlushInterval(flushInterval time.Duration) PushOption
  • func WithRequiredAcks(requiredAcks RequiredAcks) 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 RequiredAcks
• type RoundRobin
• type XXHashBalancer
  • func (x *XXHashBalancer) Balance(msg kafka.Message, partitions ...int) (partition int)

Constants

const (
	RequireNone = kafka.RequireNone
	RequireOne  = kafka.RequireOne
	RequireAll  = kafka.RequireAll
)

Functions

func HeadersFromContext

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

func NewHeadersContext

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

func WithSyncCall

func WithSyncCall() 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
	Username   string `json:",optional"`
	Password   string `json:",optional"`
}

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 WithAuth

func WithAuth(username, password string) PushOption

func WithBalancer

func WithBalancer(balancer Balancer) PushOption

func WithBatchBytes

func WithBatchBytes(batchBytes int64) PushOption

func WithBatchSize

func WithBatchSize(batchSize int) 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(flushInterval time.Duration) PushOption

func WithRequiredAcks

func WithRequiredAcks(requiredAcks RequiredAcks) 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 RequiredAcks

type RequiredAcks = kafka.RequiredAcks

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)