规则引擎

pkg/rule_engine/nodes/action_delay_node.go

@@ -0,0 +1,76 @@
+package nodes
+
+import (
+	"fmt"
+	"pandax/pkg/rule_engine/message"
+	"sync"
+	"time"
+
+	"github.com/sirupsen/logrus"
+)
+
+const DelayNodeName = "DelayNode"
+
+type delayNode struct {
+	bareNode
+	PeriodTs           int               `json:"periodTs" yaml:"periodTs" jpath:"periodTs"`
+	MaxPendingMessages int               `json:"maxPendingMessages" yaml:"maxPendingMessages" jpath:"maxPendingMessages"`
+	messageQueue       []message.Message `jpath:"-"`
+	delayTimer         *time.Timer       `jpath:"-"`
+	lock               sync.Mutex        `jpath:"-"`
+}
+
+type delayNodeFactory struct{}
+
+func (f delayNodeFactory) Name() string     { return DelayNodeName }
+func (f delayNodeFactory) Category() string { return NODE_CATEGORY_ACTION }
+
+func (f delayNodeFactory) Create(id string, meta Metadata) (Node, error) {
+	labels := []string{"Success", "Failure"}
+	node := &delayNode{
+		bareNode: newBareNode(f.Name(), id, meta, labels),
+		lock:     sync.Mutex{},
+	}
+	_, err := decodePath(meta, node)
+	node.messageQueue = make([]message.Message, node.MaxPendingMessages)
+	return node, err
+}
+
+func (n *delayNode) Handle(msg message.Message) error {
+	logrus.Infof("%s handle message '%s'", n.Name(), msg.GetType())
+
+	successLabelNode := n.GetLinkedNode("Success")
+	failureLabelNode := n.GetLinkedNode("Failure")
+	if successLabelNode == nil || failureLabelNode == nil {
+		return fmt.Errorf("no valid label linked node in %s", n.Name())
+	}
+
+	// check wethere the time had already been started, queue message if started
+	if n.delayTimer == nil {
+		n.messageQueue = append(n.messageQueue, msg)
+		n.delayTimer = time.NewTimer(time.Duration(n.PeriodTs) * time.Second)
+		// start timecallback goroutine
+		go func(n *delayNode) error {
+			defer n.delayTimer.Stop()
+			for {
+				<-n.delayTimer.C
+				n.lock.Lock()
+				defer n.lock.Unlock()
+				if len(n.messageQueue) > 0 {
+					msg := n.messageQueue[0]
+					n.messageQueue = n.messageQueue[0:]
+					return successLabelNode.Handle(msg)
+				}
+			}
+		}(n)
+		return nil
+	}
+	// the delay timer had already been created, just queue message
+	n.lock.Lock()
+	defer n.lock.Unlock()
+	if len(n.messageQueue) == n.MaxPendingMessages {
+		return failureLabelNode.Handle(msg)
+	}
+	n.messageQueue = append(n.messageQueue, msg)
+	return nil
+}

pkg/rule_engine/nodes/factory.go

@@ -0,0 +1,53 @@
+package nodes
+
+import (
+	"fmt"
+)
+
+const (
+	NODE_CATEGORY_FILTER     = "filter"
+	NODE_CATEGORY_ACTION     = "action"
+	NODE_CATEGORY_ENRICHMENT = "enrichment"
+	NODE_CATEGORY_TRANSFORM  = "transform"
+	NODE_CATEGORY_EXTERNAL   = "external"
+	NODE_CATEGORY_OTHERS     = "others"
+)
+
+// Factory is node's factory to create node based on metadata
+// factory also manage node's metadta description which can be used by other
+// service to present node in web
+type Factory interface {
+	Name() string
+	Category() string
+	Create(id string, meta Metadata) (Node, error)
+}
+
+var (
+	// allNodeFactories hold all node's factory
+	allNodeFactories map[string]Factory = make(map[string]Factory)
+
+	// allNodeCategories hold node's metadata by category
+	allNodeCategories map[string][]string = make(map[string][]string)
+)
+
+// RegisterFactory add a new node factory and classify its category for
+// metadata description
+func RegisterFactory(f Factory) {
+	allNodeFactories[f.Name()] = f
+
+	if allNodeCategories[f.Category()] == nil {
+		allNodeCategories[f.Category()] = []string{}
+	}
+	allNodeCategories[f.Category()] = append(allNodeCategories[f.Category()], f.Name())
+}
+
+// NewNode is the only way to create a new node
+func NewNode(nodeType string, id string, meta Metadata) (Node, error) {
+	if f, found := allNodeFactories[nodeType]; found {
+		return f.Create(id, meta)
+	}
+	return nil, fmt.Errorf("invalid node type '%s'", nodeType)
+}
+
+// GetCategoryNodes return specified category's all nodes
+func GetCategoryNodes() map[string][]string { return allNodeCategories }

pkg/rule_engine/nodes/filter_switch_node.go

@@ -0,0 +1,43 @@
+package nodes
+
+import (
+	"github.com/sirupsen/logrus"
+	"pandax/pkg/rule_engine/message"
+)
+
+type switchFilterNode struct {
+	bareNode
+	Scripts string `json:"scripts" yaml:"scripts"`
+}
+
+type switchFilterNodeFactory struct{}
+
+func (f switchFilterNodeFactory) Name() string     { return "SwitchNode" }
+func (f switchFilterNodeFactory) Category() string { return NODE_CATEGORY_FILTER }
+
+func (f switchFilterNodeFactory) Create(id string, meta Metadata) (Node, error) {
+	labels := []string{}
+	node := &switchFilterNode{
+		bareNode: newBareNode(f.Name(), id, meta, labels),
+	}
+	return decodePath(meta, node)
+}
+
+func (n *switchFilterNode) Handle(msg message.Message) error {
+	logrus.Infof("%s handle message '%s'", n.Name(), msg.GetType())
+
+	/*scriptEngine := NewScriptEngine()
+	SwitchResults, err := scriptEngine.ScriptOnSwitch(msg, n.Scripts)
+	if err != nil {
+		return nil
+	}
+	nodes := n.GetLinkedNodes()
+	for label, node := range nodes {
+		for _, switchresult := range SwitchResults {
+			if label == switchresult {
+				return node.Handle(msg)
+			}
+		}
+	}*/
+	return nil
+}

