规则引擎完善

pkg/actors/app_actor.go

@@ -2,29 +2,21 @@ package actors
 
 import (
 	"github.com/gogf/gf/util/guid"
-	"sparrow/pkg/actor"
 	"sparrow/pkg/protocol"
+	"sparrow/pkg/ruleEngine"
 	"sparrow/pkg/server"
 )
 
 // AppActor 服务级actor
 type AppActor struct {
-	actor.ContextBasedCreator
+	ruleEngine.ContextBasedCreator
 }
 
-func (a *AppActor) CreateActorId() string {
-	panic("implement me")
-}
-
-func (a *AppActor) CreateActor() actor.Actor {
-	panic("implement me")
-}
-
-func (a *AppActor) GetActorRef() actor.Ref {
+func (a *AppActor) GetActorRef() ruleEngine.Ref {
 	return a.Ctx
 }
 
-func (a *AppActor) Init(ctx actor.Ctx) error {
+func (a *AppActor) Init(ctx ruleEngine.Ctx) error {
 	a.Ctx = ctx
 	return nil
 }
@@ -32,31 +24,69 @@ func (a *AppActor) Init(ctx actor.Ctx) error {
 func (a *AppActor) Process(msg protocol.ActorMsg) error {
 	switch msg.GetMessageType() {
 	case protocol.APP_INIT_MSG:
-		server.Log.Debugf("收到应用初始化消息")
+		if err := a.initTenants(); err != nil {
+			server.Log.Error(err)
+		}
+	case protocol.QUEUE_TO_RULE_ENGINE_MSG:
+		return a.onQueueToRuleEngineMsg(msg.(*ruleEngine.QueueToRuleEngineMsg))
 	default:
 		server.Log.Debugf("未知的消息类型：%s", msg.GetMessageType())
 	}
 	return nil
 }
 
+func (a *AppActor) onQueueToRuleEngineMsg(msg *ruleEngine.QueueToRuleEngineMsg) error {
+	if ref, err := a.getOrCreateTenantActor(msg.TenantId); err == nil {
+		ref.Tell(msg)
+	} else {
+		return err
+	}
+	msg.Message.GetCallBack().OnSuccess()
+	return nil
+}
+
+func (a *AppActor) initTenants() error {
+	server.Log.Debug("starting main actor")
+	tenants, err := a.SystemCtx.TenantService.FindTenants()
+	if err != nil {
+		return err
+	}
+	for _, t := range tenants {
+		server.Log.Debugf("creating tenant actor :%s,%s", t.Id, t.Name)
+		_, err := a.getOrCreateTenantActor(t.Id)
+		if err != nil {
+			server.Log.Error(err)
+			continue
+		}
+		server.Log.Debugf("tenant actor :%s,%s created", t.Id, t.Name)
+	}
+	server.Log.Debugln("main actor started")
+	return nil
+}
+
+func (a *AppActor) getOrCreateTenantActor(tId string) (ruleEngine.Ref, error) {
+	return a.Ctx.GetOrCreateChildActor(tId, ruleEngine.TENANT_DISPATCHER_NAME,
+		NewTenantActorCreator(a.SystemCtx, tId))
+}
+
 func (a *AppActor) Destroy() error {
 	return nil
 }
 
-func (a *AppActor) OnProcessFailure(err error) *actor.ProcessFailureStrategy {
+func (a *AppActor) OnProcessFailure(err error) *ruleEngine.ProcessFailureStrategy {
 	if err != nil {
-		return actor.Stop()
+		return ruleEngine.Stop()
 	} else {
-		return actor.Resume()
+		return ruleEngine.Resume()
 	}
 }
 
 // AppActorCreator app actor creator implements creator interface
 type AppActorCreator struct {
-	actor.ContextBasedCreator
+	ruleEngine.ContextBasedCreator
 }
 
-func NewAppActorCreator(systemCtx *actor.SystemContext) *AppActorCreator {
+func NewAppActorCreator(systemCtx *ruleEngine.SystemContext) *AppActorCreator {
 	ins := new(AppActorCreator)
 	ins.SystemCtx = systemCtx
 	return ins
@@ -66,7 +96,7 @@ func (a *AppActorCreator) CreateActorId() string {
 	return guid.S()
 }
 
-func (a *AppActorCreator) CreateActor() actor.Actor {
+func (a *AppActorCreator) CreateActor() ruleEngine.Actor {
 	appC := new(AppActor)
 	appC.SystemCtx = a.SystemCtx
 	return appC

pkg/actors/rule_chain_actor.go

@@ -3,7 +3,6 @@ package actors
 import (
 	"errors"
 	"fmt"
-	"sparrow/pkg/actor"
 	"sparrow/pkg/entities"
 	"sparrow/pkg/protocol"
 	"sparrow/pkg/queue"
@@ -12,39 +11,28 @@ import (
 	"strings"
 )
 
-var ruleNodes = map[string]*ruleEngine.RuleNode{
-	"1": {
-		RuleNodeId:  "1",
-		RuleChainId: "11",
-		Type:        "MsgTypeFilterNode",
-		Name:        "simple node",
-		IsDebug:     true,
-		Config:      "",
-	},
-}
-
 type RuleChainActor struct {
-	actor.ContextBasedCreator
+	ruleEngine.ContextBasedCreator
 	ruleChain   *ruleEngine.RuleChain
 	tenantId    string
 	firstId     string
 	firstNode   *ruleEngine.RuleNodeCtx
 	started     bool
 	ruleChainId string
-	parent      actor.Ref
+	parent      ruleEngine.Ref
 	nodeActors  map[string]*ruleEngine.RuleNodeCtx
 	nodeRoutes  map[string][]*ruleEngine.RuleNodeRelation
 
-	ruleChainName string
-
-	clusterService queue.ClusterService
+	ruleChainName  string
+	clusterService ruleEngine.ClusterService
+	state          ruleEngine.ComponentLifecycleState
 }
 
 func newRuleChainActor(
-	sysCtx *actor.SystemContext,
+	sysCtx *ruleEngine.SystemContext,
 	ruleChain *ruleEngine.RuleChain,
 	tenantId string,
-	parent actor.Ref,
+	parent ruleEngine.Ref,
 ) *RuleChainActor {
 	item := &RuleChainActor{
 		ruleChainId:    ruleChain.ChainId,
@@ -59,28 +47,38 @@ func newRuleChainActor(
 	return item
 }
 
-func (r *RuleChainActor) GetActorRef() actor.Ref {
+func (r *RuleChainActor) GetActorRef() ruleEngine.Ref {
 	return r.Ctx
 }
 
-func (r *RuleChainActor) Init(ctx actor.Ctx) error {
+func (r *RuleChainActor) Init(ctx ruleEngine.Ctx) error {
 	if r.ruleChain != nil {
 		r.ruleChainName = r.ruleChain.Name
 	}
 	r.Ctx = ctx
-	return nil
+	return r.start()
 }
 
 func (r *RuleChainActor) Process(msg protocol.ActorMsg) error {
 	switch msg.GetMessageType() {
 	case protocol.QUEUE_TO_RULE_ENGINE_MSG:
-		return r.onQueueToRuleEngineMsg(msg.(*actor.QueueToRuleEngineMsg))
-
+		return r.onQueueToRuleEngineMsg(msg.(*ruleEngine.QueueToRuleEngineMsg))
+	case protocol.RULE_TO_RULE_CHAIN_TELL_NEXT_MSG:
+		return r.onTellNextRuleNode(msg.(*ruleEngine.RuleNodeToRuleChanTellNextMsg))
+	case protocol.RULE_CHAIN_TO_RULE_CHAIN_MSG:
+		return r.onRuleChainToRuleChain(msg.(*ruleEngine.RuleChainToRuleChainMsg))
 	}
 	return nil
 }
-
-func (r *RuleChainActor) onQueueToRuleEngineMsg(msg *actor.QueueToRuleEngineMsg) error {
+func (r *RuleChainActor) onRuleChainToRuleChain(msg *ruleEngine.RuleChainToRuleChainMsg) error {
+	if r.firstNode != nil {
+		r.pushMsgToNode(r.firstNode, msg.Message, msg.FromRelationType)
+	} else {
+		msg.Message.GetCallBack().OnSuccess()
+	}
+	return nil
+}
+func (r *RuleChainActor) onQueueToRuleEngineMsg(msg *ruleEngine.QueueToRuleEngineMsg) error {
 	actorMsg := msg.Message
 	server.Log.Debugf("Processing message")
 	if len(msg.RelationTypes) == 0 {
@@ -88,11 +86,12 @@ func (r *RuleChainActor) onQueueToRuleEngineMsg(msg *actor.QueueToRuleEngineMsg)
 		var targetCtx *ruleEngine.RuleNodeCtx
 		if ruleNodeId == "" {
 			targetCtx = r.firstNode
+			actorMsg = actorMsg.CopyWithRuleChainId(r.ruleChainId)
 		} else {
 			targetCtx = r.nodeActors[ruleNodeId]
 		}
 		if targetCtx != nil {
-			server.Log.Debugf("pushing message to target rule node,%s, %s", r.ruleChainId, ruleNodeId)
+			server.Log.Debugf("pushing message to target rule node,%s, %s", r.ruleChainId, targetCtx.Self.RuleNodeId)
 			r.pushMsgToNode(targetCtx, actorMsg, "")
 		} else {
 			server.Log.Debugf("Rule node dose not exist. probably old message,%s, %s", r.ruleChainId, ruleNodeId)
@@ -104,11 +103,20 @@ func (r *RuleChainActor) onQueueToRuleEngineMsg(msg *actor.QueueToRuleEngineMsg)
 	return nil
 }
 
+func (r *RuleChainActor) onTellNextRuleNode(msg *ruleEngine.RuleNodeToRuleChanTellNextMsg) error {
+	var errStr string
+	if msg.FailureMessage != nil {
+		errStr = msg.FailureMessage.Error()
+	}
+	r.onTellNext(msg.Message, msg.RuleNodeId, msg.RelationTypes.ToStrArray(), errStr)
+	return nil
+}
+
 // on tell next actor
 func (r *RuleChainActor) onTellNext(msg *protocol.Message, originatorNodeId string,
 	relationTypes []string, errMsg string) {
 	originatorId := msg.Originator
-	tpi := queue.ResolvePartition(queue.RULE_ENGINE, msg.GetQueueName(), r.tenantId, originatorId)
+	tpi := queue.ResolvePartition(ruleEngine.RULE_ENGINE, msg.GetQueueName(), r.tenantId, originatorId)
 	var relations []*ruleEngine.RuleNodeRelation
 	if rs, ok := r.nodeRoutes[originatorNodeId]; ok {
 		for _, item := range rs {
@@ -135,7 +143,6 @@ func (r *RuleChainActor) onTellNext(msg *protocol.Message, originatorNodeId stri
 			r.pushMsgToTarget(tpi, msg, rl.Out, rl.Type)
 		}
 	} else {
-
 		for _, rl := range relations {
 			target := rl.Out
 			r.putToQueue(tpi, msg, queue.NewMultipleMsgCallbackWrapper(int32(len(relations)), msg.GetCallBack()), target)
@@ -151,7 +158,7 @@ func (r *RuleChainActor) pushMsgToTarget(tpi *queue.TopicPartitionInfo, msg *pro
 			targetCtx := r.nodeActors[entityId.GetId()]
 			r.pushMsgToNode(targetCtx, msg, fromRelationType)
 		case entities.RULE_CHAIN:
-			r.parent.Tell(&actor.RuleChainToRuleChainMsg{
+			r.parent.Tell(&ruleEngine.RuleChainToRuleChainMsg{
 				TargetId:         entityId.GetId(),
 				SourceId:         r.ruleChainId,
 				Message:          msg,
@@ -196,7 +203,7 @@ func contains(relations []string, relation string) bool {
 // push a message to node actor
 func (r *RuleChainActor) pushMsgToNode(targetCtx *ruleEngine.RuleNodeCtx, msg *protocol.Message, relationType string) {
 	if targetCtx != nil {
-		targetCtx.SelfActor.Tell(&actor.RuleChainToRuleNodeMsg{
+		targetCtx.SelfActor.Tell(&ruleEngine.RuleChainToRuleNodeMsg{
 			Message:          msg,
 			Ctx:              ruleEngine.NewDefaultContext(targetCtx, r.SystemCtx),
 			FromRelationType: relationType,
@@ -210,12 +217,79 @@ func (r *RuleChainActor) Destroy() error {
 	return nil
 }
 
-func (r *RuleChainActor) OnProcessFailure(err error) *actor.ProcessFailureStrategy {
+func (r *RuleChainActor) OnProcessFailure(err error) *ruleEngine.ProcessFailureStrategy {
 	if err != nil {
-		return actor.Stop()
+		return ruleEngine.Stop()
+	} else {
+		return ruleEngine.Resume()
+	}
+}
+
+func (r *RuleChainActor) start() error {
+	if !r.started {
+		ruleChain, err := r.SystemCtx.RuleChainService.FindRuleChainById(r.tenantId, r.ruleChainId)
+		if err != nil {
+			return err
+		}
+		if ruleChain != nil {
+			nodes, err := r.SystemCtx.RuleChainService.GetRuleChainNodes(r.tenantId, ruleChain.ChainId)
+			if err != nil {
+				return err
+			}
+			server.Log.Debugf("starting rule chain with %d nodes", len(nodes))
+			for _, node := range nodes {
+				server.Log.Debugf("creating rule node actor:%s,%s", node.RuleNodeId, node.Name)
+				ref, err := r.createNodeActor(node.RuleNodeId)
+				if err != nil {
+					continue
+				}
+				r.nodeActors[node.RuleNodeId] = &ruleEngine.RuleNodeCtx{
+					TenantId:   r.tenantId,
+					ChainActor: r.Ctx,
+					SelfActor:  ref,
+					Self:       node,
+				}
+				r.started = true
+			}
+			r.initRoutes(r.ruleChain, nodes)
+		}
 	} else {
-		return actor.Resume()
+		return r.update()
+	}
+	return nil
+}
+
+func (r *RuleChainActor) initRoutes(ruleChain *ruleEngine.RuleChain, nodes []*ruleEngine.RuleNode) {
+	for _, node := range nodes {
+		relations, err := r.SystemCtx.RuleChainService.GetRuleNodeRelations(r.tenantId, node.RuleNodeId)
+		if err != nil {
+			continue
+		}
+		var rs []*ruleEngine.RuleNodeRelation
+		for _, relation := range relations {
+
+			rs = append(rs, &ruleEngine.RuleNodeRelation{
+				In:   &entities.RuleNodeId{Id: node.RuleNodeId},
+				Out:  &entities.RuleNodeId{Id: relation.To},
+				Type: relation.Type,
+			})
+		}
+		r.nodeRoutes[node.RuleNodeId] = rs
 	}
+	r.firstId = ruleChain.FirstNodeId
+	r.firstNode = r.nodeActors[r.firstId]
+	r.state = ruleEngine.ACTIVE
+}
+
+func (r *RuleChainActor) update() error {
+	return nil
+}
+
+func (r *RuleChainActor) createNodeActor(nodeId string) (ruleEngine.Ref, error) {
+	return r.Ctx.GetOrCreateChildActor(nodeId,
+		ruleEngine.RULE_DISPATCHER_NAME,
+		NewRuleNodeActorCreator(r.SystemCtx, r.tenantId,
+			r.ruleChainId, r.ruleChainName, nodeId, r.Ctx))
 }
 
 // RuleChainCreator
@@ -225,10 +299,10 @@ type RuleChainCreator struct {
 
 //NewRuleChainCreator create a instance
 func NewRuleChainCreator(
-	sysCtx *actor.SystemContext,
+	sysCtx *ruleEngine.SystemContext,
 	tenantId string,
 	ruleChan *ruleEngine.RuleChain,
-	parent actor.Ref,
+	parent ruleEngine.Ref,
 ) *RuleChainCreator {
 	item := &RuleChainCreator{}
 	item.tenantId = tenantId
@@ -242,6 +316,6 @@ func (r *RuleChainCreator) CreateActorId() string {
 	return r.ruleChain.ChainId
 }
 
-func (r *RuleChainCreator) CreateActor() actor.Actor {
+func (r *RuleChainCreator) CreateActor() ruleEngine.Actor {
 	return newRuleChainActor(r.SystemCtx, r.ruleChain, r.tenantId, r.parent)
 }

pkg/actors/rule_node_actor.go

@@ -1 +1,157 @@
 package actors
+
+import (
+	"errors"
+	"fmt"
+	"sparrow/pkg/protocol"
+	"sparrow/pkg/ruleEngine"
+	"sparrow/pkg/ruleEngine/nodes"
+	"sparrow/pkg/server"
+)
+
+type RuleNodeActor struct {
+	ruleChainName   string
+	self            ruleEngine.Ref
+	ruleNode        *ruleEngine.RuleNode
+	node            ruleEngine.Node
+	ruleChainNodeId string
+	ruleNodeId      string
+	ruleEngine.ContextBasedCreator
+	tenantId       string
+	defaultContext ruleEngine.Context
+	parent         ruleEngine.Ref
+
+	info *protocol.RuleNodeInfo
+}
+
+func (r *RuleNodeActor) GetActorRef() ruleEngine.Ref {
+	return r.Ctx
+}
+
+func (r *RuleNodeActor) Init(ctx ruleEngine.Ctx) error {
+	r.Ctx = ctx
+	result, err := r.SystemCtx.RuleChainService.FindRuleNodeById(r.tenantId, r.ruleNodeId)
+	if err != nil {
+		return err
+	}
+	r.ruleNode = result
+	r.defaultContext = ruleEngine.NewDefaultContext(&ruleEngine.RuleNodeCtx{
+		TenantId:   "",
+		ChainActor: nil,
+		SelfActor:  nil,
+		Self:       nil,
+	}, r.SystemCtx)
+	r.info = &protocol.RuleNodeInfo{
+		RuleNodeId:    r.ruleNodeId,
+		RuleChainName: r.ruleChainName,
+		RuleNodeName:  r.ruleNode.Name,
+	}
+	node, err := r.initComponent(r.ruleNode)
+	if err != nil {
+		return err
+	}
+	if node != nil {
+		r.node = node
+	}
+	return nil
+}
+
+// 实例化node
+func (r *RuleNodeActor) initComponent(ruleNode *ruleEngine.RuleNode) (ruleEngine.Node, error) {
+	if ruleNode != nil {
+		s, ok := nodes.CreateNodeByConfig(ruleNode.Type, r.defaultContext, r.ruleNode.Config)
+		if !ok {
+			return nil, errors.New(fmt.Sprintf("create node instance faild, %s", ruleNode.Type))
+		}
+		return s.(ruleEngine.Node), nil
+	}
+	return nil, nil
+}
+
+func (r *RuleNodeActor) Process(msg protocol.ActorMsg) error {
+	switch msg.GetMessageType() {
+	case protocol.RULE_CHAIN_TO_RULE_MSG:
+		return r.onRuleChainToNodeMsg(msg.(*ruleEngine.RuleChainToRuleNodeMsg))
+	case protocol.RULE_TO_SELF_MSG:
+		return r.onRuleToSelfMsg(msg.(*ruleEngine.RuleToSelfMsg))
+	case protocol.RULE_TO_SELF_ERROR_MSG:
+		return r.onRuleToSelfErrorMsg(msg.(*ruleEngine.RuleToSelfErrorMsg))
+	default:
+		return errors.New("未知的消息类型")
+	}
+}
+
+func (r *RuleNodeActor) onRuleToSelfErrorMsg(msg *ruleEngine.RuleToSelfErrorMsg) error {
+	server.Log.Error(msg.Err)
+	return nil
+}
+
+// TODO:处理消息数量支持实现以及最大处理能力逻辑
+func (r *RuleNodeActor) onRuleToSelfMsg(msg *ruleEngine.RuleToSelfMsg) error {
+	server.Log.Debugf("Going to process rule msg:%s,%s", r.ruleChainName, r.ruleNode.Name)
+	actorMsg := msg.Message
+	if err := r.node.OnMessage(r.defaultContext, actorMsg); err != nil {
+		r.defaultContext.TellError(actorMsg, errors.New("onRuleToSelfMsg error"))
+		return err
+	}
+	return nil
+}
+
+func (r *RuleNodeActor) onRuleChainToNodeMsg(msg *ruleEngine.RuleChainToRuleNodeMsg) error {
+	server.Log.Debugf("Going to process rule msg:%s,%s", r.ruleChainName, r.ruleNode.Name)
+	msg.Message.GetCallBack().OnProcessingStart(r.info)
+	actorMsg := msg.Message
+	// ruleNodeCount := actorMsg.GetAndIncrementRuleNodeCounter()
+	err := r.node.OnMessage(msg.Ctx, actorMsg)
+	if err != nil {
+		msg.Ctx.TellError(actorMsg, errors.New("onRuleChainToNodeMsg error"))
+		return err
+	}
+	return nil
+}
+
+func (r *RuleNodeActor) Destroy() error {
+	panic("implement me")
+}
+
+func (r *RuleNodeActor) OnProcessFailure(err error) *ruleEngine.ProcessFailureStrategy {
+	panic("implement me")
+}
+
+type RuleNodeActorCreator struct {
+	RuleNodeActor
+}
+
+func NewRuleNodeActorCreator(
+	sysCtx *ruleEngine.SystemContext,
+	tenantId string,
+	ruleChainId string,
+	ruleChainName string,
+	ruleNodeId string,
+	parent ruleEngine.Ref,
+) *RuleNodeActorCreator {
+	item := new(RuleNodeActorCreator)
+	item.SystemCtx = sysCtx
+	item.ruleNodeId = ruleNodeId
+	item.ruleChainName = ruleChainName
+	item.tenantId = tenantId
+	item.ruleChainNodeId = ruleChainId
+	item.parent = parent
+	return item
+}
+
+func (r *RuleNodeActorCreator) CreateActorId() string {
+	return r.ruleNodeId
+}
+
+func (r *RuleNodeActorCreator) CreateActor() ruleEngine.Actor {
+	item := &RuleNodeActor{
+		ruleChainName:   r.ruleChainName,
+		ruleChainNodeId: r.ruleChainNodeId,
+		ruleNodeId:      r.ruleNodeId,
+		tenantId:        r.tenantId,
+		parent:          r.parent,
+	}
+	item.SystemCtx = r.SystemCtx
+	return item
+}

pkg/actors/tenant_actor.go

@@ -2,44 +2,27 @@ package actors
 
 import (
 	"errors"
-	"sparrow/pkg/actor"
+	"sparrow/pkg/entities"
 	"sparrow/pkg/protocol"
 	"sparrow/pkg/ruleEngine"
 	"sparrow/pkg/server"
 )
 
-// TODO: 先用测试数据
-var ruleChains = map[string]*ruleEngine.RuleChain{
-	"11": {
-		TenantId:    "1",
-		Name:        "Chain1",
-		FirstNodeId: "1",
-		IsRoot:      true,
-		IsDebug:     false,
-		Config:      "",
-		ChainId:     "11",
-	},
-	"22": {
-		TenantId:    "2",
-		Name:        "Chain2",
-		FirstNodeId: "1",
-		IsRoot:      false,
-		IsDebug:     false,
-		Config:      "",
-		ChainId:     "22",
-	},
-}
-
 // TenantActor 租户 actor
 type TenantActor struct {
-	actor.ContextBasedCreator
-	tenantId       string
-	rootChain      *ruleEngine.RuleChain
-	rootChainActor actor.Ref
-	cantFindTenant bool
+	ruleEngine.ContextBasedCreator
+	tenantId         string
+	rootChain        *ruleEngine.RuleChain
+	rootChainActor   ruleEngine.Ref
+	cantFindTenant   bool
+	ruleChainService ruleEngine.RuleChainService
 }
 
-func (t *TenantActor) initRuleChains() {
+func (t *TenantActor) initRuleChains() error {
+	ruleChains, err := t.SystemCtx.RuleChainService.FindRuleChains(t.tenantId)
+	if err != nil {
+		return err
+	}
 	for _, ruleChain := range ruleChains {
 		server.Log.Debugf("Creating rule chain actor:%s", ruleChain.ChainId)
 		actorRef, err := t.getOrCreateActor(ruleChain.ChainId, ruleChain)
@@ -53,58 +36,107 @@ func (t *TenantActor) initRuleChains() {
 		}
 		server.Log.Debugf("Rule chain actor created:%s", ruleChain.ChainId)
 	}
+	return nil
 }
 
-func (t *TenantActor) destroyRuleChains() {
+func (t *TenantActor) destroyRuleChains() error {
+	ruleChains, err := t.ruleChainService.FindRuleChains(t.tenantId)
+	if err != nil {
+		return err
+	}
 	for _, ruleChain := range ruleChains {
 		_ = t.Ctx.Stop(ruleChain.ChainId)
 	}
+	return nil
 }
 
-func (t *TenantActor) getOrCreateActor(ruleChainId string, ruleChain *ruleEngine.RuleChain) (actor.Ref, error) {
+func (t *TenantActor) getOrCreateActor(ruleChainId string, ruleChain *ruleEngine.RuleChain) (ruleEngine.Ref, error) {
 	return t.Ctx.GetOrCreateChildActor(ruleChainId,
-		actor.RULE_DISPATCHER_NAME,
+		ruleEngine.RULE_DISPATCHER_NAME,
 		NewRuleChainCreator(t.SystemCtx, t.tenantId, ruleChain, t.Ctx.GetParentRef()))
 }
 
-func (t *TenantActor) GetActorRef() actor.Ref {
+func (t *TenantActor) GetActorRef() ruleEngine.Ref {
 	return t.Ctx
 }
 
-func (t *TenantActor) Init(ctx actor.Ctx) error {
+func (t *TenantActor) Init(ctx ruleEngine.Ctx) error {
 	t.Ctx = ctx
 	server.Log.Debugf("Starting tenant actor:%s", t.tenantId)
-	t.initRuleChains()
-	return nil
+	return t.initRuleChains()
 }
 
 func (t *TenantActor) Process(msg protocol.ActorMsg) error {
 	if t.cantFindTenant {
 		server.Log.Debugf("Processing missing Tenant msg")
 		if msg.GetMessageType() == protocol.QUEUE_TO_RULE_ENGINE_MSG {
-			qMsg := msg.(*actor.QueueToRuleEngineMsg)
+			qMsg := msg.(*ruleEngine.QueueToRuleEngineMsg)
 			qMsg.Message.GetCallBack().OnSuccess()
 		} else if msg.GetMessageType() == protocol.TRANSPORT_TO_DEVICE_ACTOR_MSG {
-			tMsg := msg.(*actor.TransportToDeviceActorMsg)
+			tMsg := msg.(*ruleEngine.TransportToDeviceActorMsg)
 			tMsg.Message.GetCallBack().OnSuccess()
 		}
 		return nil
 	}
 	switch msg.GetMessageType() {
 	case protocol.QUEUE_TO_RULE_ENGINE_MSG:
-		return t.onQueueToRuleEngineMsg(msg.(*actor.QueueToRuleEngineMsg))
+		return t.onQueueToRuleEngineMsg(msg.(*ruleEngine.QueueToRuleEngineMsg))
 	case protocol.RULE_CHAIN_TO_RULE_CHAIN_MSG:
-		return t.onRuleChainToRuleChainMsg(msg.(*actor.RuleChainToRuleChainMsg))
+		return t.onRuleChainToRuleChainMsg(msg.(*ruleEngine.RuleChainToRuleChainMsg))
 	case protocol.TRANSPORT_TO_DEVICE_ACTOR_MSG:
 		//TODO:实现到设备的消息处理
+	case protocol.COMPONENT_LIFE_CYCLE_MSG:
+		return t.onComponentLifecycleMsg(msg.(*ruleEngine.ComponentLifecycleMsg))
+	default:
+		return errors.New("未知的消息类型")
+	}
+	return nil
+}
+
+func (t *TenantActor) onComponentLifecycleMsg(msg *ruleEngine.ComponentLifecycleMsg) error {
+	target := t.getEntityActorRef(msg.EntityId)
+	if target != nil {
+		if msg.EntityId.GetEntityType() == entities.RULE_CHAIN {
+			ruleChain, err := t.ruleChainService.FindRuleChainById(t.tenantId, msg.EntityId.GetId())
+			if err != nil {
+				return err
+			}
+			if ruleChain != nil {
+				if ruleChain.IsRoot {
+					t.rootChain = ruleChain
+					t.rootChainActor = target
+				}
+			}
+		}
+		target.TellWithHighPriority(msg)
+	} else {
+		server.Log.Debugln("Invalid component lifecycle msg")
+	}
+	return nil
+}
+
+func (t *TenantActor) getEntityActorRef(id entities.EntityId) ruleEngine.Ref {
+	if id.GetEntityType() == entities.RULE_CHAIN {
+		ruleChain, err := t.ruleChainService.FindRuleChainById(t.tenantId, id.GetId())
+		if err != nil {
+			return nil
+		}
+		ref, err := t.getOrCreateActor(id.GetId(), ruleChain)
+		if err != nil {
+			return nil
+		}
+		return ref
 	}
 	return nil
 }
 
-// TODO:基于services查找rule chain对象
-func (t *TenantActor) onRuleChainToRuleChainMsg(msg *actor.RuleChainToRuleChainMsg) error {
+func (t *TenantActor) onRuleChainToRuleChainMsg(msg *ruleEngine.RuleChainToRuleChainMsg) error {
 	ruleChainId := msg.Message.RuleChanId
-	ref, err := t.getOrCreateActor(ruleChainId, ruleChains[ruleChainId])
+	ruleChain, err := t.SystemCtx.RuleChainService.FindRuleChainById(t.tenantId, ruleChainId)
+	if err != nil {
+		return err
+	}
+	ref, err := t.getOrCreateActor(ruleChainId, ruleChain)
 	if err != nil {
 		return err
 	}
@@ -112,7 +144,7 @@ func (t *TenantActor) onRuleChainToRuleChainMsg(msg *actor.RuleChainToRuleChainM
 	return nil
 }
 
-func (t *TenantActor) onQueueToRuleEngineMsg(msg *actor.QueueToRuleEngineMsg) error {
+func (t *TenantActor) onQueueToRuleEngineMsg(msg *ruleEngine.QueueToRuleEngineMsg) error {
 	actorMsg := msg.Message
 	if actorMsg.RuleChanId == "" {
 		if t.rootChainActor != nil {
@@ -132,21 +164,21 @@ func (t *TenantActor) Destroy() error {
 	return nil
 }
 
-func (t *TenantActor) OnProcessFailure(err error) *actor.ProcessFailureStrategy {
+func (t *TenantActor) OnProcessFailure(err error) *ruleEngine.ProcessFailureStrategy {
 	if err != nil {
-		return actor.Stop()
+		return ruleEngine.Stop()
 	} else {
-		return actor.Resume()
+		return ruleEngine.Resume()
 	}
 }
 
 // TenantActorCreator 租户actor creator
 type TenantActorCreator struct {
-	actor.ContextBasedCreator
+	ruleEngine.ContextBasedCreator
 	tenantId string
 }
 
-func NewTenantActorCreator(sysCtx *actor.SystemContext, tenantId string) *TenantActorCreator {
+func NewTenantActorCreator(sysCtx *ruleEngine.SystemContext, tenantId string) *TenantActorCreator {
 	t := new(TenantActorCreator)
 	t.SystemCtx = sysCtx
 	t.tenantId = tenantId
@@ -157,7 +189,7 @@ func (t *TenantActorCreator) CreateActorId() string {
 	return t.tenantId
 }
 
-func (t *TenantActorCreator) CreateActor() actor.Actor {
+func (t *TenantActorCreator) CreateActor() ruleEngine.Actor {
 	ins := new(TenantActor)
 	ins.tenantId = t.tenantId
 	ins.SystemCtx = t.SystemCtx

pkg/entities/entities.go

@@ -16,11 +16,11 @@ const (
 )
 
 type RuleNodeId struct {
-	id string
+	Id string
 }
 
 func (r *RuleNodeId) GetId() string {
-	return r.id
+	return r.Id
 }
 
 func (r *RuleNodeId) GetEntityType() EntityType {
@@ -28,11 +28,11 @@ func (r *RuleNodeId) GetEntityType() EntityType {
 }
 
 type RuleChainId struct {
-	id string
+	Id string
 }
 
 func (r *RuleChainId) GetId() string {
-	return r.id
+	return r.Id
 }
 
 func (r *RuleChainId) GetEntityType() EntityType {

pkg/protocol/actor_msg.go

@@ -15,4 +15,6 @@ const (
 	APP_INIT_MSG                     MsgType = "APP_INIT_MSG"
 	TRANSPORT_TO_DEVICE_ACTOR_MSG    MsgType = "TRANSPORT_TO_DEVICE_ACTOR_MSG"
 	RULE_CHAIN_TO_RULE_CHAIN_MSG     MsgType = "RULE_CHAIN_TO_RULE_CHAIN_MSG"
+	RULE_TO_SELF_ERROR_MSG           MsgType = "RULE_TO_SELF_ERROR_MSG"
+	COMPONENT_LIFE_CYCLE_MSG         MsgType = "COMPONENT_LIFE_CYCLE_MSG"
 )

pkg/protocol/message.go

@@ -3,6 +3,7 @@ package protocol
 import (
 	"bytes"
 	"encoding/gob"
+	"sync/atomic"
 	"time"
 )
 
@@ -12,16 +13,21 @@ type MessageSerializer interface {
 }
 
 type Message struct {
-	QueueName  string
-	Id         string
-	Ts         *time.Time
-	Type       string
-	Data       string
-	RuleChanId string
-	RuleNodeId string
-	Callback   IMessageCallBack
-	MetaData   map[string]interface{}
-	Originator string
+	QueueName   string
+	Id          string
+	Ts          *time.Time
+	Type        string
+	Data        string
+	RuleChanId  string
+	RuleNodeId  string
+	Callback    IMessageCallBack
+	MetaData    map[string]interface{}
+	Originator  string
+	execCounter int32
+}
+
+func (a *Message) GetAndIncrementRuleNodeCounter() int32 {
+	return atomic.AddInt32(&a.execCounter, 1)
 }
 
 func (a *Message) Encode() ([]byte, error) {

pkg/protocol/schema.go

@@ -5,6 +5,20 @@ import "fmt"
 // RelationType 默认节点关系
 type RelationType string
 
+type RelationTypes []RelationType
+
+func (a RelationType) String() string {
+	return string(a)
+}
+
+func (a RelationTypes) ToStrArray() []string {
+	var str []string
+	for _, item := range a {
+		str = append(str, item.String())
+	}
+	return str
+}
+
 const (
 	Success RelationType = "Success" // 成功
 	Failure RelationType = "Failure" // 失败
@@ -27,13 +41,13 @@ const (
 
 // RuleNodeInfo rule node info for output
 type RuleNodeInfo struct {
-	ruleNodeId    string
-	ruleChainName string
-	ruleNodeName  string
+	RuleNodeId    string
+	RuleChainName string
+	RuleNodeName  string
 }
 
 func (r *RuleNodeInfo) String() string {
-	return fmt.Sprintf("[RuleChain:%s|RuleNode:%s|RuleNodeId:%s]", r.ruleChainName, r.ruleNodeName, r.ruleNodeId)
+	return fmt.Sprintf("[RuleChain:%s|RuleNode:%s|RuleNodeId:%s]", r.RuleChainName, r.RuleNodeName, r.RuleNodeId)
 }
 
 type RuleNodeRelation struct {

pkg/queue/cluster_service.go

@@ -1,5 +0,0 @@
-package queue
-
-type ClusterService interface {
-	PushMessageToRuleEngine(info *TopicPartitionInfo, msgId string, msg []byte, callback Callback)
-}

pkg/actor/actor.go → pkg/ruleEngine/actor.go

@@ -1,4 +1,4 @@
-package actor
+package ruleEngine
 
 import "sparrow/pkg/protocol"
 

pkg/actor/actor_ref.go → pkg/ruleEngine/actor_ref.go

@@ -1,4 +1,4 @@
-package actor
+package ruleEngine
 
 import "sparrow/pkg/protocol"
 

pkg/actor/actor_system.go → pkg/ruleEngine/actor_system.go

@@ -1,20 +1,21 @@
-package actor
+package ruleEngine
 
 import (
 	"errors"
 	"fmt"
 	"github.com/gogf/gf/os/grpool"
 	"sparrow/pkg/protocol"
-	"sparrow/pkg/queue"
 	"sparrow/pkg/server"
 	"sync"
 )
 
 // SystemContext actor system context, with some func
 type SystemContext struct {
-	ActorSystem    System
-	AppActor       Ref
-	ClusterService queue.ClusterService
+	ActorSystem      System
+	AppActor         Ref
+	ClusterService   ClusterService
+	RuleChainService RuleChainService
+	TenantService    TenantService
 }
 
 func NewSystemContext(sys System) *SystemContext {
@@ -103,7 +104,11 @@ func (d *DefaultActorSystem) DestroyDispatcher(name string) error {
 }
 
 func (d *DefaultActorSystem) GetActor(actorId string) Ref {
-	return d.actors[actorId]
+	if ref, ok := d.actors[actorId]; !ok {
+		return nil
+	} else {
+		return ref
+	}
 }
 
 func (d *DefaultActorSystem) CreateRootActor(dispatcherName string, creator Creator) (Ref, error) {
@@ -187,7 +192,7 @@ func (d *DefaultActorSystem) creator(name string, creator Creator, parentId stri
 			}
 		}
 		mailBox := NewMailBox(d, actorId, parentActor, actor, dispatcher)
-		d.actors[actorId] = actorMailBox
+		d.actors[actorId] = mailBox
 		if err := mailBox.Init(); err != nil {
 			server.Log.Error(err)
 			return nil, err

pkg/ruleEngine/cluster_service.go

@@ -0,0 +1,7 @@
+package ruleEngine
+
+import "sparrow/pkg/queue"
+
+type ClusterService interface {
+	PushMessageToRuleEngine(info *queue.TopicPartitionInfo, msgId string, msg []byte, callback queue.Callback)
+}

pkg/ruleEngine/context.go

@@ -1,7 +1,6 @@
 package ruleEngine
 
 import (
-	"sparrow/pkg/actor"
 	"sparrow/pkg/protocol"
 	"time"
 )
@@ -22,10 +21,10 @@ type Context interface {
 // DefaultContext 默认的上下文
 type DefaultContext struct {
 	nodeCtx *RuleNodeCtx
-	mainCtx *actor.SystemContext
+	mainCtx *SystemContext
 }
 
-func NewDefaultContext(nodeCtx *RuleNodeCtx, mainCtx *actor.SystemContext) *DefaultContext {
+func NewDefaultContext(nodeCtx *RuleNodeCtx, mainCtx *SystemContext) *DefaultContext {
 	return &DefaultContext{nodeCtx: nodeCtx, mainCtx: mainCtx}
 }
 
@@ -43,7 +42,7 @@ func (d *DefaultContext) tellNext(msg *protocol.Message, relationTypes []protoco
 	}
 	msg.GetCallBack().OnProcessingEnd(d.nodeCtx.Self.RuleNodeId)
 	d.nodeCtx.ChainActor.Tell(
-		&actor.RuleNodeToRuleChanTellNextMsg{
+		&RuleNodeToRuleChanTellNextMsg{
 			RuleNodeId:     d.nodeCtx.Self.RuleNodeId,
 			RelationTypes:  relationTypes,
 			Message:        msg,
@@ -54,7 +53,7 @@ func (d *DefaultContext) tellNext(msg *protocol.Message, relationTypes []protoco
 func (d *DefaultContext) TellSelf(msg *protocol.Message, duration time.Duration) {
 	if duration > 0 {
 		time.AfterFunc(duration, func() {
-			d.nodeCtx.SelfActor.Tell(&actor.RuleToSelfMsg{Message: msg})
+			d.nodeCtx.SelfActor.Tell(&RuleToSelfMsg{Message: msg})
 		})
 	}
 }
@@ -63,7 +62,7 @@ func (d *DefaultContext) TellError(msg *protocol.Message, err error) {
 	if d.nodeCtx.Self.IsDebug {
 		// TODO: 处理调试
 	}
-	d.nodeCtx.ChainActor.Tell(&actor.RuleNodeToRuleChanTellNextMsg{
+	d.nodeCtx.ChainActor.Tell(&RuleNodeToRuleChanTellNextMsg{
 		RuleNodeId:     d.nodeCtx.Self.RuleNodeId,
 		RelationTypes:  []protocol.RelationType{protocol.Failure},
 		Message:        msg,

pkg/actor/creator.go → pkg/ruleEngine/creator.go

@@ -1,4 +1,4 @@
-package actor
+package ruleEngine
 
 // Creator interface
 type Creator interface {

pkg/actor/dispatcher.go → pkg/ruleEngine/dispatcher.go

@@ -1,4 +1,4 @@
-package actor
+package ruleEngine
 
 import (
 	"github.com/gogf/gf/os/grpool"

pkg/actor/mailbox.go → pkg/ruleEngine/mailbox.go

@@ -1,7 +1,6 @@
-package actor
+package ruleEngine
 
 import (
-	"fmt"
 	"github.com/gogf/gf/container/gqueue"
 	"sparrow/pkg/protocol"
 	"sparrow/pkg/server"
@@ -118,7 +117,6 @@ func (m *MailBox) processMailbox() {
 		if getQueue == nil {
 			break
 		}
-		fmt.Printf("get queue")
 		msg = getQueue.Pop().(protocol.ActorMsg)
 		if msg != nil {
 			server.Log.Debugf("Going to process message:%s, %v", m.id, msg)

pkg/actor/msg.go → pkg/ruleEngine/msg.go

@@ -1,14 +1,14 @@
-package actor
+package ruleEngine
 
 import (
+	"sparrow/pkg/entities"
 	"sparrow/pkg/protocol"
-	"sparrow/pkg/ruleEngine"
 )
 
 // RuleChainToRuleNodeMsg 规则链到规则节点的消息定义
 type RuleChainToRuleNodeMsg struct {
 	Message          *protocol.Message
-	Ctx              ruleEngine.Context
+	Ctx              Context
 	FromRelationType string
 }
 
@@ -51,7 +51,7 @@ func (q *QueueToRuleEngineMsg) GetMessageType() protocol.MsgType {
 
 type RuleNodeToRuleChanTellNextMsg struct {
 	RuleNodeId     string
-	RelationTypes  []protocol.RelationType
+	RelationTypes  protocol.RelationTypes
 	Message        *protocol.Message
 	FailureMessage error
 }
@@ -68,6 +68,15 @@ func (r *RuleToSelfMsg) GetMessageType() protocol.MsgType {
 	return protocol.RULE_TO_SELF_MSG
 }
 
+type RuleToSelfErrorMsg struct {
+	Message *protocol.Message
+	Err     error
+}
+
+func (r *RuleToSelfErrorMsg) GetMessageType() protocol.MsgType {
+	return protocol.RULE_TO_SELF_ERROR_MSG
+}
+
 // AppInitMsg app 初始化消息
 type AppInitMsg struct {
 }
@@ -75,3 +84,41 @@ type AppInitMsg struct {
 func (a *AppInitMsg) GetMessageType() protocol.MsgType {
 	return protocol.APP_INIT_MSG
 }
+
+type ComponentLifecycleMsg struct {
+	TenantId  string
+	EntityId  entities.EntityId
+	EventType ComponentLifecycleEvent
+}
+
+func (c *ComponentLifecycleMsg) GetRuleChainId() string {
+	if c.EntityId.GetEntityType() == entities.RULE_CHAIN {
+		return c.EntityId.GetId()
+	} else {
+		return ""
+	}
+}
+
+func (c *ComponentLifecycleMsg) GetMessageType() protocol.MsgType {
+	return protocol.COMPONENT_LIFE_CYCLE_MSG
+}
+
+// ComponentLifecycleEvent 组件节点生命周期
+type ComponentLifecycleEvent string
+
+const (
+	CREATED   ComponentLifecycleEvent = "CREATED"
+	STARTED   ComponentLifecycleEvent = "STARTED"
+	ACTIVATED ComponentLifecycleEvent = "ACTIVATED"
+	UPDATED   ComponentLifecycleEvent = "UPDATED"
+	STOPPED   ComponentLifecycleEvent = "STOPPED"
+	DELETED   ComponentLifecycleEvent = "DELETED"
+)
+
+// ComponentLifecycleState 生命周期状态
+type ComponentLifecycleState int
+
+const (
+	ACTIVE ComponentLifecycleState = iota
+	SUSPENDED
+)

pkg/ruleEngine/node.go

@@ -1,7 +1,6 @@
 package ruleEngine
 
 import (
-	"sparrow/pkg/actor"
 	"sparrow/pkg/entities"
 	"sparrow/pkg/protocol"
 )
@@ -14,8 +13,8 @@ type Node interface {
 // RuleNodeCtx 节点上下文
 type RuleNodeCtx struct {
 	TenantId   string    // 租户Id
-	ChainActor actor.Ref // 规则链 actor
-	SelfActor  actor.Ref // 当前节点的 actor
+	ChainActor Ref       // 规则链 actor
+	SelfActor  Ref       // 当前节点的 actor
 	Self       *RuleNode // 当前节点
 }
 

pkg/ruleEngine/nodes/msg_type_filter_node.go

@@ -33,13 +33,21 @@ func (m *MsgTypeFilterNode) Init(ctx ruleEngine.Context, config string) error {
 
 func (m *MsgTypeFilterNode) OnMessage(ctx ruleEngine.Context, message *protocol.Message) error {
 	var relation protocol.RelationType
+	var found bool
 	for _, msgType := range m.config.MessageTypes {
 		if message.Type == msgType {
-			relation = protocol.True
+			found = true
+			break
 		} else {
-			relation = protocol.False
+			found = false
+			break
 		}
 	}
+	if found {
+		relation = protocol.True
+	} else {
+		relation = protocol.False
+	}
 	ctx.TellNext(message, relation)
 	return nil
 }

pkg/ruleEngine/nodes/reg_types.go

@@ -0,0 +1,44 @@
+package nodes
+
+import (
+	"reflect"
+	"sparrow/pkg/ruleEngine"
+)
+
+var registerTypeMap = make(map[string]*item)
+
+type item struct {
+	T reflect.Type
+	V reflect.Value
+}
+
+func init() {
+	registerType((*MsgTypeFilterNode)(nil))
+	registerType((*MsgTypeSwitchNode)(nil))
+}
+
+func registerType(elem interface{}) {
+	v := reflect.ValueOf(elem)
+	t := reflect.TypeOf(elem)
+	registerTypeMap[t.Elem().Name()] = &item{
+		T: t.Elem(),
+		V: v,
+	}
+}
+
+func CreateNodeByConfig(name string, ctx ruleEngine.Context, config string) (interface{}, bool) {
+	item, ok := registerTypeMap[name]
+	if !ok {
+		return nil, false
+	}
+	var ptr reflect.Value
+	ptr = reflect.New(item.T)
+	temp := ptr
+	initMethod := temp.MethodByName("Init")
+	args := []reflect.Value{
+		reflect.ValueOf(ctx),
+		reflect.ValueOf(config),
+	}
+	initMethod.Call(args)
+	return ptr.Interface(), true
+}

pkg/ruleEngine/nodes/reg_types_test.go

@@ -0,0 +1,21 @@
+package nodes
+
+import (
+	"encoding/json"
+	"sparrow/pkg/ruleEngine"
+	"testing"
+)
+
+func TestCreateNodeByType(t *testing.T) {
+	config, err := json.Marshal(&MsgTypeFilterNodeConfig{MessageTypes: []string{"POST_ATTRIBUTES_REQUEST"}})
+	if err != nil {
+		t.Error(err)
+	}
+	s, ok := CreateNodeByConfig("MsgTypeFilterNode", &ruleEngine.DefaultContext{}, string(config))
+	if !ok {
+		t.Fatalf("not found")
+	}
+	node := s.(ruleEngine.Node)
+
+	t.Log(node)
+}

pkg/actor/procee_failure_strategy.go → pkg/ruleEngine/procee_failure_strategy.go

@@ -1,4 +1,4 @@
-package actor
+package ruleEngine
 
 type ProcessFailureStrategy struct {
 	_stop bool

pkg/ruleEngine/rule_chain.go

@@ -6,7 +6,6 @@ const (
 	ruleChan EntityType = iota
 )
 
-
 type RuleChain struct {
 	TenantId    string
 	Name        string
@@ -27,8 +26,17 @@ type RuleNode struct {
 }
 
 type Relation struct {
-	From           string
-	To             string
-	Type           string
-	AdditionalInfo string
+	From              string
+	To                string
+	Type              string
+	AdditionalInfo    string
+	RelationTypeGroup RelationTypeGroup
 }
+
+type RelationTypeGroup int
+
+const (
+	COMMON RelationTypeGroup = iota
+	RULE_CHAIN
+	RULE_NODE
+)

pkg/ruleEngine/rule_chain_service.go

@@ -0,0 +1,99 @@
+package ruleEngine
+
+type RuleChainService interface {
+	// 根据id查询规则链数据
+	FindRuleChainById(tenantId, ruleChainId string) (*RuleChain, error)
+	// 根据id查询规则节点数据
+	FindRuleNodeById(tenantId, ruleNodeId string) (*RuleNode, error)
+	// 查询规则链的节点列表
+	GetRuleChainNodes(tenantId, ruleChainId string) ([]*RuleNode, error)
+	// 查询租户的全部规则链
+	FindRuleChains(tenantId string) ([]*RuleChain, error)
+	// 查询节点的连接关系列表
+	GetRuleNodeRelations(tenantId, nodeId string) ([]*Relation, error)
+}
+
+type TestRuleChainService struct {
+}
+
+func (t *TestRuleChainService) GetRuleNodeRelations(tenantId, nodeId string) ([]*Relation, error) {
+	return []*Relation{
+		{
+			From:              "node1",
+			To:                "node2",
+			Type:              "True",
+			RelationTypeGroup: COMMON,
+		},
+	}, nil
+}
+
+func (t *TestRuleChainService) FindRuleChainById(tenantId, ruleChainId string) (*RuleChain, error) {
+	return &RuleChain{
+		TenantId:    "tenant_1",
+		Name:        "test rule chain 1",
+		FirstNodeId: "node1",
+		IsDebug:     false,
+		IsRoot:      true,
+		Config:      "",
+		ChainId:     "chain id 1",
+	}, nil
+}
+
+func (t *TestRuleChainService) FindRuleNodeById(tenantId, ruleNodeId string) (*RuleNode, error) {
+	switch ruleNodeId {
+	case "node1":
+		return &RuleNode{
+			RuleChainId: "chain id 1",
+			Type:        "MsgTypeFilterNode",
+			Name:        "filternode1",
+			IsDebug:     false,
+			Config:      "",
+			RuleNodeId:  "node1",
+		}, nil
+	case "node2":
+		return &RuleNode{
+			RuleChainId: "chain id 2",
+			Type:        "MsgTypeFilterNode",
+			Name:        "filternode2",
+			IsDebug:     false,
+			Config:      "",
+			RuleNodeId:  "node2",
+		}, nil
+	}
+	return nil, nil
+}
+
+func (t *TestRuleChainService) GetRuleChainNodes(tenantId, ruleChainId string) ([]*RuleNode, error) {
+	return []*RuleNode{
+		{
+			RuleChainId: "chain id 1",
+			Type:        "MsgTypeFilterNode",
+			Name:        "filternode",
+			IsDebug:     false,
+			Config:      "",
+			RuleNodeId:  "node1",
+		},
+		{
+			RuleChainId: "chain id 2",
+			Type:        "MsgTypeFilterNode",
+			Name:        "filternode",
+			IsDebug:     false,
+			Config:      "",
+			RuleNodeId:  "node2",
+		},
+	}, nil
+}
+
+func (t *TestRuleChainService) FindRuleChains(tenantId string) ([]*RuleChain, error) {
+	return []*RuleChain{
+		{
+			TenantId:    "tenant_1",
+			Name:        "test rule chain 1",
+			FirstNodeId: "node1",
+			IsDebug:     false,
+			IsRoot:      true,
+			Config:      "",
+			ChainId:     "chain id 1",
+		},
+	}, nil
+}

pkg/queue/service_type.go → pkg/ruleEngine/service_type.go

@@ -1,4 +1,4 @@
-package queue
+package ruleEngine
 
 const (
 	SP_CORE     = "sp_core"

pkg/ruleEngine/tenant_service.go

@@ -0,0 +1,28 @@
+package ruleEngine
+
+type TenantService interface {
+	FindTenants() ([]*Tenant, error)
+	GetTenant(tId string) (*Tenant, error)
+}
+
+// Tenant 租户
+type Tenant struct {
+	Id   string
+	Name string
+}
+
+type TestTenantService struct {
+}
+
+func (t *TestTenantService) FindTenants() ([]*Tenant, error) {
+	return []*Tenant{
+		{
+			Id:   "tenant_1",
+			Name: "测试租户",
+		},
+	}, nil
+}
+
+func (t *TestTenantService) GetTenant(tId string) (*Tenant, error) {
+	panic("implement me")
+}

services/controller/controller.go

@@ -1,11 +1,14 @@
 package main
 
 import (
+	"sparrow/pkg/actors"
 	"sparrow/pkg/mongo"
 	"sparrow/pkg/queue"
 	"sparrow/pkg/rpcs"
 	"sparrow/pkg/rule"
+	"sparrow/pkg/ruleEngine"
 	"sparrow/pkg/server"
+	"time"
 )
 
 const (
@@ -138,3 +141,36 @@ func getAccessRPCHost(deviceid uint64) (string, error) {
 
 	return reply.AccessRPCHost, nil
 }
+
+type ActorSystem struct {
+	rootActor ruleEngine.Ref
+}
+
+func initActorSystem() (*ActorSystem, error) {
+	actorContext := new(ruleEngine.SystemContext)
+	system := ruleEngine.NewDefaultActorSystem(&ruleEngine.DefaultActorSystemConfig{
+		SchedulerPoolSize:            5,
+		AppDispatcherPoolSize:        4,
+		TenantDispatcherPoolSize:     4,
+		RuleEngineDispatcherPoolSize: 4,
+	})
+	system.CreateDispatcher(ruleEngine.APP_DISPATCHER_NAME, ruleEngine.NewPoolDispatcher(5))
+	system.CreateDispatcher(ruleEngine.TENANT_DISPATCHER_NAME, ruleEngine.NewPoolDispatcher(4))
+	system.CreateDispatcher(ruleEngine.RULE_DISPATCHER_NAME, ruleEngine.NewPoolDispatcher(4))
+	actorContext.ActorSystem = system
+	// init services
+	tenantService := &ruleEngine.TestTenantService{}
+	rulechainService := &ruleEngine.TestRuleChainService{}
+	actorContext.TenantService = tenantService
+	actorContext.RuleChainService = rulechainService
+	appActor, err := system.CreateRootActor(ruleEngine.APP_DISPATCHER_NAME,
+		actors.NewAppActorCreator(actorContext))
+	if err != nil {
+		return nil, err
+	}
+	actorContext.AppActor = appActor
+	server.Log.Debugln("actor system initialized")
+	time.Sleep(time.Second * 1)
+	appActor.Tell(&ruleEngine.AppInitMsg{})
+	return &ActorSystem{rootActor: appActor}, nil
+}

services/controller/controller_test.go

@@ -0,0 +1,36 @@
+package main
+
+import (
+	"sparrow/pkg/protocol"
+	"sparrow/pkg/ruleEngine"
+	"sparrow/pkg/server"
+	"testing"
+	"time"
+)
+
+func TestInit(t *testing.T) {
+	server.InitLog("test", "debug")
+	as, err := initActorSystem()
+	if err != nil {
+		t.Error(err)
+	}
+	time.Sleep(5 * time.Second)
+	as.rootActor.Tell(&ruleEngine.QueueToRuleEngineMsg{
+		TenantId: "tenant_1",
+		Message: &protocol.Message{
+			QueueName:  "",
+			Id:         "",
+			Ts:         nil,
+			Type:       "POST_ATTRIBUTES_REQUEST",
+			Data:       "",
+			RuleChanId: "",
+			RuleNodeId: "",
+			Callback:   nil,
+			MetaData:   nil,
+			Originator: "",
+		},
+		RelationTypes:  nil,
+		FailureMessage: nil,
+	})
+	select {}
+}