publisher

Documentation

Index

• type BulkError
  • func (b BulkError) Error() string
• type Client
• type Kafka
  • func NewKafka() (*Kafka, error)
  • func NewKafkaFromClient(client Client, flushInterval int, topicFormat string) *Kafka
  • func (pr *Kafka) Close() int
  • func (pr *Kafka) ProduceBulk(events []*pb.Event, connGroup string, deliveryChannel chan kafka.Event) error
  • func (pr *Kafka) ReportStats()
• type KafkaProducer
• type ProducerStats

Types

type BulkError

type BulkError struct {
	Errors []error
}

func (BulkError) Error

func (b BulkError) Error() string

type Client

type Client interface {
	Produce(*kafka.Message, chan kafka.Event) error
	Close()
	Flush(int) int
	Events() chan kafka.Event
}

type Kafka

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

func NewKafka

func NewKafka() (*Kafka, error)

func NewKafkaFromClient

func NewKafkaFromClient(client Client, flushInterval int, topicFormat string) *Kafka

func (*Kafka) Close

func (pr *Kafka) Close() int

Close wait for outstanding messages to be delivered within given flush interval timeout.

func (*Kafka) ProduceBulk

func (pr *Kafka) ProduceBulk(events []*pb.Event, connGroup string, deliveryChannel chan kafka.Event) error

ProduceBulk messages to kafka. Block until all messages are sent. Return array of error. Order of Errors is guaranteed. DeliveryChannel needs to be exclusive. DeliveryChannel is exposed for recyclability purpose.

func (*Kafka) ReportStats

func (pr *Kafka) ReportStats()

type KafkaProducer

type KafkaProducer interface {
	// ProduceBulk message to kafka. Block until all messages are sent. Return array of error. Order is not guaranteed.
	ProduceBulk(events []*pb.Event, connGroup string, deliveryChannel chan kafka.Event) error
}

KafkaProducer Produce data to kafka synchronously

type ProducerStats

type ProducerStats struct {
	EventCounts map[string]int
	ErrorCounts map[string]int
}