easykafka

README

go-easykafka

easy to use kafka

Documentation

Index

• Variables
• type AckFunc
• type Codec
• type ConsumHandler
  • func (h *ConsumHandler) Cleanup(sarama.ConsumerGroupSession) error
  • func (h *ConsumHandler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
  • func (h *ConsumHandler) Setup(sarama.ConsumerGroupSession) error
• type Consumer
  • func NewConsumer(hosts, topics []string, groupName string, options ...Option) (*Consumer, error)
  • func (c *Consumer) Close() error
  • func (c *Consumer) DeleteGroup() error
  • func (c *Consumer) Group() string
  • func (c *Consumer) Run(handler sarama.ConsumerGroupHandler)
• type ErrFunc
• type Option
• type Producer
  • func NewProducer(hosts []string, options ...Option) (*Producer, error)
  • func (p *Producer) Close() error
  • func (p *Producer) GetAsyncProducer() sarama.AsyncProducer
  • func (p *Producer) Publish(topic string, data interface{}) error
  • func (p *Producer) PublishRawMsg(msg *sarama.ProducerMessage) error
  • func (p *Producer) PublishString(topic, message string) error
  • func (p *Producer) SetCodec(codec Codec)
  • func (p *Producer) SetError(errFunc ErrFunc)
  • func (p *Producer) SetSuccess(ackFunc AckFunc)

Variables

var (
	ErrAlreadyClosed = errors.New("producer already closed")
	ErrGroupNotFound = errors.New("group not found")
)

errors defined

Types

type AckFunc

type AckFunc func(*sarama.ProducerMessage)

type Codec

type Codec interface {
	Marshal(interface{}) ([]byte, error)
}

var DefaultCodec Codec = jsonCodec{}

type ConsumHandler

type ConsumHandler struct {
}

ConsumHandler ...

func (*ConsumHandler) Cleanup

func (h *ConsumHandler) Cleanup(sarama.ConsumerGroupSession) error

Cleanup ..

func (*ConsumHandler) ConsumeClaim

func (h *ConsumHandler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error

ConsumeClaim ..

func (*ConsumHandler) Setup

func (h *ConsumHandler) Setup(sarama.ConsumerGroupSession) error

Setup ..

type Consumer

type Consumer struct {
	Topics    []string
	Reconnect time.Duration
	// contains filtered or unexported fields
}

Consumer ...

func NewConsumer

func NewConsumer(hosts, topics []string, groupName string, options ...Option) (*Consumer, error)

NewConsumer ...

func (*Consumer) Close

func (c *Consumer) Close() error

Close ...

func (*Consumer) DeleteGroup

func (c *Consumer) DeleteGroup() error

func (*Consumer) Group

func (c *Consumer) Group() string

func (*Consumer) Run

func (c *Consumer) Run(handler sarama.ConsumerGroupHandler)

Run ...

type ErrFunc

type ErrFunc func(*sarama.ProducerError)

type Option

type Option func(*sarama.Config)

type Producer

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

Producer ...

func NewProducer

func NewProducer(hosts []string, options ...Option) (*Producer, error)

NewProducer ...

func (*Producer) Close

func (p *Producer) Close() error

Close ...

func (*Producer) GetAsyncProducer

func (p *Producer) GetAsyncProducer() sarama.AsyncProducer

GetAsyncProducer ...

func (*Producer) Publish

func (p *Producer) Publish(topic string, data interface{}) error

Publish ...

func (*Producer) PublishRawMsg

func (p *Producer) PublishRawMsg(msg *sarama.ProducerMessage) error

PublishRawMsg ...

func (*Producer) PublishString

func (p *Producer) PublishString(topic, message string) error

PublishString ...

func (*Producer) SetCodec

func (p *Producer) SetCodec(codec Codec)

SetCodec ...

func (*Producer) SetError

func (p *Producer) SetError(errFunc ErrFunc)

SetError ...

func (*Producer) SetSuccess

func (p *Producer) SetSuccess(ackFunc AckFunc)

SetSuccess ...