Kafka

- add producer

go.mod

@@ -5,6 +5,7 @@ go 1.20
 require (
 	github.com/ClickHouse/clickhouse-go/v2 v2.13.3
 	github.com/Nerzal/gocloak/v13 v13.8.0
+	github.com/confluentinc/confluent-kafka-go/v2 v2.2.0
 	github.com/gofrs/uuid v4.4.0+incompatible
 	github.com/golang-jwt/jwt/v4 v4.5.0
 	github.com/google/uuid v1.3.1

kafka/config.go

@@ -0,0 +1,17 @@
+package kafka
+
+import (
+	"fmt"
+)
+
+type Config struct {
+	Hosts []string `yaml:"hosts"`
+}
+
+func (c Config) Invalidate() error {
+	if len(c.Hosts) < 1 {
+		return fmt.Errorf("at least one bootstrap server / host must be provided")
+	}
+
+	return nil
+}

kafka/iproducer.go

@@ -0,0 +1,11 @@
+package kafka
+
+import (
+	"git.beejay.kim/Craft/Api/service"
+	lib "github.com/confluentinc/confluent-kafka-go/v2/kafka"
+)
+
+type Producer interface {
+	service.Service
+	Produce(*lib.Message) error
+}

kafka/producer.go

@@ -0,0 +1,83 @@
+package kafka
+
+import (
+	"fmt"
+	lib "github.com/confluentinc/confluent-kafka-go/v2/kafka"
+	"strings"
+	"sync"
+	"time"
+)
+
+type _producer struct {
+	config  *Config
+	session *lib.Producer
+}
+
+func New(cfg *Config) (Producer, error) {
+	var (
+		p = &_producer{
+			config: cfg,
+		}
+		err error
+	)
+
+	if cfg == nil {
+		return nil, fmt.Errorf("config must be provided")
+	}
+
+	opts := &lib.ConfigMap{
+		"bootstrap.servers": strings.Join(p.config.Hosts, ","),
+	}
+
+	if p.session, err = lib.NewProducer(opts); err != nil {
+		return nil, err
+	}
+
+	return p, nil
+}
+
+func (p *_producer) Produce(message *lib.Message) error {
+	var (
+		delivery = make(chan lib.Event)
+		err      error
+	)
+
+	if message.TopicPartition.Topic == nil {
+		return fmt.Errorf("illegal message: at least one topic must be provided for topic partition")
+	}
+
+	if err = p.session.Produce(message, delivery); err != nil {
+		return err
+	}
+
+	e := <-delivery
+	m := e.(*lib.Message)
+
+	if m.TopicPartition.Error != nil {
+		return fmt.Errorf("delivery failed: %v", m.TopicPartition.Error)
+	}
+
+	close(delivery)
+	return nil
+}
+
+func (p *_producer) String() string {
+	return "kafka:producer"
+}
+
+func (p *_producer) Close() error {
+	var wg sync.WaitGroup
+	wg.Add(1)
+
+	go func() {
+		defer wg.Done()
+
+		if p.session != nil {
+			time.Sleep(time.Second)
+			p.session.Close() //nolint:errcheck
+		}
+	}()
+
+	wg.Wait()
+	return nil
+}