kafka

Documentation

Index

• Variables
• func NewClient(addresses []string) (sarama.Client, error)
• type BaseEncoder
  • func NewBaseEncoder(typ string) *BaseEncoder
  • func (e *BaseEncoder) Encode(message proto.Message) ([]*sarama.ProducerMessage, error)
  • func (e *BaseEncoder) HandleError(msg *sarama.ProducerError)
  • func (e *BaseEncoder) HandleSuccess(msg *sarama.ProducerMessage)
• type MessageEncoder
• type Metadata

Variables

var Addresses = flag.String("kafkaaddrs", "localhost:9092", "kafka's comma-separated addresses")

Addresses is the flag for kafka's comma-separated addresses

var Topic = flag.String("kafkatopic", filepath.Base(os.Args[0]), "kafka's topic")

Topic is the flag for kafka's topic

Functions

func NewClient

func NewClient(addresses []string) (sarama.Client, error)

NewClient returns a Kafka client

Types

type BaseEncoder

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

BaseEncoder implements MessageEncoder interface and mainly handle monitoring

func NewBaseEncoder

func NewBaseEncoder(typ string) *BaseEncoder

NewBaseEncoder returns a new base MessageEncoder

func (*BaseEncoder) Encode

func (e *BaseEncoder) Encode(message proto.Message) ([]*sarama.ProducerMessage,
	error)

Encode encodes the proto message to a sarama.ProducerMessage

func (*BaseEncoder) HandleError

func (e *BaseEncoder) HandleError(msg *sarama.ProducerError)

HandleError process the metadata of messages from kafka producer Errors channel

func (*BaseEncoder) HandleSuccess

func (e *BaseEncoder) HandleSuccess(msg *sarama.ProducerMessage)

HandleSuccess process the metadata of messages from kafka producer Successes channel

type MessageEncoder

type MessageEncoder interface {
	Encode(proto.Message) ([]*sarama.ProducerMessage, error)
	HandleSuccess(*sarama.ProducerMessage)
	HandleError(*sarama.ProducerError)
}

MessageEncoder is an encoder interface which handles encoding proto.Message to sarama.ProducerMessage

type Metadata

type Metadata struct {
	StartTime   time.Time
	NumMessages int
}

Metadata is used to store metadata for the sarama.ProducerMessages