mqtt添加日志和recovery

mqtt_binding/middleware/middleware.go

@@ -0,0 +1,39 @@
+package middleware
+
+import (
+	"git.sxidc.com/go-tools/api_binding/mqtt_binding/mqtt_client/router"
+	"git.sxidc.com/service-supports/fslog"
+	"runtime/debug"
+	"slices"
+	"time"
+)
+
+func Logger(skipTopics []string) router.Handler {
+	return func(item *router.Item, data []byte) {
+		start := time.Now()
+		item.Next()
+
+		if slices.Contains(skipTopics, item.Topic) {
+			return
+		}
+
+		end := time.Now()
+		fslog.Info("| %d | %s |",
+			end.Sub(start),
+			item.Topic,
+		)
+	}
+}
+
+func Recovery() router.Handler {
+	return func(item *router.Item, data []byte) {
+		defer func() {
+			err := recover()
+			if err != nil {
+				fslog.Error("%s", debug.Stack())
+			}
+		}()
+
+		item.Next()
+	}
+}

mqtt_binding/mqtt_binding.go

@@ -2,6 +2,7 @@ package mqtt_binding
 
 import (
 	"errors"
+	"git.sxidc.com/go-tools/api_binding/mqtt_binding/middleware"
 	"git.sxidc.com/go-tools/api_binding/mqtt_binding/mqtt_client"
 	"git.sxidc.com/go-tools/api_binding/mqtt_binding/mqtt_client/router"
 	"git.sxidc.com/go-tools/api_binding/mqtt_binding/request"
@@ -25,6 +26,11 @@ func NewBinding(apiVersion string, responseIdentifier response.Identifier, handl
 		group += "/" + apiVersion
 	}
 
+	handlers = append([]router.Handler{
+		middleware.Logger([]string{group + "/version"}),
+		middleware.Recovery(),
+	}, handlers...)
+
 	r := mqttClientInstance.GetRouter(group, handlers)
 
 	return &Binding{

mqtt_binding_test.go

@@ -20,8 +20,11 @@ func (callerIdentifier *CallerIdentifier) Identifier() string {
 	return callerIdentifier.CallerIdentifier
 }
 
-var bindingCallerIdentifier = &CallerIdentifier{CallerIdentifier: ""}
-var itemCallerIdentifier = &CallerIdentifier{CallerIdentifier: ""}
+var bindingCallerIdentifier = &CallerIdentifier{CallerIdentifier: "binding"}
+var itemCallerIdentifier = &CallerIdentifier{CallerIdentifier: "item"}
+
+//var bindingCallerIdentifier = nil
+//var itemCallerIdentifier = nil
 
 func TestMqttBinding(t *testing.T) {
 	err := mqtt_binding.Init("test_prefix", &mqtt_client.MqttClientOptions{
@@ -51,6 +54,7 @@ func TestMqttBinding(t *testing.T) {
 		BusinessFunc: func(c *mqtt_client.MqttClient, inputModel struct {
 			Time string `json:"time"`
 		}) (map[string]interface{}, error) {
+			panic("11111111")
 			fmt.Printf("Received: %v\n", inputModel)
 
 			return map[string]interface{}{
@@ -75,9 +79,9 @@ func TestMqttBinding(t *testing.T) {
 		SetWill("test-client/will", "dead", 2, true).
 		SetOnConnectHandler(func(client mqtt.Client) {
 			replyTopic := "test_prefix/test/test-topic/reply"
-			if itemCallerIdentifier.Identifier() != "" {
+			if itemCallerIdentifier != nil {
 				replyTopic = "test_prefix/test/test-topic/" + itemCallerIdentifier.Identifier() + "/reply"
-			} else if bindingCallerIdentifier.Identifier() != "" {
+			} else if bindingCallerIdentifier != nil {
 				replyTopic = "test_prefix/test/test-topic/" + bindingCallerIdentifier.Identifier() + "/reply"
 			}
 
@@ -90,7 +94,8 @@ func TestMqttBinding(t *testing.T) {
 			}
 
 			sendMap := map[string]any{
-				"time": time.Now().Format(time.DateTime),
+				"callerIdentifier": "test",
+				"time":             time.Now().Format(time.DateTime),
 			}
 
 			if itemCallerIdentifier.Identifier() != "" {