kafka

Documentation

Index

• type ConsumerInfo
• type KafkaConnector
  • func NewKafkaConnector(brokers []string) *KafkaConnector
  • func (c *KafkaConnector) NewConsumer(topic string, partition int32, onMessage func(data []byte))
  • func (c *KafkaConnector) NewProducer(topic string) (*sarama.SyncProducer, error)
  • func (c *KafkaConnector) TerminateConsumer(topic string, partition int32)

Types

type ConsumerInfo

type ConsumerInfo struct {
	Signal    chan os.Signal
	Partition int32
	Consumer  sarama.Consumer
}

type KafkaConnector

type KafkaConnector struct {
	Brokers   []string
	Config    sarama.Config
	Producers map[string]sarama.SyncProducer
	Consumers map[string][]ConsumerInfo
}

func NewKafkaConnector

func NewKafkaConnector(brokers []string) *KafkaConnector

func (*KafkaConnector) NewConsumer

func (c *KafkaConnector) NewConsumer(topic string, partition int32, onMessage func(data []byte))

func (*KafkaConnector) NewProducer

func (c *KafkaConnector) NewProducer(topic string) (*sarama.SyncProducer, error)

func (*KafkaConnector) TerminateConsumer

func (c *KafkaConnector) TerminateConsumer(topic string, partition int32)