use NATS mqtt broker

Signed-off-by: Arvindh <arvindh91@gmail.com>

Author: arvindh123

cmd/mqtt/main.go

@@ -26,7 +26,6 @@ import (
 	smqlog "github.com/absmach/supermq/logger"
 	"github.com/absmach/supermq/mqtt"
 	"github.com/absmach/supermq/mqtt/events"
-	mqtttracing "github.com/absmach/supermq/mqtt/tracing"
 	domainsAuthz "github.com/absmach/supermq/pkg/domains/grpcclient"
 	"github.com/absmach/supermq/pkg/errors"
 	"github.com/absmach/supermq/pkg/grpcclient"
@@ -36,7 +35,6 @@ import (
 	brokerstracing "github.com/absmach/supermq/pkg/messaging/brokers/tracing"
 	msgevents "github.com/absmach/supermq/pkg/messaging/events"
 	"github.com/absmach/supermq/pkg/messaging/handler"
-	mqttpub "github.com/absmach/supermq/pkg/messaging/mqtt"
 	"github.com/absmach/supermq/pkg/server"
 	"github.com/absmach/supermq/pkg/uuid"
 	"github.com/caarlos0/env/v11"
@@ -135,31 +133,6 @@ func main() {
 	}()
 	tracer := tp.Tracer(svcName)
 
-	bsub, err := brokers.NewPubSub(ctx, cfg.BrokerURL, logger)
-	if err != nil {
-		logger.Error(fmt.Sprintf("failed to connect to message broker: %s", err))
-		exitCode = 1
-		return
-	}
-	defer bsub.Close()
-	bsub = brokerstracing.NewPubSub(serverConfig, tracer, bsub)
-
-	mpub, err := mqttpub.NewPublisher(fmt.Sprintf("mqtt://%s:%s", cfg.MQTTTargetHost, cfg.MQTTTargetPort), cfg.MQTTTargetUsername, cfg.MQTTTargetPassword, cfg.MQTTQoS, cfg.MQTTForwarderTimeout)
-	if err != nil {
-		logger.Error(fmt.Sprintf("failed to create MQTT publisher: %s", err))
-		exitCode = 1
-		return
-	}
-	defer mpub.Close()
-
-	fwd := mqtt.NewForwarder(brokers.SubjectAllMessages, logger)
-	fwd = mqtttracing.New(serverConfig, tracer, fwd, brokers.SubjectAllMessages)
-	if err := fwd.Forward(ctx, svcName, bsub, mpub); err != nil {
-		logger.Error(fmt.Sprintf("failed to forward message broker messages: %s", err))
-		exitCode = 1
-		return
-	}
-
 	np, err := brokers.NewPublisher(ctx, cfg.BrokerURL)
 	if err != nil {
 		logger.Error(fmt.Sprintf("failed to connect to message broker: %s", err))
@@ -251,9 +224,7 @@ func main() {
 		go chc.CallHome(ctx)
 	}
 
-	beforeHandler := beforeHandler{
-		resolver: messaging.NewTopicResolver(channelsClient, domainsClient),
-	}
+	beforeHandler := mqtt.NewBeforeHandler(messaging.NewTopicResolver(channelsClient, domainsClient), parser, logger)
 
 	afterHandler := afterHandler{
 		username: cfg.MQTTTargetUsername,
@@ -382,42 +353,3 @@ func (ah afterHandler) Intercept(ctx context.Context, pkt packets.ControlPacket,
 
 	return pkt, nil
 }
-
-type beforeHandler struct {
-	resolver messaging.TopicResolver
-}
-
-// This interceptor is used to replace domain and channel routes with relevant domain and channel IDs in the message topic.
-func (bh beforeHandler) Intercept(ctx context.Context, pkt packets.ControlPacket, dir session.Direction) (packets.ControlPacket, error) {
-	switch pt := pkt.(type) {
-	case *packets.SubscribePacket:
-		for i, topic := range pt.Topics {
-			ft, err := bh.resolver.ResolveTopic(ctx, topic)
-			if err != nil {
-				return nil, err
-			}
-			pt.Topics[i] = ft
-		}
-
-		return pt, nil
-	case *packets.UnsubscribePacket:
-		for i, topic := range pt.Topics {
-			ft, err := bh.resolver.ResolveTopic(ctx, topic)
-			if err != nil {
-				return nil, err
-			}
-			pt.Topics[i] = ft
-		}
-		return pt, nil
-	case *packets.PublishPacket:
-		ft, err := bh.resolver.ResolveTopic(ctx, pt.TopicName)
-		if err != nil {
-			return nil, err
-		}
-		pt.TopicName = ft
-
-		return pt, nil
-	}
-
-	return pkt, nil
-}

docker/.env

@@ -40,19 +40,19 @@ SMQ_MESSAGE_BROKER_TYPE=msg_nats
 SMQ_MESSAGE_BROKER_URL=${SMQ_NATS_URL}
 
 ## MQTT Broker
-SMQ_MQTT_BROKER_TYPE=rabbitmq
-SMQ_MQTT_BROKER_HEALTH_CHECK=
-SMQ_MQTT_ADAPTER_MQTT_QOS=${SMQ_RABBITMQ_MQTT_QOS}
+SMQ_MQTT_BROKER_TYPE=nats
+SMQ_MQTT_BROKER_HEALTH_CHECK=${SMQ_NATS_HEALTH_CHECK}
+SMQ_MQTT_ADAPTER_MQTT_QOS=${SMQ_NATS_MQTT_QOS}
 SMQ_MQTT_ADAPTER_MQTT_TARGET_PROTOCOL=mqtt
 SMQ_MQTT_ADAPTER_MQTT_TARGET_HOST=${SMQ_MQTT_BROKER_TYPE}
 SMQ_MQTT_ADAPTER_MQTT_TARGET_PORT=1883
-SMQ_MQTT_ADAPTER_MQTT_TARGET_USERNAME=${SMQ_RABBITMQ_USER}
-SMQ_MQTT_ADAPTER_MQTT_TARGET_PASSWORD=${SMQ_RABBITMQ_PASS}
-SMQ_MQTT_ADAPTER_MQTT_TARGET_HEALTH_CHECK=${SMQ_MQTT_BROKER_HEALTH_CHECK}
+SMQ_MQTT_ADAPTER_MQTT_TARGET_USERNAME=
+SMQ_MQTT_ADAPTER_MQTT_TARGET_PASSWORD=
+SMQ_MQTT_ADAPTER_MQTT_TARGET_HEALTH_CHECK=${SMQ_NATS_HEALTH_CHECK}
 SMQ_MQTT_ADAPTER_WS_TARGET_PROTOCOL=http
 SMQ_MQTT_ADAPTER_WS_TARGET_HOST=${SMQ_MQTT_BROKER_TYPE}
-SMQ_MQTT_ADAPTER_WS_TARGET_PORT=${SMQ_RABBITMQ_WS_PORT}
-SMQ_MQTT_ADAPTER_WS_TARGET_PATH=${SMQ_RABBITMQ_WS_TARGET_PATH}
+SMQ_MQTT_ADAPTER_WS_TARGET_PORT=8080
+SMQ_MQTT_ADAPTER_WS_TARGET_PATH=${SMQ_NATS_WS_TARGET_PATH}
 
 ## Redis
 SMQ_REDIS_TCP_PORT=6379

docker/docker-compose.yaml

@@ -1129,7 +1129,6 @@ services:
     container_name: supermq-mqtt
     depends_on:
       - clients
-      - rabbitmq
       - nats
     restart: on-failure
     environment:
@@ -1565,27 +1564,6 @@ services:
         bind:
           create_host_path: true
 
-  rabbitmq:
-    image: rabbitmq:4.0.5-management-alpine
-    container_name: supermq-rabbitmq
-    restart: on-failure
-    environment:
-      RABBITMQ_ERLANG_COOKIE: ${SMQ_RABBITMQ_COOKIE}
-      RABBITMQ_DEFAULT_USER: ${SMQ_RABBITMQ_USER}
-      RABBITMQ_DEFAULT_PASS: ${SMQ_RABBITMQ_PASS}
-      RABBITMQ_DEFAULT_VHOST: ${SMQ_RABBITMQ_VHOST}
-      RABBITMQ_CONFIG_FILES: /etc/rabbitmq/conf.d/
-    ports:
-      - ${SMQ_RABBITMQ_PORT}:${SMQ_RABBITMQ_PORT}
-      - ${SMQ_RABBITMQ_HTTP_PORT}:${SMQ_RABBITMQ_HTTP_PORT}
-      - ${SMQ_RABBITMQ_WS_PORT}:${SMQ_RABBITMQ_WS_PORT}
-    volumes:
-      - ./rabbitmq/enabled_plugins:/etc/rabbitmq/enabled_plugins
-      - ./rabbitmq/rabbitmq.conf:/etc/rabbitmq/conf.d/10-defaults.conf
-      - supermq-mqtt-broker-volume:/var/lib/rabbitmq
-    networks:
-      - supermq-base-net
-
   nats:
     image: nats:2.10.25-alpine
     container_name: supermq-nats

mqtt/forwarder.go

@@ -8,7 +8,6 @@ import (
 	"fmt"
 	"log/slog"
 	"strings"
-	"time"
 
 	"github.com/absmach/mgate/pkg/session"
 	grpcChannelsV1 "github.com/absmach/supermq/api/grpc/channels/v1"
@@ -158,44 +157,11 @@ func (h *handler) Connect(ctx context.Context) error {
 
 // Publish - after client successfully published.
 func (h *handler) Publish(ctx context.Context, topic *string, payload *[]byte) error {
-	s, ok := session.FromContext(ctx)
-	if !ok {
-		return errors.Wrap(ErrFailedPublish, ErrClientNotInitialized)
-	}
-	h.logger.Info(fmt.Sprintf(LogInfoPublished, s.ID, *topic))
-
-	domainID, chanID, subTopic, topicType, err := h.parser.ParsePublishTopic(ctx, *topic, false)
-	if err != nil {
-		return errors.Wrap(ErrFailedPublish, err)
-	}
-
-	msg := messaging.Message{
-		Protocol:  protocol,
-		Domain:    domainID,
-		Channel:   chanID,
-		Subtopic:  subTopic,
-		Publisher: s.Username,
-		Payload:   *payload,
-		Created:   time.Now().UnixNano(),
-	}
-
-	if topicType == messaging.MessageType {
-		if err := h.publisher.Publish(ctx, messaging.EncodeMessageTopic(&msg), &msg); err != nil {
-			return errors.Wrap(ErrFailedPublishToMsgBroker, err)
-		}
-	}
-
 	return nil
 }
 
 // Subscribe - after client successfully subscribed.
 func (h *handler) Subscribe(ctx context.Context, topics *[]string) error {
-	s, ok := session.FromContext(ctx)
-	if !ok {
-		return errors.Wrap(ErrFailedSubscribe, ErrClientNotInitialized)
-	}
-	h.logger.Info(fmt.Sprintf(LogInfoSubscribed, s.ID, strings.Join(*topics, ",")))
-
 	return nil
 }
 

mqtt/handler.go

@@ -0,0 +1,101 @@
+package mqtt
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"time"
+
+	"github.com/absmach/mgate/pkg/session"
+	"github.com/absmach/supermq/pkg/errors"
+	"github.com/absmach/supermq/pkg/messaging"
+	"github.com/eclipse/paho.mqtt.golang/packets"
+	"google.golang.org/protobuf/proto"
+)
+
+type beforeHandler struct {
+	resolver messaging.TopicResolver
+	parser   messaging.TopicParser
+	logger   *slog.Logger
+}
+
+func NewBeforeHandler(resolver messaging.TopicResolver, parser messaging.TopicParser, logger *slog.Logger) session.Interceptor {
+	return &beforeHandler{
+		resolver: resolver,
+		parser:   parser,
+		logger:   logger,
+	}
+}
+
+// This interceptor is used to replace domain and channel routes with relevant domain and channel IDs in the message topic.
+func (bh beforeHandler) Intercept(ctx context.Context, pkt packets.ControlPacket, dir session.Direction) (packets.ControlPacket, error) {
+	switch pt := pkt.(type) {
+	case *packets.SubscribePacket:
+		for i, topic := range pt.Topics {
+			ft, err := bh.resolver.ResolveTopic(ctx, topic)
+			if err != nil {
+				return nil, err
+			}
+			pt.Topics[i] = ft
+		}
+
+		return pt, nil
+	case *packets.UnsubscribePacket:
+		for i, topic := range pt.Topics {
+			ft, err := bh.resolver.ResolveTopic(ctx, topic)
+			if err != nil {
+				return nil, err
+			}
+			pt.Topics[i] = ft
+		}
+		return pt, nil
+	case *packets.PublishPacket:
+		ft, err := bh.resolver.ResolveTopic(ctx, pt.TopicName)
+		if err != nil {
+			return nil, err
+		}
+		pt.TopicName = ft
+
+		s, ok := session.FromContext(ctx)
+		if !ok {
+			return pt, errors.Wrap(ErrFailedPublish, ErrClientNotInitialized)
+		}
+		bh.logger.Info(fmt.Sprintf(LogInfoPublished, s.ID, ft))
+
+		switch dir {
+		case session.Up:
+			domainID, chanID, subTopic, _, err := bh.parser.ParsePublishTopic(ctx, ft, false)
+			if err != nil {
+				return pt, errors.Wrap(ErrFailedPublish, err)
+			}
+
+			msg := &messaging.Message{
+				Protocol:  "mqtt",
+				Domain:    domainID,
+				Channel:   chanID,
+				Subtopic:  subTopic,
+				Publisher: s.Username,
+				Payload:   pt.Payload,
+				Created:   time.Now().UnixNano(),
+			}
+
+			data, err := proto.Marshal(msg)
+			if err != nil {
+				return pt, err
+			}
+			pt.Payload = data
+
+		case session.Down:
+			var msg messaging.Message
+
+			if err := proto.Unmarshal(pt.Payload, &msg); err != nil {
+				return pt, errors.Wrap(ErrFailedPublish, err)
+			}
+			pt.Payload = msg.GetPayload()
+		}
+
+		return pt, nil
+	}
+
+	return pkt, nil
+}