pkg/rule_engine/nodes/init.go

@@ -0,0 +1,6 @@
+package nodes
+
+// init register all node's factory
+func init() {
+	RegisterFactory(inputNodeFactory{})
+}

pkg/rule_engine/nodes/input_node.go

@@ -0,0 +1,35 @@
+package nodes
+
+import (
+	"github.com/sirupsen/logrus"
+	"pandax/pkg/rule_engine/message"
+)
+
+const InputNodeName = "InputNode"
+
+type inputNode struct {
+	bareNode
+}
+
+type inputNodeFactory struct{}
+
+func (f inputNodeFactory) Name() string     { return "InputNode" }
+func (f inputNodeFactory) Category() string { return NODE_CATEGORY_OTHERS }
+
+func (f inputNodeFactory) Create(id string, meta Metadata) (Node, error) {
+	labels := []string{}
+	node := &inputNode{
+		bareNode: newBareNode(InputNodeName, id, meta, labels),
+	}
+	return node, nil
+}
+
+func (n *inputNode) Handle(msg message.Message) error {
+	logrus.Infof("%s handle message '%s'", n.Name(), msg.GetType())
+
+	nodes := n.GetLinkedNodes()
+	for _, node := range nodes {
+		return node.Handle(msg)
+	}
+	return nil
+}

pkg/rule_engine/nodes/metadata.go

@@ -0,0 +1,62 @@
+package nodes
+
+import (
+	"fmt"
+	"github.com/XM-GO/PandaKit/utils"
+)
+
+const (
+	NODE_CONFIG_MESSAGE_TYPE_KEY    = "messageTypeKey"
+	NODE_CONFIG_ORIGINATOR_TYPE_KEY = "originatorTypeKey"
+)
+
+type Metadata interface {
+	Keys() []string
+	With(key string, val interface{}) Metadata
+	Value(key string) (interface{}, error)
+	DecodePath(rawVal interface{}) error
+}
+
+type nodeMetadata struct {
+	keypairs map[string]interface{}
+}
+
+func NewMetadata() Metadata {
+	return &nodeMetadata{
+		keypairs: make(map[string]interface{}),
+	}
+}
+
+func NewMetadataWithString(vals string) Metadata {
+	return &nodeMetadata{}
+}
+
+func NewMetadataWithValues(vals map[string]interface{}) Metadata {
+	return &nodeMetadata{
+		keypairs: vals,
+	}
+}
+
+func (c *nodeMetadata) Keys() []string {
+	keys := []string{}
+	for key, _ := range c.keypairs {
+		keys = append(keys, key)
+	}
+	return keys
+}
+
+func (c *nodeMetadata) Value(key string) (interface{}, error) {
+	if val, found := c.keypairs[key]; found {
+		return val, nil
+	}
+	return nil, fmt.Errorf("key '%s' not found", key)
+}
+
+func (c *nodeMetadata) With(key string, val interface{}) Metadata {
+	c.keypairs[key] = val
+	return c
+}
+
+func (c *nodeMetadata) DecodePath(rawVal interface{}) error {
+	return utils.Map2Struct(c.keypairs, rawVal)
+}

pkg/rule_engine/nodes/node.go

@@ -0,0 +1,65 @@
+package nodes
+
+import (
+	"errors"
+	"pandax/pkg/rule_engine/message"
+
+	"github.com/sirupsen/logrus"
+)
+
+type Node interface {
+	Name() string
+	Id() string
+	Metadata() Metadata
+	MustLabels() []string
+	Handle(message.Message) error
+
+	AddLinkedNode(label string, node Node)
+	GetLinkedNode(label string) Node
+	GetLinkedNodes() map[string]Node
+}
+
+type bareNode struct {
+	name   string
+	id     string
+	nodes  map[string]Node
+	meta   Metadata
+	labels []string
+}
+
+func newBareNode(name string, id string, meta Metadata, labels []string) bareNode {
+	return bareNode{
+		name:   name,
+		id:     id,
+		nodes:  make(map[string]Node),
+		meta:   meta,
+		labels: labels,
+	}
+}
+
+func (n *bareNode) Name() string                          { return n.name }
+func (n *bareNode) WithId(id string)                      { n.id = id }
+func (n *bareNode) Id() string                            { return n.id }
+func (n *bareNode) MustLabels() []string                  { return n.labels }
+func (n *bareNode) AddLinkedNode(label string, node Node) { n.nodes[label] = node }
+
+func (n *bareNode) GetLinkedNode(label string) Node {
+	if node, found := n.nodes[label]; found {
+		return node
+	}
+	logrus.Errorf("no label '%s' in node '%s'", label, n.name)
+	return nil
+}
+
+func (n *bareNode) GetLinkedNodes() map[string]Node { return n.nodes }
+
+func (n *bareNode) Metadata() Metadata { return n.meta }
+
+func (n *bareNode) Handle(message.Message) error { return errors.New("not implemented") }
+
+func decodePath(meta Metadata, n Node) (Node, error) {
+	if err := meta.DecodePath(n); err != nil {
+		return n, err
+	}
+	return n, nil
+}