refactor: update to use topic parser

Signed-off-by: Felix Gateru <[email protected]>

Author: felixgateru

cmd/http/main.go

@@ -21,6 +21,7 @@ import (
 	"github.com/absmach/supermq"
 	grpcChannelsV1 "github.com/absmach/supermq/api/grpc/channels/v1"
 	grpcClientsV1 "github.com/absmach/supermq/api/grpc/clients/v1"
+	grpcDomainsV1 "github.com/absmach/supermq/api/grpc/domains/v1"
 	adapter "github.com/absmach/supermq/http"
 	httpapi "github.com/absmach/supermq/http/api"
 	"github.com/absmach/supermq/http/middleware"
@@ -201,12 +202,15 @@ func main() {
 		return
 	}
 
-	svc, err := newService(pub, authn, cacheConfig, clientsClient, channelsClient, domainsClient, logger, tracer)
+	resolver := messaging.NewTopicResolver(channelsClient, domainsClient)
+	handler, err := newHandler(nps, authn, cacheConfig, clientsClient, channelsClient, domainsClient, logger, tracer)
 	if err != nil {
 		logger.Error(fmt.Sprintf("failed to create service: %s", err))
 		exitCode = 1
 		return
 	}
+	svc := newService(clientsClient, channelsClient, nps, logger, tracer)
+
 	targetServerCfg := server.Config{Port: targetHTTPPort}
 
 	hs := httpserver.NewServer(ctx, cancel, svcName, targetServerCfg, httpapi.MakeHandler(ctx, svc, resolver, logger, cfg.InstanceID), logger)
@@ -233,27 +237,28 @@ func main() {
 	}
 }
 
