kafka

Documentation

Index

• func GetKafkaStats() (send uint64, received uint64)
• func NewConsumerGroupHandler(queue *chan Message) sarama.ConsumerGroupHandler
• type Client
  • func NewKafkaClient(opts NewClientOptions) (client *Client, err error)
  • func (c *Client) ChangeSubscribedTopics(newTopicRegex *regexp.Regexp)
  • func (c *Client) Close() error
  • func (c *Client) EnqueueMessage(msg Message) (err error)
  • func (c *Client) GetMessages() <-chan Message
  • func (c *Client) Ready() bool
• type ConsumerGroupHandler
  • func (c ConsumerGroupHandler) Cleanup(session sarama.ConsumerGroupSession) error
  • func (c ConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
  • func (c ConsumerGroupHandler) Setup(session sarama.ConsumerGroupSession) error
• type Message
  • func ToKafkaMessage(topic string, message interface{}) (Message, error)
• type NewClientOptions

Functions

func GetKafkaStats

func GetKafkaStats() (send uint64, received uint64)

func NewConsumerGroupHandler

func NewConsumerGroupHandler(queue *chan Message) sarama.ConsumerGroupHandler

Types

type Client

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

func NewKafkaClient

func NewKafkaClient(opts NewClientOptions) (client *Client, err error)

func (*Client) ChangeSubscribedTopics

func (c *Client) ChangeSubscribedTopics(newTopicRegex *regexp.Regexp)

func (*Client) Close

func (c *Client) Close() error

func (*Client) EnqueueMessage

func (c *Client) EnqueueMessage(msg Message) (err error)

func (*Client) GetMessages

func (c *Client) GetMessages() <-chan Message

func (*Client) Ready

func (c *Client) Ready() bool

type ConsumerGroupHandler

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

func (ConsumerGroupHandler) Cleanup

func (c ConsumerGroupHandler) Cleanup(session sarama.ConsumerGroupSession) error

func (ConsumerGroupHandler) ConsumeClaim

func (c ConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error

func (ConsumerGroupHandler) Setup

func (c ConsumerGroupHandler) Setup(session sarama.ConsumerGroupSession) error

type Message

type Message struct {
	Topic  string
	Value  []byte
	Header map[string][]byte
	Key    []byte
}

func ToKafkaMessage

func ToKafkaMessage(topic string, message interface{}) (Message, error)

type NewClientOptions

type NewClientOptions struct {
	ListenTopicRegex  *regexp.Regexp
	ConsumerName      string
	Brokers           []string
	StartOffset       int64
	Partitions        int32
	ReplicationFactor int16
	EnableTLS         bool
}