PandaX/pkg/rule_engine/nodes/external_mqtt_node.go

package nodes

import (
	"encoding/json"
	"fmt"
	"pandax/pkg/rule_engine/message"
	"time"

	mqtt "github.com/eclipse/paho.mqtt.golang"
	"github.com/sirupsen/logrus"
)

type externalMqttNode struct {
	bareNode
	TopicPattern      string `json:"topicPattern"`
	Host              string `json:"host"`
	Port              string `json:"port"`
	ConnectTimeoutSec int    `json:"connectTimeoutSec"`
	ClientId          string `json:"clientId"`
	CleanSession      bool   `json:"cleanSession"`
	Ssl               bool   `json:"ssl"`
	MqttCli           mqtt.Client
}

type externalMqttNodeFactory struct{}

func (f externalMqttNodeFactory) Name() string     { return "ExternalMqttNode" }
func (f externalMqttNodeFactory) Category() string { return NODE_CATEGORY_EXTERNAL }
func (f externalMqttNodeFactory) Labels() []string { return []string{"Success", "Failure"} }
func (f externalMqttNodeFactory) Create(id string, meta Metadata) (Node, error) {

	node := &externalMqttNode{
		bareNode: newBareNode(f.Name(), id, meta, f.Labels()),
	}
	broker := fmt.Sprintf("tcp://%s:%s", node.Host, node.Port)
	opts := mqtt.NewClientOptions().AddBroker(broker)
	opts.SetClientID(node.ClientId)
	opts.SetCleanSession(node.CleanSession)
	opts.SetConnectTimeout(time.Duration(node.ConnectTimeoutSec) * time.Second)
	node.MqttCli = mqtt.NewClient(opts)

	if token := node.MqttCli.Connect(); token.Wait() && token.Error() != nil {
		logrus.WithError(token.Error())
		return nil, token.Error()
	}
	return decodePath(meta, node)
}

func (n *externalMqttNode) Handle(msg message.Message) error {
	logrus.Infof("%s handle message '%s'", n.Name(), msg.GetType())

	successLabelNode := n.GetLinkedNode("Success")
	failureLabelNode := n.GetLinkedNode("Failure")
	topic := n.TopicPattern //need fix add msg.metadata in it
	sendmqttmsg, err := json.Marshal(msg.GetMsg())
	if err != nil {
		return nil
	}
	token := n.MqttCli.Publish(topic, 1, false, sendmqttmsg)
	if token.Wait() && token.Error() != nil {
		fmt.Println(token.Error())
		return failureLabelNode.Handle(msg)
	}
	return successLabelNode.Handle(msg)
}