kafka

Documentation

Index

• func FeatureLogToKafka(kafkaName string, f FeatureLogKafkaFunc)
• func Load(config *recconf.RecommendConfig)
• type FeatureLogKafkaFunc
• type KafkaLog
  • func (l *KafkaLog) Errorf(msg string, args ...interface{})
• type KafkaProducer
  • func GetKafkaProducer(name string) (*KafkaProducer, error)
  • func NewKafkaProducer(bootstrapServers, topic string) *KafkaProducer
  • func (k *KafkaProducer) Close()
  • func (k *KafkaProducer) Init() error
  • func (k *KafkaProducer) SendMessage(message []byte)
  • func (k *KafkaProducer) SendMessages(module []byte, message []byte)

Functions

func FeatureLogToKafka

func FeatureLogToKafka(kafkaName string, f FeatureLogKafkaFunc)

func Load

func Load(config *recconf.RecommendConfig)

Types

type FeatureLogKafkaFunc

type FeatureLogKafkaFunc func(*KafkaProducer, *module.User, []*module.Item, *context.RecommendContext)

type KafkaLog

type KafkaLog struct {
}

func (*KafkaLog) Errorf

func (l *KafkaLog) Errorf(msg string, args ...interface{})

type KafkaProducer

type KafkaProducer struct {
	BootstrapServers string
	Topic            string
	Producer         *kafka.Writer
}

func GetKafkaProducer

func GetKafkaProducer(name string) (*KafkaProducer, error)

func NewKafkaProducer

func NewKafkaProducer(bootstrapServers, topic string) *KafkaProducer

func (*KafkaProducer) Close

func (k *KafkaProducer) Close()

func (*KafkaProducer) Init

func (k *KafkaProducer) Init() error

func (*KafkaProducer) SendMessage

func (k *KafkaProducer) SendMessage(message []byte)

func (*KafkaProducer) SendMessages

func (k *KafkaProducer) SendMessages(module []byte, message []byte)