Merge branch 'topic_polling' into NOISSUE-kafka-support

Signed-off-by: rodneyosodo <[email protected]>

Author: rodneyosodo

go.mod

@@ -5,7 +5,6 @@ go 1.19
 require (
 	github.com/caarlos0/env/v7 v7.0.0
 	github.com/cenkalti/backoff/v4 v4.1.3
-	github.com/confluentinc/confluent-kafka-go/v2 v2.0.2
 	github.com/docker/docker v20.10.21+incompatible
 	github.com/eclipse/paho.mqtt.golang v1.4.2
 	github.com/fatih/color v1.13.0
@@ -39,6 +38,7 @@ require (
 	github.com/prometheus/client_golang v1.13.0
 	github.com/rabbitmq/amqp091-go v1.5.0
 	github.com/rubenv/sql-migrate v1.2.0
+	github.com/segmentio/kafka-go v0.4.38
 	github.com/spf13/cobra v1.6.1
 	github.com/spf13/viper v1.13.0
 	github.com/stretchr/testify v1.8.1
@@ -102,6 +102,7 @@ require (
 	github.com/jackc/pgio v1.0.0 // indirect
 	github.com/jackc/pgpassfile v1.0.0 // indirect
 	github.com/jackc/pgservicefile v0.0.0-20200714003250-2b9c44734f2b // indirect
+	github.com/jhump/protoreflect v1.12.0 // indirect
 	github.com/klauspost/compress v1.15.11 // indirect
 	github.com/magiconair/properties v1.8.6 // indirect
 	github.com/mattn/go-colorable v0.1.13 // indirect
@@ -121,6 +122,7 @@ require (
 	github.com/opencontainers/runc v1.1.4 // indirect
 	github.com/pelletier/go-toml/v2 v2.0.5 // indirect
 	github.com/pierrec/lz4 v2.6.1+incompatible // indirect
+	github.com/pierrec/lz4/v4 v4.1.15 // indirect
 	github.com/pion/dtls/v2 v2.1.5 // indirect
 	github.com/pion/logging v0.2.2 // indirect
 	github.com/pion/transport v0.13.1 // indirect

go.sum

@@ -4,27 +4,41 @@
 package kafka
 
 import (
+	"context"
 	"fmt"
+	"sync"
+	"time"
 
-	kf "github.com/confluentinc/confluent-kafka-go/v2/kafka"
 	"github.com/mainflux/mainflux/pkg/messaging"
+	"github.com/segmentio/kafka-go"
 	"google.golang.org/protobuf/proto"
 )
 
 var _ messaging.Publisher = (*publisher)(nil)
 
+var (
+	numPartitions     = 1
+	replicationFactor = 1
+	batchTimeout      = time.Microsecond
+)
+
 type publisher struct {
-	prod *kf.Producer
+	url    string
+	conn   *kafka.Conn
+	mu     sync.Mutex
+	topics map[string]*kafka.Writer
 }
 
 // NewPublisher returns Kafka message Publisher.
 func NewPublisher(url string) (messaging.Publisher, error) {
-	prod, err := kf.NewProducer(&kf.ConfigMap{"bootstrap.servers": url})
+	conn, err := kafka.Dial("tcp", url)
 	if err != nil {
 		return &publisher{}, err
 	}
 	ret := &publisher{
-		prod: prod,
+		url:    url,
+		conn:   conn,
+		topics: make(map[string]*kafka.Writer),
 	}
 	return ret, nil
 
@@ -43,20 +57,66 @@ func (pub *publisher) Publish(topic string, msg *messaging.Message) error {
 		subject = fmt.Sprintf("%s.%s", subject, msg.Subtopic)
 	}
 
-	kafkaMsg := kf.Message{
-		TopicPartition: kf.TopicPartition{Topic: &subject},
-		Value:          data,
+	kafkaMsg := kafka.Message{
+		Value: data,
 	}
 
-	if err := pub.prod.Produce(&kafkaMsg, nil); err != nil {
-		return err
+	writer, ok := pub.topics[subject]
+	if ok {
+		if err := writer.WriteMessages(context.Background(), kafkaMsg); err != nil {
+			return err
+		}
+		return nil
 	}
 
-	pub.prod.Flush(1 * 1000)
+	topicConfigs := []kafka.TopicConfig{
+		{
+			Topic:             subject,
+			NumPartitions:     numPartitions,
+			ReplicationFactor: replicationFactor,
+		},
+	}
+	if err := pub.conn.CreateTopics(topicConfigs...); err != nil {
+		return err
+	}
+	writer = &kafka.Writer{
+		Addr:                   kafka.TCP(pub.url),
+		Topic:                  subject,
+		RequiredAcks:           kafka.RequireAll,
+		Balancer:               &kafka.LeastBytes{},
+		BatchTimeout:           batchTimeout,
+		AllowAutoTopicCreation: true,
+	}
+	if err := writer.WriteMessages(context.Background(), kafkaMsg); err != nil {
+		return err
+	}
+	pub.mu.Lock()
+	defer pub.mu.Unlock()
+	pub.topics[subject] = writer
 	return nil
 }
 
 func (pub *publisher) Close() error {
-	pub.prod.Close()
+	defer pub.conn.Close()
+
+	pub.mu.Lock()
+	defer pub.mu.Unlock()
+
+	topics := make([]string, 0, len(pub.topics))
+	for topic := range pub.topics {
+		topics = append(topics, topic)
+		pub.topics[topic].Close()
+	}
+
+	req := &kafka.DeleteTopicsRequest{
+		Addr:   kafka.TCP(pub.url),
+		Topics: topics,
+	}
+	client := kafka.Client{
+		Addr: kafka.TCP(pub.url),
+	}
+	if _, err := client.DeleteTopics(context.Background(), req); err != nil {
+		return err
+	}
 	return nil
 }

pkg/messaging/kafka/publisher.go

@@ -4,19 +4,26 @@
 package kafka
 
 import (
+	"context"
 	"errors"
 	"fmt"
-	"strings"
+	"regexp"
 	"sync"
+	"time"
 
-	kf "github.com/confluentinc/confluent-kafka-go/v2/kafka"
 	log "github.com/mainflux/mainflux/logger"
+
 	"github.com/mainflux/mainflux/pkg/messaging"
+	kafka "github.com/segmentio/kafka-go"
+	ktopics "github.com/segmentio/kafka-go/topics"
 	"google.golang.org/protobuf/proto"
 )
 
 const (
-	chansPrefix = "channels"
+	chansPrefix               = "channels"
+	SubjectAllChannels        = "channels.*"
+	offset                    = kafka.LastOffset
+	defaultScanningIntervalMS = 500
 )
 
 var (
@@ -29,31 +36,33 @@ var (
 var _ messaging.PubSub = (*pubsub)(nil)
 
 type subscription struct {
-	*kf.Consumer
+	*kafka.Reader
 	cancel func() error
 }
 type pubsub struct {
 	publisher
-	url           string
-	queue         string
+	client        *kafka.Client
 	logger        log.Logger
 	mu            sync.Mutex
 	subscriptions map[string]map[string]subscription
 }
 
 // NewPubSub returns Kafka message publisher/subscriber.
-func NewPubSub(url, queue string, logger log.Logger) (messaging.PubSub, error) {
-	prod, err := kf.NewProducer(&kf.ConfigMap{"bootstrap.servers": url})
+func NewPubSub(url, _ string, logger log.Logger) (messaging.PubSub, error) {
+	conn, err := kafka.Dial("tcp", url)
 	if err != nil {
 		return &pubsub{}, err
 	}
-
+	client := &kafka.Client{
+		Addr: conn.LocalAddr(),
+	}
 	ret := &pubsub{
 		publisher: publisher{
-			prod: prod,
+			url:    url,
+			conn:   conn,
+			topics: make(map[string]*kafka.Writer),
 		},
-		url:           url,
-		queue:         queue,
+		client:        client,
 		subscriptions: make(map[string]map[string]subscription),
 		logger:        logger,
 	}
@@ -70,49 +79,26 @@ func (ps *pubsub) Subscribe(id, topic string, handler messaging.MessageHandler)
 	ps.mu.Lock()
 	defer ps.mu.Unlock()
 
-	// Check topic
-	s, ok := ps.subscriptions[topic]
-	switch ok {
-	case true:
-		// Check topic ID
-		if _, ok := s[id]; ok {
-			return ErrAlreadySubscribed
-		}
-	default:
-		s = make(map[string]subscription)
-		ps.subscriptions[topic] = s
-	}
-
-	consumer, err := kf.NewConsumer(&kf.ConfigMap{
-		"bootstrap.servers":        ps.url,
-		"broker.address.family":    "v4",
-		"group.id":                 "mainflux",
-		"auto.offset.reset":        "latest",
-		"metadata.max.age.ms":      1,
-		"allow.auto.create.topics": "true",
-	})
+	s, err := ps.checkTopic(topic, id, ErrAlreadySubscribed)
 	if err != nil {
 		return err
 	}
-	if err = consumer.SubscribeTopics([]string{formatTopic(topic)}, nil); err != nil {
-		return err
-	}
-
-	go func() {
-		for {
-			message, err := consumer.ReadMessage(-1)
-			ps.handle(message, handler)
-			if err == nil {
-				ps.handle(message, handler)
-			} else if !err.(kf.Error).IsTimeout() {
-				ps.logger.Error(err.Error())
-				return
+	ps.configReader(id, topic, s, handler)
+
+	// Subscribe to all topic by prediocially scanning for all topics and consuming them
+	if topic == SubjectAllChannels {
+		go func() {
+			for {
+				topics, _ := ps.listTopic()
+				for _, t := range topics {
+					s, err := ps.checkTopic(t, id, ErrAlreadySubscribed)
+					if err == nil {
+						ps.configReader(id, t, s, handler)
+					}
+				}
+				time.Sleep(defaultScanningIntervalMS * time.Millisecond)
 			}
-		}
-	}()
-	s[id] = subscription{
-		Consumer: consumer,
-		cancel:   handler.Cancel,
+		}()
 	}
 	return nil
 }
@@ -157,12 +143,12 @@ func (ps *pubsub) Close() error {
 	return nil
 }
 
-func (ps *pubsub) handle(message *kf.Message, h messaging.MessageHandler) {
-	var msg = &messaging.Message{}
-	if err := proto.Unmarshal(message.Value, msg); err != nil {
+func (ps *pubsub) handle(message kafka.Message, h messaging.MessageHandler) {
+	var msg messaging.Message
+	if err := proto.Unmarshal(message.Value, &msg); err != nil {
 		ps.logger.Warn(fmt.Sprintf("Failed to unmarshal received message: %s", err))
 	}
-	if err := h.Handle(msg); err != nil {
+	if err := h.Handle(&msg); err != nil {
 		ps.logger.Warn(fmt.Sprintf("Failed to handle Mainflux message: %s", err))
 	}
 }
@@ -176,9 +162,53 @@ func (s subscription) close() error {
 	return s.Close()
 }
 
-func formatTopic(topic string) string {
-	if strings.Contains(topic, "*") {
-		return fmt.Sprintf("^%s", topic)
+func (ps *pubsub) listTopic() ([]string, error) {
+	allRegex := regexp.MustCompile(SubjectAllChannels)
+	allTopics, err := ktopics.ListRe(context.Background(), ps.client, allRegex)
+	if err != nil {
+		return []string{}, err
+	}
+	var topics []string
+	for _, t := range allTopics {
+		topics = append(topics, t.Name)
+	}
+	return topics, nil
+}
+
+func (ps *pubsub) checkTopic(topic, id string, err error) (map[string]subscription, error) {
+	// Check topic
+	s, ok := ps.subscriptions[topic]
+	switch ok {
+	case true:
+		// Check topic ID
+		if _, ok := s[id]; ok {
+			return map[string]subscription{}, err
+		}
+	default:
+		s = make(map[string]subscription)
+		ps.subscriptions[topic] = s
+	}
+	return s, nil
+}
+
+func (ps *pubsub) configReader(id, topic string, s map[string]subscription, handler messaging.MessageHandler) {
+	reader := kafka.NewReader(kafka.ReaderConfig{
+		Brokers:     []string{ps.url},
+		GroupID:     id,
+		Topic:       topic,
+		StartOffset: offset,
+	})
+	go func() {
+		for {
+			message, err := reader.ReadMessage(context.Background())
+			if err != nil {
+				break
+			}
+			ps.handle(message, handler)
+		}
+	}()
+	s[id] = subscription{
+		Reader: reader,
+		cancel: handler.Cancel,
 	}
-	return topic
 }