SMQ-2888 - Change internal message topic format (#2889)

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

Author: arvindh123

cmd/mqtt/main.go

@@ -148,8 +148,8 @@ func main() {
 	}
 	defer mpub.Close()
 
-	fwd := mqtt.NewForwarder(brokers.SubjectAllChannels, logger)
-	fwd = mqtttracing.New(serverConfig, tracer, fwd, brokers.SubjectAllChannels)
+	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

coap/README.md

@@ -76,5 +76,5 @@ Setting `SMQ_CLIENTS_GRPC_CLIENT_CERT` and `SMQ_CLIENTS_GRPC_CLIENT_KEY` will en
 
 ## Usage
 
-If CoAP adapter is running locally (on default 5683 port), a valid URL would be: `coap://localhost/channels/<channel_id>/messages?auth=<client_auth_key>`.
+If CoAP adapter is running locally (on default 5683 port), a valid URL would be: `coap://localhost/m/<domain_id>/c/<channel_id>/<subtopic>?auth=<client_auth_key>`.
 Since CoAP protocol does not support `Authorization` header (option) and options have limited size, in order to send CoAP messages, valid `auth` value (a valid Client key) must be present in `Uri-Query` option.

coap/adapter.go

@@ -8,7 +8,6 @@ package coap
 
 import (
 	"context"
-	"fmt"
 
 	grpcChannelsV1 "github.com/absmach/supermq/api/grpc/channels/v1"
 	grpcClientsV1 "github.com/absmach/supermq/api/grpc/clients/v1"
@@ -21,8 +20,6 @@ import (
 
 var errFailedToDisconnectClient = errors.New("failed to disconnect client")
 
-const chansPrefix = "channels"
-
 // Service specifies CoAP service API.
 type Service interface {
 	// Publish publishes message to specified channel.
@@ -37,7 +34,7 @@ type Service interface {
 	Unsubscribe(ctx context.Context, key, domainID, chanID, subptopic, token string) error
 
 	// DisconnectHandler method is used to disconnected the client
-	DisconnectHandler(ctx context.Context, chanID, subptopic, token string) error
+	DisconnectHandler(ctx context.Context, domainID, chanID, subptopic, token string) error
 }
 
 var _ Service = (*adapterService)(nil)
@@ -87,7 +84,7 @@ func (svc *adapterService) Publish(ctx context.Context, key string, msg *messagi
 
 	msg.Publisher = authnRes.GetId()
 
-	return svc.pubsub.Publish(ctx, msg.GetChannel(), msg)
+	return svc.pubsub.Publish(ctx, messaging.EncodeMessageTopic(msg), msg)
 }
 
 func (svc *adapterService) Subscribe(ctx context.Context, key, domainID, chanID, subtopic string, c Client) error {
@@ -116,11 +113,7 @@ func (svc *adapterService) Subscribe(ctx context.Context, key, domainID, chanID,
 		return svcerr.ErrAuthorization
 	}
 
-	subject := fmt.Sprintf("%s.%s", chansPrefix, chanID)
-	if subtopic != "" {
-		subject = fmt.Sprintf("%s.%s", subject, subtopic)
-	}
-
+	subject := messaging.EncodeTopic(domainID, chanID, subtopic)
 	authzc := newAuthzClient(clientID, domainID, chanID, subtopic, svc.channels, c)
 	subCfg := messaging.SubscriberConfig{
 		ID:       c.Token(),
@@ -156,19 +149,13 @@ func (svc *adapterService) Unsubscribe(ctx context.Context, key, domainID, chanI
 		return svcerr.ErrAuthorization
 	}
 
-	subject := fmt.Sprintf("%s.%s", chansPrefix, chanID)
-	if subtopic != "" {
-		subject = fmt.Sprintf("%s.%s", subject, subtopic)
-	}
+	subject := messaging.EncodeTopic(domainID, chanID, subtopic)
 
 	return svc.pubsub.Unsubscribe(ctx, token, subject)
 }
 
-func (svc *adapterService) DisconnectHandler(ctx context.Context, chanID, subtopic, token string) error {
-	subject := fmt.Sprintf("%s.%s", chansPrefix, chanID)
-	if subtopic != "" {
-		subject = fmt.Sprintf("%s.%s", subject, subtopic)
-	}
+func (svc *adapterService) DisconnectHandler(ctx context.Context, domainID, chanID, subtopic, token string) error {
+	subject := messaging.EncodeTopic(domainID, chanID, subtopic)
 
 	return svc.pubsub.Unsubscribe(ctx, token, subject)
 }

coap/api/logging.go

@@ -97,10 +97,11 @@ func (lm *loggingMiddleware) Unsubscribe(ctx context.Context, key, domainID, cha
 
 // DisconnectHandler logs the disconnect handler. It logs the channel ID, subtopic (if any) and the time it took to complete the request.
 // If the request fails, it logs the error.
-func (lm *loggingMiddleware) DisconnectHandler(ctx context.Context, chanID, subtopic, token string) (err error) {
+func (lm *loggingMiddleware) DisconnectHandler(ctx context.Context, domainID, chanID, subtopic, token string) (err error) {
 	defer func(begin time.Time) {
 		args := []any{
 			slog.String("duration", time.Since(begin).String()),
+			slog.String("domain_id", domainID),
 			slog.String("channel_id", chanID),
 			slog.String("token", token),
 		}
@@ -115,5 +116,5 @@ func (lm *loggingMiddleware) DisconnectHandler(ctx context.Context, chanID, subt
 		lm.logger.Info("Unsubscribe completed successfully", args...)
 	}(time.Now())
 
-	return lm.svc.DisconnectHandler(ctx, chanID, subtopic, token)
+	return lm.svc.DisconnectHandler(ctx, domainID, chanID, subtopic, token)
 }

coap/api/metrics.go

@@ -62,11 +62,11 @@ func (mm *metricsMiddleware) Unsubscribe(ctx context.Context, key, domainID, cha
 }
 
 // DisconnectHandler instruments DisconnectHandler method with metrics.
-func (mm *metricsMiddleware) DisconnectHandler(ctx context.Context, chanID, subtopic, token string) error {
+func (mm *metricsMiddleware) DisconnectHandler(ctx context.Context, domainID, chanID, subtopic, token string) error {
 	defer func(begin time.Time) {
 		mm.counter.With("method", "disconnect_handler").Add(1)
 		mm.latency.With("method", "disconnect_handler").Observe(time.Since(begin).Seconds())
 	}(time.Now())
 
-	return mm.svc.DisconnectHandler(ctx, chanID, subtopic, token)
+	return mm.svc.DisconnectHandler(ctx, domainID, chanID, subtopic, token)
 }

coap/api/transport.go

@@ -9,8 +9,6 @@ import (
 	"io"
 	"log/slog"
 	"net/http"
-	"net/url"
-	"regexp"
 	"strings"
 	"time"
 
@@ -33,18 +31,9 @@ const (
 	startObserve = 0 // observe option value that indicates start of observation
 )
 
-var channelPartRegExp = regexp.MustCompile(`^/m/([\w\-]+)/c/([\w\-]+)(/[^?]*)?(\?.*)?$`)
-
-const (
-	numGroups    = 4 // entire expression+ domain group + channel group + subtopic group
-	domainGroup  = 1 // domain group is first in channel regexp
-	channelGroup = 3 // channel group is third in channel regexp
-)
-
 var (
-	errMalformedSubtopic = errors.New("malformed subtopic")
-	errBadOptions        = errors.New("bad options")
-	errMethodNotAllowed  = errors.New("method not allowed")
+	errBadOptions       = errors.New("bad options")
+	errMethodNotAllowed = errors.New("method not allowed")
 )
 
 var (
@@ -133,7 +122,7 @@ func handleGet(m *mux.Message, w mux.ResponseWriter, msg *messaging.Message, key
 	if obs == startObserve {
 		c := coap.NewClient(w.Conn(), m.Token(), logger)
 		w.Conn().AddOnClose(func() {
-			_ = service.DisconnectHandler(context.Background(), msg.GetChannel(), msg.GetSubtopic(), c.Token())
+			_ = service.DisconnectHandler(context.Background(), msg.GetDomain(), msg.GetChannel(), msg.GetSubtopic(), c.Token())
 		})
 		return service.Subscribe(w.Conn().Context(), key, msg.GetDomain(), msg.GetChannel(), msg.GetSubtopic(), c)
 	}
@@ -148,20 +137,23 @@ func decodeMessage(msg *mux.Message) (*messaging.Message, error) {
 	if err != nil {
 		return &messaging.Message{}, err
 	}
-	channelParts := channelPartRegExp.FindStringSubmatch(path)
-	if len(channelParts) < numGroups {
-		return &messaging.Message{}, errMalformedSubtopic
-	}
 
-	st, err := parseSubtopic(channelParts[channelGroup])
+	var domainID, channelID, subTopic string
+	switch msg.Code() {
+	case codes.GET:
+		domainID, channelID, subTopic, err = messaging.ParseSubscribeTopic(path)
+	case codes.POST:
+		domainID, channelID, subTopic, err = messaging.ParsePublishTopic(path)
+	}
 	if err != nil {
 		return &messaging.Message{}, err
 	}
+
 	ret := &messaging.Message{
 		Protocol: protocol,
-		Domain:   channelParts[domainGroup],
-		Channel:  channelParts[2],
-		Subtopic: st,
+		Domain:   domainID,
+		Channel:  channelID,
+		Subtopic: subTopic,
 		Payload:  []byte{},
 		Created:  time.Now().UnixNano(),
 	}
@@ -187,32 +179,3 @@ func parseKey(msg *mux.Message) (string, error) {
 	}
 	return vars[1], nil
 }
-
-func parseSubtopic(subtopic string) (string, error) {
-	if subtopic == "" {
-		return subtopic, nil
-	}
-
-	subtopic, err := url.QueryUnescape(subtopic)
-	if err != nil {
-		return "", errMalformedSubtopic
-	}
-	subtopic = strings.ReplaceAll(subtopic, "/", ".")
-
-	elems := strings.Split(subtopic, ".")
-	filteredElems := []string{}
-	for _, elem := range elems {
-		if elem == "" {
-			continue
-		}
-
-		if len(elem) > 1 && (strings.Contains(elem, "*") || strings.Contains(elem, ">")) {
-			return "", errMalformedSubtopic
-		}
-
-		filteredElems = append(filteredElems, elem)
-	}
-
-	subtopic = strings.Join(filteredElems, ".")
-	return subtopic, nil
-}

coap/tracing/adapter.go

@@ -66,11 +66,12 @@ func (tm *tracingServiceMiddleware) Unsubscribe(ctx context.Context, key, domain
 }
 
 // DisconnectHandler traces a CoAP disconnect operation.
-func (tm *tracingServiceMiddleware) DisconnectHandler(ctx context.Context, chanID, subptopic, token string) error {
+func (tm *tracingServiceMiddleware) DisconnectHandler(ctx context.Context, domainID, chanID, subptopic, token string) error {
 	ctx, span := tm.tracer.Start(ctx, disconnectHandlerOp, trace.WithAttributes(
+		attribute.String("domain_id", domainID),
 		attribute.String("channel_id", chanID),
 		attribute.String("subtopic", subptopic),
 	))
 	defer span.End()
-	return tm.svc.DisconnectHandler(ctx, chanID, subptopic, token)
+	return tm.svc.DisconnectHandler(ctx, domainID, chanID, subptopic, token)
 }

consumers/messages.go

@@ -123,7 +123,7 @@ type config struct {
 func loadConfig(configPath string) (config, error) {
 	cfg := config{
 		SubscriberCfg: subscriberConfig{
-			Subjects: []string{brokers.SubjectAllChannels},
+			Subjects: []string{brokers.SubjectAllMessages},
 		},
 		TransformerCfg: transformerConfig{
 			Format:      defFormat,

http/api/endpoint_test.go

@@ -28,6 +28,7 @@ import (
 	smqauthn "github.com/absmach/supermq/pkg/authn"
 	authnMocks "github.com/absmach/supermq/pkg/authn/mocks"
 	"github.com/absmach/supermq/pkg/connections"
+	"github.com/absmach/supermq/pkg/messaging"
 	pubsub "github.com/absmach/supermq/pkg/messaging/mocks"
 	"github.com/absmach/supermq/pkg/policies"
 	"github.com/stretchr/testify/assert"
@@ -257,7 +258,7 @@ func TestPublish(t *testing.T) {
 				ClientType: policies.ClientType,
 				Type:       uint32(connections.Publish),
 			}).Return(tc.authzRes, tc.authzErr)
-			svcCall := pub.On("Publish", mock.Anything, tc.chanID, mock.Anything).Return(nil)
+			svcCall := pub.On("Publish", mock.Anything, messaging.EncodeTopicSuffix(tc.domainID, tc.chanID, ""), mock.Anything).Return(nil)
 			req := testRequest{
 				client:      ts.Client(),
 				method:      http.MethodPost,

http/handler.go

@@ -8,8 +8,6 @@ import (
 	"fmt"
 	"log/slog"
 	"net/http"
-	"net/url"
-	"regexp"
 	"strings"
 	"time"
 
@@ -49,14 +47,10 @@ var (
 	errClientNotInitialized     = errors.New("client is not initialized")
 	errFailedPublish            = errors.New("failed to publish")
 	errFailedPublishToMsgBroker = errors.New("failed to publish to supermq message broker")
-	errMalformedSubtopic        = mgate.NewHTTPProxyError(http.StatusBadRequest, errors.New("malformed subtopic"))
 	errMalformedTopic           = mgate.NewHTTPProxyError(http.StatusBadRequest, errors.New("malformed topic"))
 	errMissingTopicPub          = mgate.NewHTTPProxyError(http.StatusBadRequest, errors.New("failed to publish due to missing topic"))
-	errFailedParseSubtopic      = mgate.NewHTTPProxyError(http.StatusBadRequest, errors.New("failed to parse subtopic"))
 )
 
-var channelRegExp = regexp.MustCompile(`^\/?m\/([\w\-]+)\/c\/([\w\-]+)(\/[^?]*)?(\?.*)?$`)
-
 // Event implements events.Event interface.
 type handler struct {
 	publisher messaging.Publisher
@@ -125,6 +119,11 @@ func (h *handler) Publish(ctx context.Context, topic *string, payload *[]byte) e
 		return errors.Wrap(errFailedPublish, errClientNotInitialized)
 	}
 
+	domainID, chanID, subtopic, err := messaging.ParsePublishTopic(*topic)
+	if err != nil {
+		return errors.Wrap(errMalformedTopic, err)
+	}
+
 	var clientID, clientType string
 	switch {
 	case strings.HasPrefix(string(s.Password), "Client"):
@@ -153,11 +152,6 @@ func (h *handler) Publish(ctx context.Context, topic *string, payload *[]byte) e
 		return mgate.NewHTTPProxyError(http.StatusUnauthorized, svcerr.ErrAuthentication)
 	}
 
-	domainID, chanID, subtopic, err := parseTopic(*topic)
-	if err != nil {
-		return mgate.NewHTTPProxyError(http.StatusBadRequest, err)
-	}
-
 	msg := messaging.Message{
 		Protocol: protocol,
 		Domain:   domainID,
@@ -186,7 +180,7 @@ func (h *handler) Publish(ctx context.Context, topic *string, payload *[]byte) e
 		msg.Publisher = clientID
 	}
 
-	if err := h.publisher.Publish(ctx, msg.Channel, &msg); err != nil {
+	if err := h.publisher.Publish(ctx, messaging.EncodeMessageTopic(&msg), &msg); err != nil {
 		return errors.Wrap(errFailedPublishToMsgBroker, err)
 	}
 
@@ -209,52 +203,3 @@ func (h *handler) Unsubscribe(ctx context.Context, topics *[]string) error {
 func (h *handler) Disconnect(ctx context.Context) error {
 	return nil
 }
-
-func parseTopic(topic string) (string, string, string, error) {
-	// Topics are in the format:
-	// m/<domain_id>/c/<channel_id>/<subtopic>/.../ct/<content_type>
-	channelParts := channelRegExp.FindStringSubmatch(topic)
-	if len(channelParts) < 3 {
-		return "", "", "", errors.Wrap(errFailedPublish, errMalformedTopic)
-	}
-
-	domainID := channelParts[1]
-	chanID := channelParts[2]
-	subtopic := channelParts[3]
-
-	subtopic, err := parseSubtopic(subtopic)
-	if err != nil {
-		return "", "", "", errors.Wrap(errFailedParseSubtopic, err)
-	}
-
-	return domainID, chanID, subtopic, nil
-}
-
-func parseSubtopic(subtopic string) (string, error) {
-	if subtopic == "" {
-		return subtopic, nil
-	}
-
-	subtopic, err := url.QueryUnescape(subtopic)
-	if err != nil {
-		return "", mgate.NewHTTPProxyError(http.StatusBadRequest, errMalformedSubtopic)
-	}
-	subtopic = strings.ReplaceAll(subtopic, "/", ".")
-
-	elems := strings.Split(subtopic, ".")
-	filteredElems := []string{}
-	for _, elem := range elems {
-		if elem == "" {
-			continue
-		}
-
-		if len(elem) > 1 && (strings.Contains(elem, "*") || strings.Contains(elem, ">")) {
-			return "", mgate.NewHTTPProxyError(http.StatusBadRequest, errMalformedSubtopic)
-		}
-
-		filteredElems = append(filteredElems, elem)
-	}
-
-	subtopic = strings.Join(filteredElems, ".")
-	return subtopic, nil
-}