sparrow/pkg/actor/actor_system.go

package actor

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
}

func NewSystemContext(sys System) *SystemContext {
	return &SystemContext{
		ActorSystem: sys,
	}
}

func (s *SystemContext) tell(msg protocol.ActorMsg) {
	s.AppActor.Tell(msg)
}

func (s *SystemContext) tellWithHighPriority(msg protocol.ActorMsg) {
	s.AppActor.TellWithHighPriority(msg)
}

// System actor system interface
type System interface {
	// 创建分发器
	CreateDispatcher(name string, dispatcher IDispatcher) error
	// 销毁分发器
	DestroyDispatcher(name string) error
	// 获取一个actor ref
	GetActor(actorId string) Ref
	// create root actor
	CreateRootActor(dispatcherName string, creator Creator) (Ref, error)
	// create child actor by parent actor
	CreateChildActor(dispatcherName string, creator Creator, parentId string) (Ref, error)
	// tell actor a message
	Tell(actorId string, msg protocol.ActorMsg) error
	// tell actor message with high priority
	TellWithHighPriority(actorId string, msg protocol.ActorMsg) error
	// stop actor by actor id
	StopActorById(actorId string) error
	// stop actor by actor ref
	StopActorByRef(ref Ref) error
	// broadcast message to children
	BroadcastToChildren(parentActorId string, msg protocol.ActorMsg) error
}

// DefaultActorSystem a default actor system implements System interface
type DefaultActorSystem struct {
	dispatchers  map[string]IDispatcher
	actors       map[string]*MailBox
	parentActors map[string][]string
	scheduler    *grpool.Pool
	config       *DefaultActorSystemConfig
	mu           sync.Mutex
}

// DefaultActorSystemConfig system config
type DefaultActorSystemConfig struct {
	SchedulerPoolSize            int // 系统调度执行池大小
	AppDispatcherPoolSize        int // 应用调度执行池大小
	TenantDispatcherPoolSize     int // 租户执行池大小
	RuleEngineDispatcherPoolSize int // 规则引擎执行池大小
}

func NewDefaultActorSystem(config *DefaultActorSystemConfig) *DefaultActorSystem {
	return &DefaultActorSystem{
		dispatchers:  make(map[string]IDispatcher),
		parentActors: make(map[string][]string),
		scheduler:    grpool.New(config.SchedulerPoolSize),
		config:       config,
		actors:       make(map[string]*MailBox),
	}
}

func (d *DefaultActorSystem) CreateDispatcher(name string, dispatcher IDispatcher) error {
	if _, ok := d.dispatchers[name]; ok {
		return errors.New(fmt.Sprintf("dispatcher name :%s is already registered!", name))
	}
	d.dispatchers[name] = dispatcher
	return nil
}

func (d *DefaultActorSystem) DestroyDispatcher(name string) error {
	if _, ok := d.dispatchers[name]; !ok {
		return errors.New(fmt.Sprintf("dispatcher %s is not registered!", name))
	}
	if err := d.dispatchers[name].Destroy(); err != nil {
		return err
	}
	delete(d.dispatchers, name)
	return nil
}

func (d *DefaultActorSystem) GetActor(actorId string) Ref {
	return d.actors[actorId]
}

func (d *DefaultActorSystem) CreateRootActor(dispatcherName string, creator Creator) (Ref, error) {
	return d.creator(dispatcherName, creator, "")
}

func (d *DefaultActorSystem) Tell(actorId string, msg protocol.ActorMsg) error {
	return d.tell(actorId, msg, false)
}

func (d *DefaultActorSystem) TellWithHighPriority(actorId string, msg protocol.ActorMsg) error {
	return d.tell(actorId, msg, true)
}

func (d *DefaultActorSystem) tell(actorId string, msg protocol.ActorMsg, isHighPriority bool) error {
	if mailBox, ok := d.actors[actorId]; ok {
		if isHighPriority {
			mailBox.TellWithHighPriority(msg)
		} else {
			mailBox.Tell(msg)
		}
	} else {
		return errors.New(fmt.Sprintf("actor with id %s is not registered!", actorId))
	}
	return nil
}

func (d *DefaultActorSystem) stop(actorId string) error {
	children := d.parentActors[actorId]
	if len(children) > 0 {
		for _, child := range children {
			_ = d.stop(child)
		}
	}
	if m, found := d.actors[actorId]; found {
		return m.destroy()
	}
	return nil
}

func (d *DefaultActorSystem) StopActorById(actorId string) error {
	return d.stop(actorId)
}

func (d *DefaultActorSystem) StopActorByRef(ref Ref) error {
	return d.stop(ref.GetActorId())
}

func (d *DefaultActorSystem) BroadcastToChildren(parentActorId string, msg protocol.ActorMsg) error {
	children := d.parentActors[parentActorId]
	for _, item := range children {
		if err := d.Tell(item, msg); err != nil {
			return err
		}
	}
	return nil
}
func (d *DefaultActorSystem) CreateChildActor(dispatcherName string, creator Creator, parentId string) (Ref, error) {
	return d.creator(dispatcherName, creator, parentId)
}
func (d *DefaultActorSystem) creator(name string, creator Creator, parentId string) (Ref, error) {
	d.mu.Lock()
	defer d.mu.Unlock()
	dispatcher := d.dispatchers[name]
	if dispatcher == nil {
		return nil, errors.New(fmt.Sprintf("dispatcher %s not found!", name))
	}
	actorId := creator.CreateActorId()
	var actorMailBox *MailBox
	actorMailBox = d.actors[actorId]
	if actorMailBox != nil {
		server.Log.Debugf("actor with id :%s is already registered!", actorId)
	} else {
		server.Log.Debugf("creating actor with id :%s", actorId)
		actor := creator.CreateActor()
		var parentActor Ref
		if parentId != "" {
			parentActor = d.GetActor(parentId)
			if parentActor == nil {
				return nil, errors.New(fmt.Sprintf("Parent actor with id :%s is not registered!", parentId))
			}
		}
		mailBox := NewMailBox(d, actorId, parentActor, actor, dispatcher)
		d.actors[actorId] = actorMailBox
		if err := mailBox.Init(); err != nil {
			server.Log.Error(err)
			return nil, err
		}
		actorMailBox = mailBox
		if parentActor != nil {
			d.parentActors[parentId] = append(d.parentActors[parentId], actorId)
		}
	}

	return actorMailBox, nil
}