kafka

Documentation

Index

• func DividePartitions(partitions, splitBy int) [][]int
• type Client
  • func (c *Client) NewConsumer(name, topic string, startingOffsets map[int32]int64) *Consumer
  • func (c *Client) SendBytes(bts []byte, topic, key string) error
  • func (c *Client) SendBytesAsync(bts []byte, topic, key string)
• type Consumer
  • func NewConsumer(hosts []string, clientName, topic string, startingOffsets map[int32]int64) *Consumer
  • func (c *Consumer) Events() <-chan *ConsumerEvent
  • func (c *Consumer) Run()
  • func (c *Consumer) Shutdown()
• type ConsumerChannel
• type ConsumerEvent
• type Producer
• type Sender

Functions

func DividePartitions

func DividePartitions(partitions, splitBy int) [][]int

DividePartitions - splits the number of partitions in to buckets numbering the splitBy

Types

type Client

type Client struct {
	Hosts []string
	Log   logging.Logger `inject:""`
	// contains filtered or unexported fields
}

func (*Client) NewConsumer

func (c *Client) NewConsumer(name, topic string, startingOffsets map[int32]int64) *Consumer

func (*Client) SendBytes

func (c *Client) SendBytes(bts []byte, topic, key string) error

func (*Client) SendBytesAsync

func (c *Client) SendBytesAsync(bts []byte, topic, key string)

type Consumer

type Consumer struct {
	ClientName      string
	StartingOffsets map[int32]int64
	Log             logging.Logger `inject:""`
	Topic           string
	// contains filtered or unexported fields
}

func NewConsumer

func NewConsumer(hosts []string, clientName, topic string, startingOffsets map[int32]int64) *Consumer

func (*Consumer) Events

func (c *Consumer) Events() <-chan *ConsumerEvent

func (*Consumer) Run

func (c *Consumer) Run()

Run - Nonblocking

func (*Consumer) Shutdown

func (c *Consumer) Shutdown()

type ConsumerChannel

type ConsumerChannel interface {
	Events() <-chan *ConsumerEvent
	Close() error
}

type ConsumerEvent

type ConsumerEvent struct {
	Error     error
	Message   []byte
	Offset    int64
	Partition int32
}

type Producer

type Producer interface {
	Close()
	SendBytes(message []byte, topic, key string) error
}

type Sender

type Sender interface {
	SendBytes(message []byte, topic, key string) error
}