easykafka

README

go-easykafka

easy to use kafka

Documentation

Index

• Variables
• 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()
  • func (c *Consumer) Run()
  • func (c *Consumer) SetHandler(h sarama.ConsumerGroupHandler)
• type Option
• type Producer
  • func NewProducer(hosts []string, options ...Option) (*Producer, error)
  • func (p *Producer) Close() error
  • func (p *Producer) Publish(topic string, data interface{}) error
  • func (p *Producer) PublishRaw(msg *sarama.ProducerMessage) error
  • func (p *Producer) PublishString(topic, message string) error
  • func (p *Producer) SetCodec(codec Codec)

Variables

var (
	ErrAlreadyClosed = errors.New("producer already closed")
)

errors defined

Types

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 {
	// 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()

Close ...

func (*Consumer) Run

func (c *Consumer) Run()

Run ...

func (*Consumer) SetHandler

func (c *Consumer) SetHandler(h sarama.ConsumerGroupHandler)

SetHandler ...

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) Publish

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

Publish ...

func (*Producer) PublishRaw

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

PublishRaw ...

func (*Producer) PublishString

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

PublishString ...

func (*Producer) SetCodec

func (p *Producer) SetCodec(codec Codec)

SetCodec ...