【优化】规则链rpc请求逻辑

pkg/rule_engine/message/message.go

@@ -8,14 +8,15 @@ import (
 
 // 消息类型
 const (
-	ConnectMes    = "Connect"
+	ConnectMes          = "Connect"
-	DisConnectMes = "Disconnect"
+	DisConnectMes       = "Disconnect"
-	RpcRequestMes = "RpcRequest"
+	RpcRequestMes       = "RpcRequestFromDevice"
-	UpEventMes    = "Event"
+	RpcRequestServerMes = "RpcRequestFromServer"
-	AlarmMes      = "Alarm"
+	UpEventMes          = "Event"
-	RowMes        = "Row"
+	AlarmMes            = "Alarm"
-	TelemetryMes  = "Telemetry"
+	RowMes              = "Row"
-	AttributesMes = "Attributes"
+	TelemetryMes        = "Telemetry"
+	AttributesMes       = "Attributes"
 )
 
 // 数据类型Originator

pkg/rule_engine/nodes/action_rpc_request_node.go

@@ -1,14 +1,14 @@
 package nodes
 
 import (
-	"errors"
 	"pandax/iothub/client/mqttclient"
 	"pandax/pkg/rule_engine/message"
 )
 
 type rpcRequestNode struct {
 	bareNode
-	RequestId int `json:"requestId"`
+	Timeout int                   `json:"timeout"`
+	Payload mqttclient.RpcPayload `json:"payload"`
 }
 
 type rpcRequestNodeFactory struct{}
@@ -26,20 +26,10 @@ func (f rpcRequestNodeFactory) Create(id string, meta Metadata) (Node, error) {
 func (n *rpcRequestNode) Handle(msg *message.Message) error {
 	successLableNode := n.GetLinkedNode("Success")
 	failureLableNode := n.GetLinkedNode("Failure")
-	RequestId := n.RequestId
+
-	if RequestId == 0 {
+	var rpc = &mqttclient.RpcRequest{Client: mqttclient.MqttClient, Mode: "single", Timeout: n.Timeout}
-		RequestId = int(msg.Metadata.GetValue("requestId").(float64))
-	}
-	if msg.Msg.GetValue("method") == nil || msg.Msg.GetValue("params") == nil {
-		return errors.New("请求响应格式不正确")
-	}
-	var datas = mqttclient.RpcPayload{
-		Method: msg.Msg.GetValue("method").(string),
-		Params: msg.Msg.GetValue("params"),
-	}
-	var rpc = &mqttclient.RpcRequest{Client: mqttclient.MqttClient, Mode: "single"}
 	rpc.GetRequestId()
-	err := rpc.RespondTpc(datas)
+	respPayload, err := rpc.RequestCmd(n.Payload)
 	if err != nil {
 		if failureLableNode != nil {
 			return failureLableNode.Handle(msg)
@@ -47,6 +37,9 @@ func (n *rpcRequestNode) Handle(msg *message.Message) error {
 			return err
 		}
 	}
+	msgM := msg.Msg
+	msgM["payload"] = respPayload
+	msg.Msg = msgM
 	if successLableNode != nil {
 		return successLableNode.Handle(msg)
 	}

pkg/rule_engine/nodes/action_rpc_respond_node.go

@@ -0,0 +1,49 @@
+package nodes
+
+import (
+	"pandax/iothub/client/mqttclient"
+	"pandax/pkg/rule_engine/message"
+)
+
+type rpcRespondNode struct {
+	bareNode
+	RequestId int `json:"requestId"`
+}
+
+type rpcRespondFactory struct{}
+
+func (f rpcRespondFactory) Name() string     { return "RpcRespondNode" }
+func (f rpcRespondFactory) Category() string { return NODE_CATEGORY_ACTION }
+func (f rpcRespondFactory) Labels() []string { return []string{"Success", "Failure"} }
+func (f rpcRespondFactory) Create(id string, meta Metadata) (Node, error) {
+	node := &rpcRespondNode{
+		bareNode: newBareNode(f.Name(), id, meta, f.Labels()),
+	}
+	return decodePath(meta, node)
+}
+
+func (n *rpcRespondNode) Handle(msg *message.Message) error {
+	successLableNode := n.GetLinkedNode("Success")
+	failureLableNode := n.GetLinkedNode("Failure")
+	RequestId := n.RequestId
+	if RequestId == 0 {
+		RequestId = int(msg.Metadata.GetValue("requestId").(float64))
+	}
+	var datas = mqttclient.RpcPayload{
+		Method: msg.Msg.GetValue("method").(string),
+		Params: msg.Msg.GetValue("params"),
+	}
+	rpc := &mqttclient.RpcRequest{Client: mqttclient.MqttClient, RequestId: RequestId}
+	err := rpc.RespondTpc(datas)
+	if err != nil {
+		if failureLableNode != nil {
+			return failureLableNode.Handle(msg)
+		} else {
+			return err
+		}
+	}
+	if successLableNode != nil {
+		return successLableNode.Handle(msg)
+	}
+	return nil
+}

pkg/rule_engine/nodes/filter_message_type_switch_node.go

@@ -18,6 +18,7 @@ func (f messageTypeSwitchNodeFactory) Labels() []string {
 		message.AttributesMes,
 		message.TelemetryMes,
 		message.RpcRequestMes,
+		message.RpcRequestServerMes,
 		message.AlarmMes,
 		message.UpEventMes,
 		message.ConnectMes,

pkg/rule_engine/nodes/filter_switch_node.go

@@ -21,6 +21,7 @@ func (f switchFilterNodeFactory) Labels() []string {
 		message.AttributesMes,
 		message.TelemetryMes,
 		message.RpcRequestMes,
+		message.RpcRequestServerMes,
 		message.AlarmMes,
 		message.UpEventMes,
 		message.ConnectMes,

pkg/rule_engine/nodes/init.go

@@ -28,5 +28,6 @@ func init() {
 	RegisterFactory(externalSendEmailNodeFactory{})
 	RegisterFactory(externalSendSmsNodeFactory{})
 	RegisterFactory(externalRuleChainNodeFactory{})
+	RegisterFactory(rpcRespondFactory{})
 	RegisterFactory(rpcRequestNodeFactory{})
 }

shutdown.bat

@@ -1 +1 @@
-taskkill /pid 32324  -t  -f
+taskkill /pid 24048  -t  -f