kafka

Documentation

Index

• Constants
• type KafkaConfig
• type KafkaConsumer
  • func NewMQConsumer(confPath string, topicModels []*mq.ConsumerTopicModel, logPath string) *KafkaConsumer
  • func (mq *KafkaConsumer) Close() error
  • func (mq *KafkaConsumer) Register(callback mq.IConsumerCallback) error
  • func (mq *KafkaConsumer) RegisterByTopic(topic string, callback mq.IConsumerCallback) error
  • func (mq *KafkaConsumer) Start() error
• type KafkaProducer
  • func NewMQProducer(confPath string, topics []string, logPath string) *KafkaProducer
  • func (mq *KafkaProducer) Close() error
  • func (mq *KafkaProducer) Produce(topic string, msg string) error
  • func (mq *KafkaProducer) Start() error

Constants

const (
	GLOBAL_TOPIC_CALLBACK_KEY = "kafka_iot_default_topic"
)

Types

type KafkaConfig

type KafkaConfig struct {
	Version    string   `toml:"kafka_version"`
	ClientID   string   `toml:"client_id"`
	BrokerList []string `toml:"broker_list"`
}

config

type KafkaConsumer

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

func NewMQConsumer

func NewMQConsumer(confPath string, topicModels []*mq.ConsumerTopicModel, logPath string) *KafkaConsumer

func (*KafkaConsumer) Close

func (mq *KafkaConsumer) Close() error

func (*KafkaConsumer) Register

func (mq *KafkaConsumer) Register(callback mq.IConsumerCallback) error

func (*KafkaConsumer) RegisterByTopic

func (mq *KafkaConsumer) RegisterByTopic(topic string, callback mq.IConsumerCallback) error

func (*KafkaConsumer) Start

func (mq *KafkaConsumer) Start() error

type KafkaProducer

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

func NewMQProducer

func NewMQProducer(confPath string, topics []string, logPath string) *KafkaProducer

func (*KafkaProducer) Close

func (mq *KafkaProducer) Close() error

func (*KafkaProducer) Produce

func (mq *KafkaProducer) Produce(topic string, msg string) error

func (*KafkaProducer) Start

func (mq *KafkaProducer) Start() error