修改事件封装

修改世界状态更新事件处理方式
2023-12-25 10:57:54 +08:00 · 2023-12-25 10:57:54 +08:00 · e83feb89dd
parent bd947baa4e
commit e83feb89dd
3 changed files with 91 additions and 66 deletions
--- a/events.go
+++ b/events.go
@ -9,15 +9,16 @@ import (
 	"github.com/yohamta/donburi/features/events"
 )

-// 注意事件相关操作的最终执行者均为world协程
+// 事件订阅发布功能,底层使用donburi事件订阅发布
+// 注意:事件订阅处理中应避免再次发布事件

 // 事件相关定义
 type (
 	// 事件类型定义
 	EventType[T any] struct {
 		et     *events.EventType[T]
-		subscriberMap     map[string]events.Subscriber[T]
-		subscriberMapLock sync.Mutex //
+		mu     sync.Mutex // 锁
+		subMap map[string]events.Subscriber[T]
 	}
 	// 事件订阅者定义
 	Subscriber[T any] func(w World, event T)
@ -32,8 +33,8 @@ func EventsDebugEnable() {
 func NewEventType[T any]() *EventType[T] {
 	return &EventType[T]{
 		et:     events.NewEventType[T](),
-		subscriberMap:     make(map[string]events.Subscriber[T]),
-		subscriberMapLock: sync.Mutex{},
+		mu:     sync.Mutex{},
+		subMap: make(map[string]events.Subscriber[T], 0),
 	}
 }

@ -42,52 +43,47 @@ func processAllEvents(w World) {
 	events.ProcessAllEvents(w)
 }

+// 执行事件
+func (me *EventType[T]) ProcessEvents(wd World) {
+	me.mu.Lock()
+	defer me.mu.Unlock()
+	me.et.ProcessEvents(wd)
+}
+
 // 发布该类型的事件
 func (me *EventType[T]) Publish(wd World, event *T) {
-	wd.Execute(func() {
+	me.mu.Lock()
+	defer me.mu.Unlock()
 	me.et.Publish(wd, *event)
-	})
 }

-// 内部发布事件(同步发布，并直接触发)
-func (me *EventType[T]) internalPublish(wd World, event *T) {
-	me.et.Publish(wd, *event)
-	processAllEvents(wd)
-}
-
-///////////////////////////////////////////////////////////////////////////////////
-
 // 订阅该类型的事件
 func (me *EventType[T]) Subscribe(wd World, subscriber Subscriber[T]) {
-	me.subscriberMapLock.Lock()
-	defer me.subscriberMapLock.Unlock()
-	subscriberKey := subscriberKey[T](wd, subscriber)
-	if _, exist := me.subscriberMap[subscriberKey]; !exist {
-		fn := func(w donburi.World, event T) {
+	me.mu.Lock()
+	defer me.mu.Unlock()
+	sp := buildKey(wd, subscriber)
+	if _, ok := me.subMap[sp]; ok { // 已经订阅过
+		return
+	}
+	sub := func(w donburi.World, event T) {
 		subscriber(wd, event)
 	}
-		me.subscriberMap[subscriberKey] = fn
-		wd.Execute(func() {
-			me.et.Subscribe(wd, fn)
-		})
-	}
+	me.subMap[sp] = sub
+	me.et.Subscribe(wd, sub)
 }

 // 取消订阅该类型的事件
 func (me *EventType[T]) Unsubscribe(wd World, subscriber Subscriber[T]) {
-	me.subscriberMapLock.Lock()
-	defer me.subscriberMapLock.Unlock()
-	subscriberKey := subscriberKey[T](wd, subscriber)
-	if sub, ok := me.subscriberMap[subscriberKey]; ok {
-		wd.Execute(func() {
+	me.mu.Lock()
+	defer me.mu.Unlock()
+	sp := buildKey(wd, subscriber)
+	if sub, ok := me.subMap[sp]; ok { // 存在, 取消订阅
 		me.et.Unsubscribe(wd, sub)
-			delete(me.subscriberMap, subscriberKey)
-		})
 	}
 }

-func subscriberKey[T any](wd World, subscriber Subscriber[T]) string {
-	wdSubscriberPointer := reflect.ValueOf(subscriber).Pointer()
-	subscriberKey := fmt.Sprintf("%d-%d", wd.Id(), wdSubscriberPointer)
-	return subscriberKey
+// 生成订阅键
+func buildKey[T any](wd World, subscriber Subscriber[T]) string {
+	sp := fmt.Sprintf("%v_%v", wd.Id(), reflect.ValueOf(subscriber).Pointer())
+	return sp
 }
--- a/examples/rtss-event/main.go
+++ b/examples/rtss-event/main.go
@ -1,6 +1,7 @@
 package main

 import (
+	"context"
 	"fmt"
 	"time"

@ -43,8 +44,25 @@ var (
 func main() {
 	go world1()
 	go world2()
-	time.Sleep(4 * time.Second)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go func() {
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			default:
+			}
 			fireSwitchDcEventType.Publish(wd2, &FireSwitchDcEvent{Dc: false})
+			time.Sleep(10 * time.Millisecond)
+		}
+	}()
+	time.Sleep(2 * time.Second)
+
+	wd1.Close()
+	wd2.Close()
+
+	time.Sleep(5 * time.Second)
 }
 func world1() {
 	world := wd1
@ -52,6 +70,9 @@ func world1() {
 	//
 	switchSystem := NewSwitchSystem()
 	world.AddSystem(switchSystem)
+	ecs.WorldStateChangeEvent.Subscribe(wd1, func(w ecs.World, event ecs.WorldStateChange) {
+		fmt.Println("世界1状态变更", event)
+	})
 	//
 	fireSwitchDcEventType.Subscribe(world, switchSystem.WhenFireSwitchDcEvent)
 	//
@ -68,6 +89,10 @@ func world2() {
 	switchSystem := NewSwitchSystem()
 	world.AddSystem(switchSystem)
 	fireSwitchDcEventType.Subscribe(world, switchSystem.WhenFireSwitchDcEvent)
+
+	ecs.WorldStateChangeEvent.Subscribe(wd2, func(w ecs.World, event ecs.WorldStateChange) {
+		fmt.Println("世界2状态变更", event)
+	})
 	//
 	time.Sleep(3 * time.Second)
 	//
--- a/world.go
+++ b/world.go
@ -1,6 +1,7 @@
 package ecs

 import (
+	"context"
 	"fmt"
 	"log/slog"
 	"math"
@ -16,12 +17,11 @@ type WorldState int
 type WorldId = donburi.WorldId

 const (
-	WorldInit WorldState = iota
-	WorldRunning
-	WorldPause
-	WorldError
-	WorldClose
-	WorldClosed
+	WorldInit    WorldState = 0
+	WorldRunning WorldState = 1
+	WorldPause   WorldState = 2
+	WorldError   WorldState = 3
+	WorldClosed  WorldState = 4
 )

 type (
@ -38,7 +38,7 @@ type (
 		Resume()
 		// 关闭世界
 		Close()
-		// 设置时间运行倍速
+		// 设置运行倍速
 		SetSpeed(speed float64) error
 		// 添加系统
 		AddSystem(sys ...ISystem)
@ -72,6 +72,9 @@ type world struct {

 	// 待执行函数
 	toBeExecuteds chan HandleFunc
+
+	cancel context.CancelFunc
+	done   chan struct{} // 服务协程退出信号
 }

 // 新建一个组件类型
@ -109,11 +112,10 @@ func NewWorld(tick int) World {
 		speed:         1,
 		times:         1,
 		toBeExecuteds: make(chan HandleFunc, 32),
+		done:          make(chan struct{}),
 	}
 }
-func (w *world) Running() bool {
-	return w.state == WorldRunning
-}
+
 func (w *world) Tick() int {
 	return w.tick
 }
@ -147,10 +149,11 @@ func (w *world) updateState(state WorldState) {
 		old := w.state
 		slog.Debug("世界状态变更", "oldstate", old, "state", state)
 		w.state = state
-		WorldStateChangeEvent.internalPublish(w, &WorldStateChange{
+		WorldStateChangeEvent.Publish(w, &WorldStateChange{
 			OldState: old,
 			NewState: state,
 		})
+		WorldStateChangeEvent.ProcessEvents(w)
 	}
 }

@ -179,8 +182,11 @@ func (w *world) SetSpeed(speed float64) error {
 // 启动世界,世界逻辑开始执行且世界为运行状态
 func (w *world) StartUp() {
 	if w.state == WorldInit { // 避免重复运行
+		slog.Debug("启动世界", "id", w.Id())
+		ctx, cancle := context.WithCancel(context.Background())
+		go w.run(ctx)
+		w.cancel = cancle
 		w.updateState(WorldRunning)
-		go w.run()
 	}
 }

@ -188,7 +194,7 @@ func (w *world) StartUp() {
 func (w *world) Execute(fn HandleFunc) error {
 	if w.state == WorldError {
 		return fmt.Errorf("世界运行异常,无法执行请求")
-	} else if w.state != WorldRunning && w.state != WorldPause {
+	} else if w.state == WorldClosed {
 		return fmt.Errorf("世界已经关闭，无法执行请求")
 	}
 	w.toBeExecuteds <- fn
@ -197,13 +203,8 @@ func (w *world) Execute(fn HandleFunc) error {

 // 关闭世界
 func (w *world) Close() {
-	if w.state == WorldRunning || w.state == WorldPause {
-		w.Execute(func() {
-			w.updateState(WorldClose)
-		})
-	} else if w.state == WorldError {
-		w.updateState(WorldClosed)
-	}
+	w.cancel()
+	<-w.done
 }

 // 执行待处理方法
@ -222,7 +223,7 @@ func (w *world) executeTodos() {
 }

 // 世界循环
-func (w *world) run() {
+func (w *world) run(ctx context.Context) {
 	defer func() {
 		if err := recover(); err != nil {
 			w.exception(err)
@ -230,13 +231,15 @@ func (w *world) run() {
 			debug.PrintStack()
 		}
 	}()
+	defer close(w.done)
 	for range w.ticker.C {
-		// slog.Debug("世界运行")
-		if w.state == WorldClose {
-			// 世界正常关闭
+		select {
+		case <-ctx.Done():
 			w.close()
 			return
+		default:
 		}
+
 		// start := time.Now()
 		if w.state != WorldRunning { // 世界非运行状态
 			continue
@ -272,6 +275,7 @@ func (w *world) exception(err any) {

 // 世界正常关闭逻辑
 func (w *world) close() {
+	slog.Debug("关闭世界", "id", w.Id())
 	// 世界正常关闭
 	w.updateState(WorldClosed)
 	// 关闭定时器