123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165 |
- package actors
- import (
- "errors"
- "sparrow/pkg/actor"
- "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
- }
- func (t *TenantActor) initRuleChains() {
- for _, ruleChain := range ruleChains {
- server.Log.Debugf("Creating rule chain actor:%s", ruleChain.ChainId)
- actorRef, err := t.getOrCreateActor(ruleChain.ChainId, ruleChain)
- if err != nil {
- server.Log.Errorf("Creating rule chain actor:%s err:%s", ruleChain.ChainId, err.Error())
- continue
- }
- if ruleChain.IsRoot {
- t.rootChain = ruleChain
- t.rootChainActor = actorRef
- }
- server.Log.Debugf("Rule chain actor created:%s", ruleChain.ChainId)
- }
- }
- func (t *TenantActor) destroyRuleChains() {
- for _, ruleChain := range ruleChains {
- _ = t.Ctx.Stop(ruleChain.ChainId)
- }
- }
- func (t *TenantActor) getOrCreateActor(ruleChainId string, ruleChain *ruleEngine.RuleChain) (actor.Ref, error) {
- return t.Ctx.GetOrCreateChildActor(ruleChainId,
- actor.RULE_DISPATCHER_NAME,
- NewRuleChainCreator(t.SystemCtx, t.tenantId, ruleChain, t.Ctx.GetParentRef()))
- }
- func (t *TenantActor) GetActorRef() actor.Ref {
- return t.Ctx
- }
- func (t *TenantActor) Init(ctx actor.Ctx) error {
- t.Ctx = ctx
- server.Log.Debugf("Starting tenant actor:%s", t.tenantId)
- t.initRuleChains()
- return nil
- }
- 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.Message.GetCallBack().OnSuccess()
- } else if msg.GetMessageType() == protocol.TRANSPORT_TO_DEVICE_ACTOR_MSG {
- tMsg := msg.(*actor.TransportToDeviceActorMsg)
- tMsg.Message.GetCallBack().OnSuccess()
- }
- return nil
- }
- switch msg.GetMessageType() {
- case protocol.QUEUE_TO_RULE_ENGINE_MSG:
- return t.onQueueToRuleEngineMsg(msg.(*actor.QueueToRuleEngineMsg))
- case protocol.RULE_CHAIN_TO_RULE_CHAIN_MSG:
- return t.onRuleChainToRuleChainMsg(msg.(*actor.RuleChainToRuleChainMsg))
- case protocol.TRANSPORT_TO_DEVICE_ACTOR_MSG:
- //TODO:实现到设备的消息处理
- }
- return nil
- }
- // TODO:基于services查找rule chain对象
- func (t *TenantActor) onRuleChainToRuleChainMsg(msg *actor.RuleChainToRuleChainMsg) error {
- ruleChainId := msg.Message.RuleChanId
- ref, err := t.getOrCreateActor(ruleChainId, ruleChains[ruleChainId])
- if err != nil {
- return err
- }
- ref.Tell(msg)
- return nil
- }
- func (t *TenantActor) onQueueToRuleEngineMsg(msg *actor.QueueToRuleEngineMsg) error {
- actorMsg := msg.Message
- if actorMsg.RuleChanId == "" {
- if t.rootChainActor != nil {
- t.rootChainActor.Tell(msg)
- } else {
- actorMsg.GetCallBack().OnFailure(errors.New("no Root Rule Chain available"))
- server.Log.Errorf("no root chain:%s", t.tenantId)
- }
- } else {
- t.Ctx.TellActor(actorMsg.RuleChanId, msg)
- }
- actorMsg.GetCallBack().OnSuccess()
- return nil
- }
- func (t *TenantActor) Destroy() error {
- return nil
- }
- func (t *TenantActor) OnProcessFailure(err error) *actor.ProcessFailureStrategy {
- if err != nil {
- return actor.Stop()
- } else {
- return actor.Resume()
- }
- }
- // TenantActorCreator 租户actor creator
- type TenantActorCreator struct {
- actor.ContextBasedCreator
- tenantId string
- }
- func NewTenantActorCreator(sysCtx *actor.SystemContext, tenantId string) *TenantActorCreator {
- t := new(TenantActorCreator)
- t.SystemCtx = sysCtx
- t.tenantId = tenantId
- return t
- }
- func (t *TenantActorCreator) CreateActorId() string {
- return t.tenantId
- }
- func (t *TenantActorCreator) CreateActor() actor.Actor {
- ins := new(TenantActor)
- ins.tenantId = t.tenantId
- ins.SystemCtx = t.SystemCtx
- return ins
- }
|