kafka

Documentation

Index

• func NewConfig() *config
• func NewConnection(c *config) (*kafka.Conn, error)
• func WriteOnce(topic string, key, value []byte) error
• type Reader
  • func NewReader(topic string) (Reader, error)
• type Writer
  • func NewWriter(topic string) (Writer, error)

Functions

func NewConfig

func NewConfig() *config

func NewConnection

func NewConnection(c *config) (*kafka.Conn, error)

func WriteOnce

func WriteOnce(topic string, key, value []byte) error

Types

type Reader

type Reader interface {
	Topic() string
	ReadLag() (totalLag int64, err error)
	ReadAll(stoppable bool, fn func([]byte))
	Shutdown()
}

func NewReader

func NewReader(topic string) (Reader, error)

type Writer

type Writer interface {
	Topic() string
	Write(key []byte, value []byte) error
	Shutdown()
}

func NewWriter

func NewWriter(topic string) (Writer, error)