kafka

Documentation

Index

• Variables
• type Kafka
  • func New(clientid string, logger *log.Logger, zkpeers []string) (*Kafka, error)
  • func (k *Kafka) Close()
  • func (k *Kafka) Incoming() <-chan Message
  • func (k *Kafka) Listen(topic string, partition int32, offset int64) error
  • func (k *Kafka) ListenNewest(topic string, partition int32) error
  • func (k *Kafka) ListenOldest(topic string, partition int32) error
  • func (k *Kafka) Outgoing() <-chan transmithandler.Transmit
  • func (k *Kafka) Send(key, val []byte, topic string) (partition int32, offset int64, err error)
  • func (k *Kafka) Unlisten(topic string, partition int32) error
• type Message

Variables

var ErrNoBrokers = errors.New("No kafka brokers found")

var ErrTransmitHandlerClosed = errors.New("Transmit handler has closed")

Types

type Kafka

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

func New

func New(clientid string, logger *log.Logger, zkpeers []string) (*Kafka, error)

func (*Kafka) Close

func (k *Kafka) Close()

func (*Kafka) Incoming

func (k *Kafka) Incoming() <-chan Message

func (*Kafka) Listen

func (k *Kafka) Listen(topic string, partition int32, offset int64) error

func (*Kafka) ListenNewest

func (k *Kafka) ListenNewest(topic string, partition int32) error

func (*Kafka) ListenOldest

func (k *Kafka) ListenOldest(topic string, partition int32) error

func (*Kafka) Outgoing

func (k *Kafka) Outgoing() <-chan transmithandler.Transmit

func (*Kafka) Send

func (k *Kafka) Send(key, val []byte, topic string) (partition int32, offset int64, err error)

func (*Kafka) Unlisten

func (k *Kafka) Unlisten(topic string, partition int32) error

type Message

type Message struct {
	Key       []byte
	Val       []byte
	Topic     string
	Partition int32
	Offset    int64
}