add wildcard support to kafka

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

pkg/messaging/kafka/pubsub.go

@@ -8,6 +8,7 @@ import (
 	"fmt"
 	"strings"
 	"sync"
+	"time"
 
 	kf "github.com/confluentinc/confluent-kafka-go/v2/kafka"
 	log "github.com/mainflux/mainflux/logger"
@@ -70,18 +71,11 @@ 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
+	s, err := ps.checkTopic(topic, id, ErrAlreadySubscribed)
+	if err != nil {
+		return err
 	}
+	ps.configReader(id, topic, s, handler)
 
 	consumer, err := kf.NewConsumer(&kf.ConfigMap{
 		"bootstrap.servers":        ps.url,