[优化]大改动，指令下发采用规则链rpc请求

pkg/rule_engine/instance.go

@@ -43,7 +43,7 @@ func newInstanceWithManifest(m *manifest.Manifest) (*ruleChainInstance, []error)
 // StartRuleChain TODO 是否需要添加context
 func (c *ruleChainInstance) StartRuleChain(context context.Context, message *message.Message) error {
 	if node, found := c.nodes[c.firstRuleNodeId]; found {
-		node.Handle(message)
+		return node.Handle(message)
 	}
 	return nil
 }

pkg/rule_engine/message/message.go

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

pkg/rule_engine/nodes/action_create_alarm_node.go

@@ -6,8 +6,8 @@ import (
 	"pandax/apps/device/entity"
 	"pandax/apps/device/services"
 	"pandax/pkg/global"
+	"pandax/pkg/global_model"
 	"pandax/pkg/rule_engine/message"
-	"pandax/pkg/tool"
 	"time"
 )
 
@@ -52,7 +52,7 @@ func (n *createAlarmNode) Handle(msg *message.Message) error {
 		}
 	} else {
 		alarm = &entity.DeviceAlarm{}
-		alarm.Id = tool.GenerateID()
+		alarm.Id = global_model.GenerateID()
 		alarm.DeviceId = msg.Metadata.GetValue("deviceId").(string)
 		alarm.ProductId = msg.Metadata.GetValue("productId").(string)
 		alarm.Name = msg.Metadata.GetValue("deviceName").(string)

pkg/rule_engine/nodes/action_rpc_request_from_device_node.go

@@ -0,0 +1,84 @@
+package nodes
+
+import (
+	"errors"
+	"github.com/sirupsen/logrus"
+	"pandax/iothub/client/mqttclient"
+	"pandax/iothub/client/tcpclient"
+	"pandax/pkg/global"
+	"pandax/pkg/global_model"
+	"pandax/pkg/rule_engine/message"
+)
+
+type rpcRequestFromDeviceNode struct {
+	bareNode
+	RequestId int `json:"requestId"`
+}
+
+type rpcRequestFromDeviceFactory struct{}
+
+func (f rpcRequestFromDeviceFactory) Name() string     { return "RpcRequestFromDeviceNode" }
+func (f rpcRequestFromDeviceFactory) Category() string { return NODE_CATEGORY_ACTION }
+func (f rpcRequestFromDeviceFactory) Labels() []string { return []string{"Success", "Failure"} }
+func (f rpcRequestFromDeviceFactory) Create(id string, meta Metadata) (Node, error) {
+	node := &rpcRequestFromDeviceNode{
+		bareNode: newBareNode(f.Name(), id, meta, f.Labels()),
+	}
+	return decodePath(meta, node)
+}
+
+func (n *rpcRequestFromDeviceNode) Handle(msg *message.Message) error {
+	logrus.Infof("%s handle message '%s'", n.Name(), msg.MsgType)
+	successLableNode := n.GetLinkedNode("Success")
+	failureLableNode := n.GetLinkedNode("Failure")
+
+	if msg.Msg.GetValue("method") == nil || msg.Msg.GetValue("params") == nil {
+		return errors.New("指令请求格式错误")
+	}
+	var rpcp = global_model.RpcPayload{
+		Method: msg.Msg.GetValue("method").(string),
+		Params: msg.Msg.GetValue("params"),
+	}
+	result, err := rpcp.GetRequestResult()
+	if err != nil {
+		if failureLableNode != nil {
+			return failureLableNode.Handle(msg)
+		} else {
+			return err
+		}
+	}
+	// 判断设备协议，根据不通协议，发送不通内容
+	deviceProtocol := global.MQTTProtocol
+	if msg.Metadata.GetValue("deviceProtocol") != nil && msg.Metadata.GetValue("deviceProtocol").(string) != "" {
+		deviceProtocol = msg.Metadata.GetValue("deviceProtocol").(string)
+	}
+	if deviceProtocol == global.MQTTProtocol {
+		rpc := &mqttclient.RpcRequest{Client: mqttclient.MqttClient}
+		RequestId := n.RequestId
+		if RequestId == 0 {
+			if msg.Metadata.GetValue("requestId") == nil {
+				rpc.GetRequestId()
+			} else {
+				RequestId = int(msg.Metadata.GetValue("requestId").(float64))
+			}
+		} else {
+			rpc.RequestId = RequestId
+		}
+		err = rpc.Pub(result)
+	}
+	if deviceProtocol == global.TCPProtocol {
+		deviceId := msg.Metadata.GetValue("deviceId").(string)
+		err = tcpclient.Send(deviceId, result)
+	}
+	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/action_rpc_request_node.go

@@ -1,47 +0,0 @@
-package nodes
-
-import (
-	"pandax/iothub/client/mqttclient"
-	"pandax/pkg/rule_engine/message"
-)
-
-type rpcRequestNode struct {
-	bareNode
-	Timeout int                   `json:"timeout"`
-	Payload mqttclient.RpcPayload `json:"payload"`
-}
-
-type rpcRequestNodeFactory struct{}
-
-func (f rpcRequestNodeFactory) Name() string     { return "RpcRequestNode" }
-func (f rpcRequestNodeFactory) Category() string { return NODE_CATEGORY_ACTION }
-func (f rpcRequestNodeFactory) Labels() []string { return []string{"Success", "Failure"} }
-func (f rpcRequestNodeFactory) Create(id string, meta Metadata) (Node, error) {
-	node := &rpcRequestNode{
-		bareNode: newBareNode(f.Name(), id, meta, f.Labels()),
-	}
-	return decodePath(meta, node)
-}
-
-func (n *rpcRequestNode) Handle(msg *message.Message) error {
-	successLableNode := n.GetLinkedNode("Success")
-	failureLableNode := n.GetLinkedNode("Failure")
-
-	var rpc = &mqttclient.RpcRequest{Client: mqttclient.MqttClient, Mode: "single", Timeout: n.Timeout}
-	rpc.GetRequestId()
-	respPayload, err := rpc.RequestCmd(n.Payload)
-	if err != nil {
-		if failureLableNode != nil {
-			return failureLableNode.Handle(msg)
-		} else {
-			return err
-		}
-	}
-	msgM := msg.Msg
-	msgM["payload"] = respPayload
-	msg.Msg = msgM
-	if successLableNode != nil {
-		return successLableNode.Handle(msg)
-	}
-	return nil
-}

pkg/rule_engine/nodes/action_rpc_request_to_device_node.go

@@ -0,0 +1,73 @@
+package nodes
+
+import (
+	"encoding/json"
+	"errors"
+	"github.com/sirupsen/logrus"
+	"pandax/iothub/client/mqttclient"
+	"pandax/iothub/client/tcpclient"
+	"pandax/pkg/global"
+	"pandax/pkg/global_model"
+	"pandax/pkg/rule_engine/message"
+)
+
+type rpcRequestToDeviceNode struct {
+	bareNode
+	Timeout int `json:"timeout"`
+}
+
+type rpcRequestToDeviceNodeFactory struct{}
+
+func (f rpcRequestToDeviceNodeFactory) Name() string     { return "RpcRequestToDeviceNode" }
+func (f rpcRequestToDeviceNodeFactory) Category() string { return NODE_CATEGORY_ACTION }
+func (f rpcRequestToDeviceNodeFactory) Labels() []string { return []string{"Success", "Failure"} }
+func (f rpcRequestToDeviceNodeFactory) Create(id string, meta Metadata) (Node, error) {
+	node := &rpcRequestToDeviceNode{
+		bareNode: newBareNode(f.Name(), id, meta, f.Labels()),
+	}
+	return decodePath(meta, node)
+}
+
+func (n *rpcRequestToDeviceNode) Handle(msg *message.Message) error {
+	logrus.Infof("%s handle message '%s'", n.Name(), msg.MsgType)
+	successLableNode := n.GetLinkedNode("Success")
+	failureLableNode := n.GetLinkedNode("Failure")
+	if msg.Msg.GetValue("method") == nil || msg.Msg.GetValue("params") == nil {
+		return errors.New("指令下发格式错误")
+	}
+	var datas = global_model.RpcPayload{
+		Method: msg.Msg.GetValue("method").(string),
+		Params: msg.Msg.GetValue("params"),
+	}
+	payload, _ := json.Marshal(datas)
+	mode := mqttclient.SingleMode
+	if n.Timeout > 0 {
+		mode = mqttclient.DoubleMode
+	}
+	// 判断设备协议，根据不通协议，发送不通内容
+	deviceProtocol := global.MQTTProtocol
+	if msg.Metadata.GetValue("deviceProtocol") != nil && msg.Metadata.GetValue("deviceProtocol").(string) != "" {
+		deviceProtocol = msg.Metadata.GetValue("deviceProtocol").(string)
+	}
+	var err error
+	if deviceProtocol == global.MQTTProtocol {
+		var rpc = &mqttclient.RpcRequest{Client: mqttclient.MqttClient, Mode: mode, Timeout: n.Timeout}
+		rpc.GetRequestId()
+		_, err = rpc.RequestCmd(string(payload))
+	}
+	if deviceProtocol == global.TCPProtocol {
+		deviceId := msg.Metadata.GetValue("deviceId").(string)
+		err = tcpclient.Send(deviceId, string(payload))
+	}
+	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/action_rpc_respond_node.go

@@ -1,49 +0,0 @@
-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

@@ -17,8 +17,8 @@ func (f messageTypeSwitchNodeFactory) Labels() []string {
 		message.RowMes,
 		message.AttributesMes,
 		message.TelemetryMes,
-		message.RpcRequestMes,
-		message.RpcRequestServerMes,
+		message.RpcRequestFromDevice,
+		message.RpcRequestToDevice,
 		message.AlarmMes,
 		message.UpEventMes,
 		message.ConnectMes,

pkg/rule_engine/nodes/filter_switch_node.go

@@ -20,8 +20,8 @@ func (f switchFilterNodeFactory) Labels() []string {
 		message.RowMes,
 		message.AttributesMes,
 		message.TelemetryMes,
-		message.RpcRequestMes,
-		message.RpcRequestServerMes,
+		message.RpcRequestFromDevice,
+		message.RpcRequestToDevice,
 		message.AlarmMes,
 		message.UpEventMes,
 		message.ConnectMes,

pkg/rule_engine/nodes/init.go

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