kafka

package

v0.3.1 Latest Latest Go to latest Published: Apr 27, 2020 License: GPL-3.0 Imports: 13 Imported by: 2

Details

Package kafka is bp-go-lib's kafka middleware.

This section is empty.

This section is empty.

func DecodeAvroRecord(bytes []byte, obj interface{}) error

func EncodeAvroRecord(obj container.AvroRecord) ([]byte, error)

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

func (bpc *BpConsumer) Consume(topic string, subscribeFunc func(interface{}, interface{})) (err error)

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

func NewKafkaBuilder() *BpKafkaBuilder

func (bpb *BpKafkaBuilder) AddConfig(configMap *kafka.ConfigMap) *BpKafkaBuilder

AddConfig add kafka config by ConfigMap. ConfigMap is a map contaning standard librdkafka configuration properties as documented in: https://github.com/edenhill/librdkafka/tree/master/CONFIGURATION.md

func (bpb *BpKafkaBuilder) BuildConsumer() (*BpConsumer, error)

func (bpb *BpKafkaBuilder) BuildProducer() (*BpProducer, error)

func (bpb *BpKafkaBuilder) SetConfig(configMap *kafka.ConfigMap) *BpKafkaBuilder

SetConfig set kafka config by ConfigMap. ConfigMap is a map contaning standard librdkafka configuration properties as documented in: https://github.com/edenhill/librdkafka/tree/master/CONFIGURATION.md

func (bpb *BpKafkaBuilder) SetGroupId(id string) *BpKafkaBuilder

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

func (bpp *BpProducer) Produce(topic string, key []byte, value []byte) (err error)

Path	Synopsis
avsc Package avsc is bp-go-lib's kafka avsc files path.	Package avsc is bp-go-lib's kafka avsc files path.
record Package record is bp-go-lib's kafka record files path.	Package record is bp-go-lib's kafka record files path.