dev: enhance more modules

component/aibase/aibase_runtime.go

@@ -46,6 +46,7 @@ func InitAlgorithmRuntime(re typex.RuleX) *AlgorithmRuntime {
 	__DefaultAIRuntime = new(AlgorithmRuntime)
 	__DefaultAIRuntime.RuleEngine = re
 	__DefaultAIRuntime.XAlgorithms = make(map[string]XAlgorithm)
+	__DefaultAIRuntime.locker = sync.Mutex{}
 	// Yolo8
 	// err1 := LoadAlgorithm(NewYolo8ObjectDetectionCpu(), map[string]interface{}{})
 	// if err1 != nil {

component/appstack/appstack_runtime.go

@@ -18,6 +18,7 @@ package appstack
 import (
 	"context"
 	"fmt"
+	"sync"
 	"time"
 
 	lua "github.com/hootrhino/gopher-lua"
@@ -31,6 +32,7 @@ var __DefaultAppStackRuntime *AppStackRuntime
 func InitAppStack(re typex.RuleX) *AppStackRuntime {
 	__DefaultAppStackRuntime = &AppStackRuntime{
 		RuleEngine:   re,
+		locker:       sync.Mutex{},
 		Applications: make(map[string]*Application),
 	}
 	return __DefaultAppStackRuntime
@@ -45,7 +47,8 @@ func AppRuntime() *AppStackRuntime {
 *
  */
 func LoadApp(app *Application, luaSource string) error {
-
+	__DefaultAppStackRuntime.locker.Lock()
+	defer __DefaultAppStackRuntime.locker.Unlock()
 	// 重新读
 	app.VM().DoString(string(luaSource))
 	// 检查函数入口
@@ -73,6 +76,8 @@ func LoadApp(app *Application, luaSource string) error {
 *
  */
 func StartApp(uuid string) error {
+	__DefaultAppStackRuntime.locker.Lock()
+	defer __DefaultAppStackRuntime.locker.Unlock()
 	app, ok := __DefaultAppStackRuntime.Applications[uuid]
 	if !ok {
 		return fmt.Errorf("Application not exists:%s", uuid)
@@ -173,6 +178,8 @@ func StartApp(uuid string) error {
 *
  */
 func RemoveApp(uuid string) error {
+	__DefaultAppStackRuntime.locker.Lock()
+	defer __DefaultAppStackRuntime.locker.Unlock()
 	if app, ok := __DefaultAppStackRuntime.Applications[uuid]; ok {
 		app.Remove()
 		delete(__DefaultAppStackRuntime.Applications, uuid)
@@ -187,6 +194,8 @@ func RemoveApp(uuid string) error {
 *
  */
 func StopApp(uuid string) error {
+	__DefaultAppStackRuntime.locker.Lock()
+	defer __DefaultAppStackRuntime.locker.Unlock()
 	if app, ok := __DefaultAppStackRuntime.Applications[uuid]; ok {
 		app.Stop()
 	}
@@ -200,6 +209,8 @@ func StopApp(uuid string) error {
 *
  */
 func UpdateApp(app Application) error {
+	__DefaultAppStackRuntime.locker.Lock()
+	defer __DefaultAppStackRuntime.locker.Unlock()
 	if oldApp, ok := __DefaultAppStackRuntime.Applications[app.UUID]; ok {
 		oldApp.Name = app.Name
 		oldApp.AutoStart = app.AutoStart
@@ -237,6 +248,8 @@ func ListApp() []*Application {
 }
 
 func Stop() {
+	__DefaultAppStackRuntime.locker.Lock()
+	defer __DefaultAppStackRuntime.locker.Unlock()
 	for _, app := range __DefaultAppStackRuntime.Applications {
 		glogger.GLogger.Info("Stop App:", app.UUID)
 		app.Stop()

component/appstack/types.go

@@ -15,14 +15,19 @@
 
 package appstack
 
-import "github.com/hootrhino/rulex/typex"
+import (
+	"sync"
+
+	"github.com/hootrhino/rulex/typex"
+)
 
 /*
 *
 * 管理器
 *
  */
 type AppStackRuntime struct {
+	locker       sync.Mutex
 	RuleEngine   typex.RuleX
 	Applications map[string]*Application
 }

component/eventbus/eventbus.go

@@ -18,6 +18,7 @@ package eventbus
 import (
 	"context"
 	"fmt"
+	"sync"
 
 	"github.com/hootrhino/rulex/typex"
 	"github.com/hootrhino/rulex/utils"
@@ -41,9 +42,9 @@ func (E EventMessage) String() string {
 type Topic struct {
 	Topic       string
 	channel     chan EventMessage
-	Subscribers map[string]*Subscriber
 	ctx         context.Context
 	cancel      context.CancelFunc
+	Subscribers map[string]*Subscriber
 }
 type Subscriber struct {
 	id       string
@@ -53,12 +54,14 @@ type EventBus struct {
 	// Topic, chan EventMessage
 	// 给每个订阅者分配一个Channel，实现消息订阅
 	// Topic一样的会挂在同一个树上
+	locker sync.Mutex
 	Topics map[string]*Topic // 订阅树: MAP<Topic>[]Subscribers
 }
 
 func InitEventBus() *EventBus {
 	__DefaultEventBus = &EventBus{
 		Topics: map[string]*Topic{},
+		locker: sync.Mutex{},
 	}
 	return __DefaultEventBus
 }
@@ -69,6 +72,8 @@ func InitEventBus() *EventBus {
 *
  */
 func Subscribe(topic string, subscribe *Subscriber) {
+	__DefaultEventBus.locker.Lock()
+	defer __DefaultEventBus.locker.Unlock()
 	NewUUID := utils.MakeUUID("SUB")
 	subscribe.id = NewUUID
 	var T *Topic
@@ -112,6 +117,8 @@ func Subscribe(topic string, subscribe *Subscriber) {
 *
  */
 func UnSubscribe(topic string, subscribe Subscriber) {
+	__DefaultEventBus.locker.Lock()
+	defer __DefaultEventBus.locker.Unlock()
 	T, Ok1 := __DefaultEventBus.Topics[topic]
 	if Ok1 {
 		if _, Ok2 := T.Subscribers[subscribe.id]; Ok2 {
@@ -143,6 +150,8 @@ func Publish(topic string, Msg EventMessage) {
 *
  */
 func Flush() {
+	__DefaultEventBus.locker.Lock()
+	defer __DefaultEventBus.locker.Unlock()
 	for _, T := range __DefaultEventBus.Topics {
 		for _, S := range T.Subscribers {
 			delete(T.Subscribers, S.id)