Develop

.github/actions/publish-website/action.yml

@@ -39,7 +39,7 @@ runs:
       run: docker load --input /tmp/mokapi.tar
       shell: bash
     - name: Run mokapi image
-      run: docker run --name mokapi --rm -d -p 80:80 -p 8080:8080 -p 9092:9092 -p 8389:8389 -p 8025:8025 --mount type=bind,source=$(pwd)/webui/scripts/dashboard-demo/demo-configs,target=/data --env MOKAPI_Providers_File_Directory=/data ${{ inputs.image-name }}
+      run: docker run --name mokapi --rm -d -p 80:80 -p 1883:1883 -p 8080:8080 -p 9092:9092 -p 8389:8389 -p 8025:8025 --mount type=bind,source=$(pwd)/webui/scripts/dashboard-demo/demo-configs,target=/data --env MOKAPI_Providers_File_Directory=/data ${{ inputs.image-name }}
       shell: bash
     - name: Build Demo Dashboard
       working-directory: ./webui/scripts/dashboard-demo

acceptance/cmd_test.go

@@ -65,10 +65,12 @@ func Start(cfg *static.Config) (*Cmd, error) {
 	})
 
 	apiHandler := api.New(app, cfg.Api)
-	if u, err := api.BuildUrl(cfg.Api); err == nil {
-		err = http.AddInternalService("api", u, apiHandler)
-		if err != nil {
-			return nil, err
+	if urls, err := api.BuildUrl(cfg.Api); err == nil {
+		for _, u := range urls {
+			err = http.AddInternalService("api", u, apiHandler)
+			if err != nil {
+				return nil, err
+			}
 		}
 	} else {
 		return nil, err

acceptance/ldap_test.go

@@ -81,7 +81,7 @@ func (suite *LdapSuite) TestMetric() {
 	}
 	_, err := suite.Client.Search(search)
 	require.NoError(suite.T(), err)
-	require.Equal(suite.T(), float64(1), suite.cmd.App.Monitor.Ldap.RequestCounter.Sum())
+	require.Equal(suite.T(), float64(1), suite.cmd.App.Monitor.Ldap.RequestCounter.Sum(metrics.NewQuery()))
 	q := metrics.NewQuery(metrics.ByNamespace("ldap"), metrics.ByName("request_timestamp"))
 	require.Greater(suite.T(), suite.cmd.App.Monitor.Ldap.LastRequest.Value(q), float64(1))
 }

acceptance/petstore/asyncapi.yml

@@ -6,6 +6,9 @@ servers:
   broker:
     url: 127.0.0.1:19092
     protocol: kafka
+  mqtt:
+    url: 127.0.0.1:11883
+    protocol: mqtt
 channels:
   petstore.order-event:
     subscribe:

acceptance/petstore_test.go

@@ -28,6 +28,7 @@ func (suite *PetStoreSuite) SetupSuite() {
 	cfg.Providers.File.Directories = []static.FileConfig{{Path: "./petstore"}}
 	cfg.Api.Search.Enabled = true
 	cfg.Api.Search.InMemory = true
+	cfg.Api.Search.NumIndexWorker = 1
 	suite.initCmd(cfg)
 }
 
@@ -54,78 +55,21 @@ func (suite *PetStoreSuite) TestApi() {
 
 	suite.T().Run("get AsyncAPI service", func(t *testing.T) {
 		expected := map[string]interface{}{
-			"version":     "1.0.0",
-			"name":        "A sample AsyncApi Kafka streaming api",
-			"description": "",
+			"version": "1.0.0",
+			"name":    "A sample AsyncApi Kafka streaming api",
 			"servers": []interface{}{
 				map[string]interface{}{
-					"host":        "127.0.0.1:19092",
-					"name":        "broker",
-					"protocol":    "kafka",
-					"title":       "",
-					"summary":     "",
-					"description": "",
+					"host":     "127.0.0.1:19092",
+					"name":     "broker",
+					"protocol": "kafka",
 				},
 			},
 
 			"topics": []interface{}{map[string]interface{}{
-				"bindings":    map[string]interface{}{"partitions": float64(2), "segmentMs": float64(30000), "valueSchemaValidation": true},
-				"description": "",
-				"messages": map[string]interface{}{
-					"order": map[string]interface{}{
-						"name":        "order",
-						"contentType": "application/json",
-						"header": map[string]interface{}{
-							"schema": map[string]interface{}{
-								"properties": map[string]interface{}{
-									"number": map[string]interface{}{
-										"type": "number",
-									}, "test": map[string]interface{}{
-										"type": "string",
-									},
-								}, "type": "object",
-							},
-						},
-						"key": map[string]interface{}{
-							"schema": map[string]interface{}{
-								"type": "string",
-							},
-						},
-						"payload": map[string]interface{}{
-							"schema": map[string]interface{}{
-								"properties": map[string]interface{}{
-									"accepted": map[string]interface{}{
-										"properties": map[string]interface{}{
-											"timestamp": map[string]interface{}{"format": "date-time", "type": "string"},
-										},
-										"type": "object",
-									},
-									"completed": map[string]interface{}{
-										"properties": map[string]interface{}{
-											"timestamp": map[string]interface{}{"format": "date-time", "type": "string"},
-										},
-										"type": "object",
-									},
-									"id": map[string]interface{}{"type": "integer"},
-									"placed": map[string]interface{}{
-										"properties": map[string]interface{}{
-											"petid":     map[string]interface{}{"type": "integer"},
-											"quantity":  map[string]interface{}{"format": "int32", "type": "integer"},
-											"ship-date": map[string]interface{}{"format": "date-time", "type": "string"},
-										},
-										"type": "object",
-									},
-								},
-								"required": []interface{}{"id"},
-								"type":     "object",
-							},
-						},
-					},
-				},
 				"name": "petstore.order-event",
-				"partitions": []interface{}{
-					map[string]interface{}{"id": float64(0), "offset": float64(1), "segments": float64(1), "startOffset": float64(0)},
-					map[string]interface{}{"id": float64(1), "offset": float64(0), "segments": float64(0), "startOffset": float64(0)},
+				"metrics": map[string]interface{}{
+					// skip timestamp check "kafka_message_timestamp"
+					"kafka_messages_total": float64(1),
 				},
 			}},
 		}
@@ -220,7 +164,7 @@ func (suite *PetStoreSuite) TestKafka_TopicConfig() {
 	require.Equal(suite.T(), "petstore.order-event", r.Topics[0].Name)
 	require.Len(suite.T(), r.Topics[0].Partitions, 2)
 
-	require.Len(suite.T(), suite.cmd.App.ListHttp(), 1)
+	require.Equal(suite.T(), 1, suite.cmd.App.Http.Len())
 }
 
 func (suite *PetStoreSuite) TestKafka_Produce_InvalidFormat() {
@@ -303,8 +247,16 @@ func (suite *PetStoreSuite) TestEvents() {
 			assert.True(t, ok, "event should be a map[string]any")
 			assert.NotNil(t, evt)
 			assert.Equal(t, "Event", evt["type"])
-			assert.Equal(t, "GET http://127.0.0.1:18080/user/bob", evt["title"])
+			assert.Equal(t, "http://127.0.0.1:18080/user/bob", evt["title"])
 			assert.Equal(t, "Swagger Petstore", evt["domain"])
+			params := evt["params"].(map[string]any)
+			assert.Len(t, params, 6)
+			assert.Equal(t, "event", params["type"])
+			assert.Equal(t, "http", params["traits.namespace"])
+			assert.Equal(t, "Swagger Petstore", params["traits.name"])
+			assert.Equal(t, "/user/{username}", params["traits.path"])
+			assert.Equal(t, "GET", params["traits.method"])
+			assert.Equal(t, "Swagger Petstore", params["traits.name"])
 		}),
 	)
 }
@@ -389,7 +341,7 @@ func (suite *PetStoreSuite) TestSearch_Paging() {
 			assert.Len(t, items, 10)
 			evt := items[0].(map[string]interface{})
 			assert.Equal(t, "HTTP", evt["type"])
-			assert.Equal(t, "GET /pet/{petId}", evt["title"])
+			assert.Equal(t, "/pet/{petId}", evt["title"])
 			assert.Equal(t, "Swagger Petstore", evt["domain"])
 		}),
 	)

api/handler.go

@@ -7,7 +7,6 @@ import (
 	"io/fs"
 	"mokapi/config/static"
 	"mokapi/runtime"
-	"mokapi/runtime/metrics"
 	"mokapi/webui"
 	"net/http"
 	"net/url"
@@ -16,6 +15,7 @@ import (
 	"strconv"
 	"strings"
 
+	"github.com/gorilla/mux"
 	log "github.com/sirupsen/logrus"
 )
 
@@ -36,6 +36,7 @@ type handler struct {
 	healthHandler http.Handler
 	mcpPath       string
 	mcpHandler    http.Handler
+	router        *mux.Router
 }
 
 type info struct {
@@ -56,15 +57,17 @@ var (
 	ServiceKafka serviceType = "kafka"
 	ServiceMail  serviceType = "mail"
 	ServiceLdap  serviceType = "ldap"
+	ServiceMqtt  serviceType = "mqtt"
 )
 
 type service struct {
-	Name        string           `json:"name"`
-	Description string           `json:"description,omitempty"`
-	Contact     *contact         `json:"contact,omitempty"`
-	Version     string           `json:"version,omitempty"`
-	Type        serviceType      `json:"type"`
-	Metrics     []metrics.Metric `json:"metrics,omitempty"`
+	Name        string      `json:"name"`
+	Description string      `json:"description,omitempty"`
+	Contact     *contact    `json:"contact,omitempty"`
+	Version     string      `json:"version,omitempty"`
+	Type        serviceType `json:"type"`
+	Status      string      `json:"status,omitempty"`
+	Metrics     any         `json:"metrics,omitempty"`
 }
 
 type contact struct {
@@ -83,6 +86,7 @@ func New(app *runtime.App, config static.Api) Handler {
 		path:   config.Path,
 		base:   config.Base,
 		app:    app,
+		router: mux.NewRouter(),
 	}
 
 	if config.Dashboard {
@@ -101,12 +105,26 @@ func New(app *runtime.App, config static.Api) Handler {
 		h.fileServer = http.FileServer(http.FS(dist))
 	}
 
+	h.setupHttp()
+	h.setupKafka()
+	h.setupMqtt()
+
 	return h
 }
 
-func BuildUrl(cfg static.Api) (*url.URL, error) {
-	s := fmt.Sprintf("http://:%v%v", cfg.Port, cfg.Path)
-	return url.Parse(s)
+func BuildUrl(cfg static.Api) ([]*url.URL, error) {
+	var urls []*url.URL
+	u, err := url.Parse(fmt.Sprintf("http://:%v%v", cfg.Port, cfg.Path))
+	if err != nil {
+		return urls, err
+	}
+	urls = append(urls, u)
+	u, err = url.Parse(fmt.Sprintf("http://localhost:%v%v", cfg.Port, cfg.Path))
+	if err != nil {
+		return urls, err
+	}
+	urls = append(urls, u)
+	return urls, nil
 }
 
 func (h *handler) RegisterHealthHandler(path string, handler http.Handler) {
@@ -134,10 +152,6 @@ func (h *handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		h.getInfo(w, r)
 	case p == "/api/services":
 		h.getServices(w, r)
-	case strings.HasPrefix(p, "/api/services/http/"):
-		h.getHttpService(w, r, h.app.Monitor)
-	case strings.HasPrefix(p, "/api/services/kafka"):
-		h.handleKafka(w, r)
 	case strings.HasPrefix(p, "/api/services/mail/"):
 		h.handleMailService(w, r)
 	case strings.HasPrefix(p, "/api/services/ldap/"):
@@ -164,6 +178,8 @@ func (h *handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		h.healthHandler.ServeHTTP(w, r)
 	case strings.HasPrefix(p, h.mcpPath) && h.mcpHandler != nil:
 		h.mcpHandler.ServeHTTP(w, r)
+	case strings.HasPrefix(p, "/api/"):
+		h.router.ServeHTTP(w, r)
 	case h.fileServer != nil:
 		if r.Method != "GET" {
 			http.Error(w, fmt.Sprintf("method %v is not allowed", r.Method), http.StatusMethodNotAllowed)
@@ -198,14 +214,28 @@ func (h *handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
-func (h *handler) getServices(w http.ResponseWriter, _ *http.Request) {
-	services := make([]interface{}, 0)
-	services = append(services, getHttpServices(h.app.ListHttp(), h.app.Monitor)...)
-	services = append(services, getKafkaServices(h.app.Kafka, h.app.Monitor)...)
-	services = append(services, getMailServices(h.app.Mail, h.app.Monitor)...)
-	services = append(services, getLdapServices(h.app.Ldap, h.app.Monitor)...)
-	slices.SortFunc(services, func(a interface{}, b interface{}) int {
-		return compareService(a, b)
+func (h *handler) getServices(w http.ResponseWriter, r *http.Request) {
+	services := make([]service, 0)
+
+	typ := r.URL.Query().Get("type")
+
+	if typ == "" || typ == "http" {
+		services = append(services, getHttpServices(h.app.Http, h.app.Monitor)...)
+	}
+	if typ == "" || typ == "kafka" {
+		services = append(services, getKafkaServices(h.app.Kafka, h.app.Monitor)...)
+	}
+	if typ == "" || typ == "mail" {
+		services = append(services, getMailServices(h.app.Mail, h.app.Monitor)...)
+	}
+	if typ == "" || typ == "ldap" {
+		services = append(services, getLdapServices(h.app.Ldap, h.app.Monitor)...)
+	}
+	if typ == "" || typ == "mqtt" {
+		services = append(services, getMqttServices(h.app.Mqtt, h.app.Monitor)...)
+	}
+	slices.SortFunc(services, func(a service, b service) int {
+		return strings.Compare(a.Name, b.Name)
 	})
 	w.Header().Set("Content-Type", "application/json")
 	writeJsonBody(w, services)
@@ -230,22 +260,30 @@ func (h *handler) getInfo(w http.ResponseWriter, _ *http.Request) {
 	w.Header().Set("Content-Type", "application/json")
 
 	i := info{Version: h.app.Version, BuildTime: h.app.BuildTime, Search: searchInfo{Enabled: h.config.Search.Enabled}}
-	if len(h.app.ListHttp()) > 0 {
+	if h.app.Http.Len() > 0 {
 		i.ActiveServices = append(i.ActiveServices, "http")
 	}
-	if len(h.app.Kafka.List()) > 0 {
+	if h.app.Kafka.Len() > 0 {
 		i.ActiveServices = append(i.ActiveServices, "kafka")
 	}
-	if len(h.app.Mail.List()) > 0 {
+	if h.app.Mail.Len() > 0 {
 		i.ActiveServices = append(i.ActiveServices, "mail")
 	}
-	if len(h.app.Ldap.List()) > 0 {
+	if h.app.Ldap.Len() > 0 {
 		i.ActiveServices = append(i.ActiveServices, "ldap")
 	}
+	if h.app.Mqtt.Len() > 0 {
+		i.ActiveServices = append(i.ActiveServices, "mqtt")
+	}
 
 	writeJsonBody(w, i)
 }
 
+func write(w http.ResponseWriter, data any) {
+	w.Header().Set("Content-Type", "application/json")
+	writeJsonBody(w, data)
+}
+
 func writeJsonBody(w http.ResponseWriter, v interface{}) {
 	var buf bytes.Buffer
 	enc := json.NewEncoder(&buf)
@@ -278,24 +316,6 @@ func isImage(path string) bool {
 	}
 }
 
-func compareService(a, b interface{}) int {
-	return strings.Compare(getServiceName(a), getServiceName(b))
-}
-
-func getServiceName(a interface{}) string {
-	switch v := a.(type) {
-	case *httpSummary:
-		return v.Name
-	case *kafkaSummary:
-		return v.Name
-	case *ldapSummary:
-		return v.Name
-	case *mailSummary:
-		return v.Name
-	}
-	return ""
-}
-
 func getPageInfo(r *http.Request) (index int, limit int, err error) {
 	limit = 10