kafka

Documentation

Index

• func NewConsumerGroupHandler(queue *chan KafkaMessage) 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 KafkaMessage) (err error)
  • func (c *Client) GSub(ctx context.Context, cncl context.CancelFunc)
  • func (c *Client) GetMessages() <-chan KafkaMessage
  • 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 KafkaMessage
  • func ToKafkaMessage(topic string, message interface{}) (KafkaMessage, error)
• type NewClientOptions

Functions

func NewConsumerGroupHandler

func NewConsumerGroupHandler(queue *chan KafkaMessage) 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 KafkaMessage) (err error)

func (*Client) GSub

func (c *Client) GSub(ctx context.Context, cncl context.CancelFunc)

func (*Client) GetMessages

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

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 KafkaMessage

type KafkaMessage struct {
	Topic string
	Value []byte
}

func ToKafkaMessage

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

type NewClientOptions

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