-func newService(pub messaging.Publisher, authn smqauthn.Authentication, cacheCfg messaging.CacheConfig, clients grpcClientsV1.ClientsServiceClient, channels grpcChannelsV1.ChannelsServiceClient, domains grpcDomainsV1.DomainsServiceClient, logger *slog.Logger, tracer trace.Tracer) (session.Handler, error) {
+func newHandler(pub messaging.Publisher, authn smqauthn.Authentication, cacheCfg messaging.CacheConfig, clients grpcClientsV1.ClientsServiceClient, channels grpcChannelsV1.ChannelsServiceClient, domains grpcDomainsV1.DomainsServiceClient, logger *slog.Logger, tracer trace.Tracer) (session.Handler, error) {
 	parser, err := messaging.NewTopicParser(cacheCfg, channels, domains)
 	if err != nil {
 		return nil, err
 	}
-	svc := adapter.NewHandler(pub, authn, clients, channels, parser, logger)
-	svc = handler.NewTracing(tracer, svc)
-	svc = handler.LoggingMiddleware(svc, logger)
+	h := adapter.NewHandler(pub, authn, clients, channels, parser, logger)
+	h = handler.NewTracing(tracer, h)
+	h = handler.LoggingMiddleware(h, logger)
 	counter, latency := prometheus.MakeMetrics(svcName, "handler")
-	svc = handler.MetricsMiddleware(svc, counter, latency)
+	h = handler.MetricsMiddleware(h, counter, latency)
 
-	return svc
+	return h, nil
 }
 
 func newService(clientsClient grpcClientsV1.ClientsServiceClient, channels grpcChannelsV1.ChannelsServiceClient, nps messaging.PubSub, logger *slog.Logger, tracer trace.Tracer) adapter.Service {
 	svc := adapter.NewService(clientsClient, channels, nps)
 	svc = middleware.Tracing(tracer, svc)
 	svc = middleware.Logging(svc, logger)
 	counter, latency := prometheus.MakeMetrics(svcName, "api")
-	svc = handler.MetricsMiddleware(svc, counter, latency)
-	return svc, nil
+	svc = middleware.Metrics(svc, counter, latency)
+
+	return svc
 }
 
 func proxyHTTP(ctx context.Context, cfg server.Config, logger *slog.Logger, sessionHandler session.Handler) error {

docker/.env

@@ -409,17 +409,6 @@ SMQ_COAP_ADAPTER_CACHE_MAX_COST=1048576
 SMQ_COAP_ADAPTER_CACHE_BUFFER_ITEMS=64
 SMQ_COAP_ADAPTER_INSTANCE_ID=
 
-### WS
-SMQ_WS_ADAPTER_LOG_LEVEL=debug
-SMQ_WS_ADAPTER_HTTP_HOST=ws-adapter
-SMQ_WS_ADAPTER_HTTP_PORT=8186
-SMQ_WS_ADAPTER_HTTP_SERVER_CERT=
-SMQ_WS_ADAPTER_HTTP_SERVER_KEY=
-SMQ_WS_ADAPTER_CACHE_NUM_COUNTERS=200000
-SMQ_WS_ADAPTER_CACHE_MAX_COST=1048576
-SMQ_WS_ADAPTER_CACHE_BUFFER_ITEMS=64
-SMQ_WS_ADAPTER_INSTANCE_ID=
-
 ## Addons Services
 ### Vault
 SMQ_VAULT_HOST=vault

docker/docker-compose.yaml

@@ -1263,118 +1263,6 @@ services:
         bind:
           create_host_path: true
 
-  ws-adapter:
-    image: supermq/ws:${SMQ_RELEASE_TAG}
-    container_name: supermq-ws
-    depends_on:
-      - clients
-      - nats
-    restart: on-failure
-    environment:
-      SMQ_WS_ADAPTER_LOG_LEVEL: ${SMQ_WS_ADAPTER_LOG_LEVEL}
-      SMQ_WS_ADAPTER_HTTP_HOST: ${SMQ_WS_ADAPTER_HTTP_HOST}
-      SMQ_WS_ADAPTER_HTTP_PORT: ${SMQ_WS_ADAPTER_HTTP_PORT}
-      SMQ_WS_ADAPTER_HTTP_SERVER_CERT: ${SMQ_WS_ADAPTER_HTTP_SERVER_CERT}
-      SMQ_WS_ADAPTER_HTTP_SERVER_KEY: ${SMQ_WS_ADAPTER_HTTP_SERVER_KEY}
-      SMQ_WS_ADAPTER_CACHE_NUM_COUNTERS: ${SMQ_WS_ADAPTER_CACHE_NUM_COUNTERS}
-      SMQ_WS_ADAPTER_CACHE_MAX_COST: ${SMQ_WS_ADAPTER_CACHE_MAX_COST}
-      SMQ_WS_ADAPTER_CACHE_BUFFER_ITEMS: ${SMQ_WS_ADAPTER_CACHE_BUFFER_ITEMS}
-      SMQ_CLIENTS_GRPC_URL: ${SMQ_CLIENTS_GRPC_URL}
-      SMQ_CLIENTS_GRPC_TIMEOUT: ${SMQ_CLIENTS_GRPC_TIMEOUT}
-      SMQ_CLIENTS_GRPC_CLIENT_CERT: ${SMQ_CLIENTS_GRPC_CLIENT_CERT:+/clients-grpc-client.crt}
-      SMQ_CLIENTS_GRPC_CLIENT_KEY: ${SMQ_CLIENTS_GRPC_CLIENT_KEY:+/clients-grpc-client.key}
-      SMQ_CLIENTS_GRPC_SERVER_CA_CERTS: ${SMQ_CLIENTS_GRPC_SERVER_CA_CERTS:+/clients-grpc-server-ca.crt}
-      SMQ_CHANNELS_GRPC_URL: ${SMQ_CHANNELS_GRPC_URL}
-      SMQ_CHANNELS_GRPC_TIMEOUT: ${SMQ_CHANNELS_GRPC_TIMEOUT}
-      SMQ_CHANNELS_GRPC_CLIENT_CERT: ${SMQ_CHANNELS_GRPC_CLIENT_CERT:+/channels-grpc-client.crt}
-      SMQ_CHANNELS_GRPC_CLIENT_KEY: ${SMQ_CHANNELS_GRPC_CLIENT_KEY:+/channels-grpc-client.key}
-      SMQ_CHANNELS_GRPC_SERVER_CA_CERTS: ${SMQ_CHANNELS_GRPC_SERVER_CA_CERTS:+/channels-grpc-server-ca.crt}
-      SMQ_DOMAINS_GRPC_URL: ${SMQ_DOMAINS_GRPC_URL}
-      SMQ_DOMAINS_GRPC_TIMEOUT: ${SMQ_DOMAINS_GRPC_TIMEOUT}
-      SMQ_DOMAINS_GRPC_CLIENT_CERT: ${SMQ_DOMAINS_GRPC_CLIENT_CERT:+/domains-grpc-client.crt}
-      SMQ_DOMAINS_GRPC_CLIENT_KEY: ${SMQ_DOMAINS_GRPC_CLIENT_KEY:+/domains-grpc-client.key}
-      SMQ_DOMAINS_GRPC_SERVER_CA_CERTS: ${SMQ_DOMAINS_GRPC_SERVER_CA_CERTS:+/domains-grpc-server-ca.crt}
-      SMQ_AUTH_GRPC_URL: ${SMQ_AUTH_GRPC_URL}
-      SMQ_AUTH_GRPC_TIMEOUT: ${SMQ_AUTH_GRPC_TIMEOUT}
-      SMQ_AUTH_GRPC_CLIENT_CERT: ${SMQ_AUTH_GRPC_CLIENT_CERT:+/auth-grpc-client.crt}
-      SMQ_AUTH_GRPC_CLIENT_KEY: ${SMQ_AUTH_GRPC_CLIENT_KEY:+/auth-grpc-client.key}
-      SMQ_AUTH_GRPC_SERVER_CA_CERTS: ${SMQ_AUTH_GRPC_SERVER_CA_CERTS:+/auth-grpc-server-ca.crt}
-      SMQ_MESSAGE_BROKER_URL: ${SMQ_MESSAGE_BROKER_URL}
-      SMQ_JAEGER_URL: ${SMQ_JAEGER_URL}
-      SMQ_JAEGER_TRACE_RATIO: ${SMQ_JAEGER_TRACE_RATIO}
-      SMQ_SEND_TELEMETRY: ${SMQ_SEND_TELEMETRY}
-      SMQ_WS_ADAPTER_INSTANCE_ID: ${SMQ_WS_ADAPTER_INSTANCE_ID}
-      SMQ_ES_URL: ${SMQ_ES_URL}
-    ports:
-      - ${SMQ_WS_ADAPTER_HTTP_PORT}:${SMQ_WS_ADAPTER_HTTP_PORT}
-    networks:
-      - supermq-base-net
-    volumes:
-      # Clients gRPC mTLS client certificates
-      - type: bind
-        source: ${SMQ_CLIENTS_GRPC_CLIENT_CERT:-ssl/certs/dummy/client_cert}
-        target: /clients-grpc-client${SMQ_CLIENTS_GRPC_CLIENT_CERT:+.crt}
-        bind:
-          create_host_path: true
-      - type: bind
-        source: ${SMQ_CLIENTS_GRPC_CLIENT_KEY:-ssl/certs/dummy/client_key}
-        target: /clients-grpc-client${SMQ_CLIENTS_GRPC_CLIENT_KEY:+.key}
-        bind:
-          create_host_path: true
-      - type: bind
-        source: ${SMQ_CLIENTS_GRPC_SERVER_CA_CERTS:-ssl/certs/dummy/server_ca}
-        target: /clients-grpc-server-ca${SMQ_CLIENTS_GRPC_SERVER_CA_CERTS:+.crt}
-        bind:
-          create_host_path: true
-      # Channels gRPC mTLS client certificates
-      - type: bind
-        source: ${SMQ_CHANNELS_GRPC_CLIENT_CERT:-ssl/certs/dummy/client_cert}
-        target: /channels-grpc-client${SMQ_CHANNELS_GRPC_CLIENT_CERT:+.crt}
-        bind:
-          create_host_path: true
-      - type: bind
-        source: ${SMQ_CHANNELS_GRPC_CLIENT_KEY:-ssl/certs/dummy/client_key}
-        target: /channels-grpc-client${SMQ_CHANNELS_GRPC_CLIENT_KEY:+.key}
-        bind:
-          create_host_path: true
-      - type: bind
-        source: ${SMQ_CHANNELS_GRPC_SERVER_CA_CERTS:-ssl/certs/dummy/server_ca}
-        target: /channels-grpc-server-ca${SMQ_CHANNELS_GRPC_SERVER_CA_CERTS:+.crt}
-        bind:
-          create_host_path: true
-      # Auth gRPC mTLS client certificates
-      - type: bind
-        source: ${SMQ_AUTH_GRPC_CLIENT_CERT:-ssl/certs/dummy/client_cert}
-        target: /auth-grpc-client${SMQ_AUTH_GRPC_CLIENT_CERT:+.crt}
-        bind:
-          create_host_path: true
-      - type: bind
-        source: ${SMQ_AUTH_GRPC_CLIENT_KEY:-ssl/certs/dummy/client_key}
-        target: /auth-grpc-client${SMQ_AUTH_GRPC_CLIENT_KEY:+.key}
-        bind:
-          create_host_path: true
-      - type: bind
-        source: ${SMQ_AUTH_GRPC_SERVER_CA_CERTS:-ssl/certs/dummy/server_ca}
-        target: /auth-grpc-server-ca${SMQ_AUTH_GRPC_SERVER_CA_CERTS:+.crt}
-        bind:
-          create_host_path: true
-      # Domains gRPC mTLS client certificates
-      - type: bind
-        source: ${SMQ_DOMAINS_GRPC_CLIENT_CERT:-ssl/certs/dummy/client_cert}
-        target: /domains-grpc-server${SMQ_DOMAINS_GRPC_CLIENT_CERT:+.crt}
-        bind:
-          create_host_path: true
-      - type: bind
-        source: ${SMQ_DOMAINS_GRPC_CLIENT_KEY:-ssl/certs/dummy/client_key}
-        target: /domains-grpc-server${SMQ_DOMAINS_GRPC_CLIENT_KEY:+.key}
-        bind:
-          create_host_path: true
-      - type: bind
-        source: ${SMQ_DOMAINS_GRPC_SERVER_CA_CERTS:-ssl/certs/dummy/server_ca}
-        target: /domains-grpc-server-ca${SMQ_DOMAINS_GRPC_SERVER_CA_CERTS:+.crt}
-        bind:
-          create_host_path: true
-
   rabbitmq:
     image: rabbitmq:4.0.5-management-alpine
     container_name: supermq-rabbitmq

go.mod

@@ -6,7 +6,7 @@ require (
 	github.com/0x6flab/namegenerator v1.4.0
 	github.com/absmach/callhome v0.14.0
 	github.com/absmach/certs v0.0.0-20250602111612-89538302ad6a
-	github.com/absmach/mgate v0.4.6-0.20250605150648-edf967fbb46a
+	github.com/absmach/mgate v0.4.6-0.20250616124539-13181c84f1d5
 	github.com/absmach/senml v1.0.8
 	github.com/authzed/authzed-go v1.4.1
 	github.com/authzed/grpcutil v0.0.0-20250221190651-1985b19b35b8

go.sum

@@ -25,6 +25,8 @@ github.com/absmach/certs v0.0.0-20250602111612-89538302ad6a h1:swYXNJaGVQS35CeuX
 github.com/absmach/certs v0.0.0-20250602111612-89538302ad6a/go.mod h1:tEat7G8BzyWbFIFojqdzWSD6RZNFyEuUHBdnD0J+rZA=
 github.com/absmach/mgate v0.4.6-0.20250605150648-edf967fbb46a h1:1+772OQFHAS23JLAHrCZxO+DnGoiMllKcSwLQy74y+k=
 github.com/absmach/mgate v0.4.6-0.20250605150648-edf967fbb46a/go.mod h1:X2amjQg/2cnM+UKblMdpU2M4cZO74xtEHNIxtuUXCeA=
+github.com/absmach/mgate v0.4.6-0.20250616124539-13181c84f1d5 h1:cbJncI2bzHxj4y0znacoVlamVi9rN2ERVuHGMU3hCRc=
+github.com/absmach/mgate v0.4.6-0.20250616124539-13181c84f1d5/go.mod h1:X2amjQg/2cnM+UKblMdpU2M4cZO74xtEHNIxtuUXCeA=
 github.com/absmach/senml v1.0.8 h1:+opem/r4g6c6eA/JLyCIuksyEhj7eBdysY3pEmy1mqo=
 github.com/absmach/senml v1.0.8/go.mod h1:DRhzHLgvQoIUHroBgpFrSWso+bJZO9E96RlHAHy+VRI=
 github.com/antlr4-go/antlr/v4 v4.13.1 h1:SqQKkuVZ+zWkMMNkjy5FZe5mr5WURWnlpmOuzYWrPrQ=

http/api/endpoint_test.go

@@ -20,6 +20,7 @@ import (
 	grpcChannelsV1 "github.com/absmach/supermq/api/grpc/channels/v1"
 	grpcClientsV1 "github.com/absmach/supermq/api/grpc/clients/v1"
 	grpcCommonV1 "github.com/absmach/supermq/api/grpc/common/v1"
+	grpcDomainsV1 "github.com/absmach/supermq/api/grpc/domains/v1"
 	apiutil "github.com/absmach/supermq/api/http/util"
 	chmocks "github.com/absmach/supermq/channels/mocks"
 	climocks "github.com/absmach/supermq/clients/mocks"
@@ -96,7 +97,7 @@ func newService(clients grpcClientsV1.ClientsServiceClient, channels grpcChannel
 	return server.NewService(clients, channels, pubsub)
 }
 
-func newHandler(authn smqauthn.Authentication, clients grpcClientsV1.ClientsServiceClient, channels grpcChannelsV1.ChannelsServiceClient, resolver messaging.TopicResolver) (session.Handler, *pubsub.PubSub, error) {
+func newHandler(authn smqauthn.Authentication, clients grpcClientsV1.ClientsServiceClient, channels grpcChannelsV1.ChannelsServiceClient, domains grpcDomainsV1.DomainsServiceClient) (session.Handler, *pubsub.PubSub, error) {
 	pub := new(pubsub.PubSub)
 	parser, err := messaging.NewTopicParser(messaging.DefaultCacheConfig, channels, domains)
 	if err != nil {
@@ -167,9 +168,11 @@ func TestPublish(t *testing.T) {
 	msg := `[{"n":"current","t":-1,"v":1.6}]`
 	msgJSON := `{"field1":"val1","field2":"val2"}`
 	msgCBOR := `81A3616E6763757272656E746174206176FB3FF999999999999A`
-	svc, pub, err := newService(authn, clients, channels, domains)
+	handler, pubsub, err := newHandler(authn, clients, channels, domains)
 	assert.Nil(t, err, fmt.Sprintf("failed to create service with err: %v", err))
-	target := newTargetHTTPServer()
+	resolver := messaging.NewTopicResolver(channels, domains)
+	svc := newService(clients, channels, pubsub)
+	target := newTargetHTTPServer(resolver, svc)
 	defer target.Close()
 	ts, err := newProxyHTPPServer(handler, target)
 	assert.Nil(t, err, fmt.Sprintf("failed to create proxy server with err: %v", err))
@@ -339,7 +342,8 @@ func TestHandshake(t *testing.T) {
 	authn := new(authnMocks.Authentication)
 	domains := new(dmocks.DomainsServiceClient)
 	resolver := messaging.NewTopicResolver(channels, domains)
-	handler, pubsub := newHandler(authn, clients, channels, resolver)
+	handler, pubsub, err := newHandler(authn, clients, channels, domains)
+	assert.Nil(t, err, fmt.Sprintf("failed to create handler with err: %v", err))
 	svc := newService(clients, channels, pubsub)
 	target := newTargetHTTPServer(resolver, svc)
 	defer target.Close()

http/api/transport.go

@@ -100,7 +100,7 @@ func decodeWSReq(r *http.Request, resolver messaging.TopicResolver, logger *slog
 	domain := chi.URLParam(r, "domain")
 	channel := chi.URLParam(r, "channel")
 
-	domainID, channelID, err := resolver.Resolve(r.Context(), domain, channel)
+	domainID, channelID, _, err := resolver.Resolve(r.Context(), domain, channel)
 	if err != nil {
 		return connReq{}, err
 	}

http/handler.go

@@ -108,11 +108,7 @@ func (h *handler) AuthPublish(ctx context.Context, topic *string, payload *[]byt
 		return errClientNotInitialized
 	}
 
-	domain, channel, _, err := messaging.ParsePublishTopic(*topic)
-	if err != nil {
-		return mgate.NewHTTPProxyError(http.StatusBadRequest, errors.Wrap(errFailedPublish, err))
-	}
-	domainID, channelID, err := h.resolver.Resolve(ctx, domain, channel)
+	domainID, channelID, _, err := h.parser.ParsePublishTopic(ctx, *topic, true)
 	if err != nil {
 		return mgate.NewHTTPProxyError(http.StatusBadRequest, errors.Wrap(errFailedPublish, err))
 	}
@@ -140,15 +136,11 @@ func (h *handler) AuthSubscribe(ctx context.Context, topics *[]string) error {
 	}
 
 	for _, topic := range *topics {
-		domain, channel, _, err := messaging.ParseSubscribeTopic(topic)
+		domainID, channelID, _, err := h.parser.ParseSubscribeTopic(ctx, topic, true)
 		if err != nil {
 			return err
 		}
-		domainID, chanID, err := h.resolver.Resolve(ctx, domain, channel)
-		if err != nil {
-			return mgate.NewHTTPProxyError(http.StatusBadRequest, errors.Wrap(errFailedPublish, err))
-		}
-		if _, _, err := h.authAccess(ctx, string(s.Password), domainID, chanID, connections.Subscribe); err != nil {
+		if _, _, err := h.authAccess(ctx, string(s.Password), domainID, channelID, connections.Subscribe); err != nil {
 			return err
 		}
 	}

http/handler_test.go

@@ -139,7 +139,7 @@ func TestAuthConnect(t *testing.T) {
 }
 
 func TestAuthPublish(t *testing.T) {
-	handler := newHandler()
+	handler := newHandler(t)
 
 	clientKeySession := session.Session{
 		Password: []byte("Client " + clientKey),
@@ -278,7 +278,7 @@ func TestAuthPublish(t *testing.T) {
 }
 
 func TestAuthSubscribe(t *testing.T) {
-	handler := newHandler()
+	handler := newHandler(t)
 
 	clientKeySession := session.Session{
 		Password: []byte("Client " + clientKey),

pkg/sdk/message_test.go

@@ -48,12 +48,14 @@ func setupMessages(t *testing.T) (*httptest.Server, *pubsub.PubSub) {
 	domainsGRPCClient = new(dmocks.DomainsServiceClient)
 	pub := new(pubsub.PubSub)
 	authn := new(authnmocks.Authentication)
+	svc := new(httpmocks.Service)
 
 	parser, err := messaging.NewTopicParser(messaging.DefaultCacheConfig, channelsGRPCClient, domainsGRPCClient)
 	assert.Nil(t, err, fmt.Sprintf("unexpected error while setting up parser: %v", err))
 	handler := adapter.NewHandler(pub, authn, clientsGRPCClient, channelsGRPCClient, parser, smqlog.NewMock())
+	resolver := messaging.NewTopicResolver(channelsGRPCClient, domainsGRPCClient)
 
-	mux := api.MakeHandler(context.Background(), svc,resolver, smqlog.NewMock(), "")
+	mux := api.MakeHandler(context.Background(), svc, resolver, smqlog.NewMock(), "")
 	target := httptest.NewServer(mux)
 
 	ptUrl, _ := url.Parse(target.URL)