gateway/external/kafka/kafka.go

package kafka

import (
	"context"
	"fmt"

	"git.loafle.net/commons/logging-go"
	occe "git.loafle.net/overflow/commons-go/config/external"
	"github.com/segmentio/kafka-go"
)

var (
	kafkaWriter *kafka.Writer
)

func InitPackage(config *occe.Kafka) {

	kafkaWriter = kafka.NewWriter(kafka.WriterConfig{
		Brokers:  []string{"192.168.1.50:9092"},
		Topic:    "overflow-metric-topic",
		Balancer: &kafka.LeastBytes{},
	})

}

func StartPackage(config *occe.Kafka) {
	if nil == config {
		return
	}

}

func StopPackage(config *occe.Kafka) {
	if nil == config {
		return
	}

}

func DestroyPackage(config *occe.Kafka) {
	if nil == config {
		return
	}

	if err := kafkaWriter.Close(); nil != err {
		logging.Logger().Errorf("%v", err)
	}
}

func Write(key []byte, value []byte) error {
	if nil == kafkaWriter {
		return fmt.Errorf("Kafka client is not valid")
	}

	err := kafkaWriter.WriteMessages(context.Background(),
		kafka.Message{
			Key:   key,
			Value: value,
		},
	)

	return err
}