kafkax

Documentation

Index

• func TCP(address ...string) net.Addr
• type Config
  • func NewConfig() *Config
• type Consumer
  • func NewConsumer(topic, group string, config *Config) (*Consumer, error)
  • func (c *Consumer) Close() error
  • func (c *Consumer) Dequeue(handler mx.Handler) error
• type Message
  • func (m *Message) Message() kafka.Message
  • func (m *Message) Topic() string
  • func (m *Message) Value() []byte
• type Producer
  • func NewProducer(topic string, config *Config) (*Producer, error)
  • func (p *Producer) Close() error
  • func (p *Producer) Enqueue(data []byte) error
  • func (p *Producer) EnqueueMessage(m kafka.Message) error
  • func (p *Producer) EnqueueMessages(m ...kafka.Message) error
  • func (p *Producer) MultiEnqueue(data ...[]byte) error
• type Queue
  • func NewQueue(topic string, config *Config) (*Queue, error)
  • func (q *Queue) Close() error
  • func (q *Queue) Dequeue(group string, handler mx.Handler) error
  • func (q *Queue) Enqueue(data []byte) error
  • func (q *Queue) MultiEnqueue(data ...[]byte) error

Functions

func TCP

func TCP(address ...string) net.Addr

Types

type Config

type Config struct {
	Reader kafka.ReaderConfig
	Writer kafka.Writer
}

func NewConfig

func NewConfig() *Config

type Consumer

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

func NewConsumer

func NewConsumer(topic, group string, config *Config) (*Consumer, error)

func (*Consumer) Close

func (c *Consumer) Close() error

func (*Consumer) Dequeue

func (c *Consumer) Dequeue(handler mx.Handler) error

type Message

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

func (*Message) Message

func (m *Message) Message() kafka.Message

func (*Message) Topic

func (m *Message) Topic() string

func (*Message) Value

func (m *Message) Value() []byte

type Producer

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

func NewProducer

func NewProducer(topic string, config *Config) (*Producer, error)

func (*Producer) Close

func (p *Producer) Close() error

func (*Producer) Enqueue

func (p *Producer) Enqueue(data []byte) error

func (*Producer) EnqueueMessage

func (p *Producer) EnqueueMessage(m kafka.Message) error

func (*Producer) EnqueueMessages

func (p *Producer) EnqueueMessages(m ...kafka.Message) error

func (*Producer) MultiEnqueue

func (p *Producer) MultiEnqueue(data ...[]byte) error

type Queue

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

func NewQueue

func NewQueue(topic string, config *Config) (*Queue, error)

func (*Queue) Close

func (q *Queue) Close() error

func (*Queue) Dequeue

func (q *Queue) Dequeue(group string, handler mx.Handler) error

func (*Queue) Enqueue

func (q *Queue) Enqueue(data []byte) error

func (*Queue) MultiEnqueue

func (q *Queue) MultiEnqueue(data ...[]byte) error