feat(os/gevent): refactor event management and add new features

- Refactored the core logic of event management to improve code readability and maintainability
- Added support for event handling priorities, enabling flexible control over the event processing order
- Introduced a recovery function mechanism for event handling to manage exceptions during processing
- Optimized event publishing and subscription interfaces to simplify usage
- Added unit test cases to improve test coverage

Author: LanceAdd

os/gevent/gevent.go

@@ -12,26 +12,31 @@ import (
 	"sync"
 
 	"github.com/gogf/gf/v2/container/gmap"
+	"github.com/gogf/gf/v2/errors/gerror"
 	"github.com/gogf/gf/v2/util/gutil"
 )
 
 // Handler is a function type for handling events.
 // It receives the topic name and message data as parameters.
 type Handler func(topic string, message any)
 
+type RecoverFunc func(topic string, message any)
+
 // handlerInfo stores information about an event handler.
 type handlerInfo struct {
-	id       int64    // Unique identifier for the handler
-	topic    string   // Topic name
-	handler  Handler  // Handler function
-	priority Priority // Handler priority
+	id          int64       // Unique identifier for the handler
+	topic       string      // Topic name
+	handler     Handler     // Handler function
+	recoverFunc RecoverFunc // Recover function
+	priority    Priority    // Handler priority
 }
 
 // Event is the main struct for event management.
 type Event struct {
 	mu       sync.RWMutex    // Read-write mutex for concurrent safety
 	handlers *gmap.StrAnyMap // Map to store handlers by topic
 	counter  int64           // Counter for generating unique handler IDs
+	closed   bool            // Indicates whether the event manager is closed
 }
 
 // Subscriber represents a subscription to an event topic.
@@ -42,6 +47,8 @@ type Subscriber struct {
 	once  sync.Once // Ensures unsubscribe is called only once
 }
 
+var EventClosedError = gerror.New("event manager is closed")
+
 // Unsubscribe removes the subscription from the event manager.
 // It can be called multiple times safely, but only the first call will take effect.
 func (s *Subscriber) Unsubscribe() {
@@ -58,12 +65,12 @@ func New() *Event {
 	}
 }
 
-// Subscribe registers a handler function for a specific topic.
-// Handlers can be assigned a priority level, with higher priority handlers executed first.
-// Returns a Subscriber that can be used to unsubscribe the handler.
-func (e *Event) Subscribe(topic string, handler Handler, priority ...Priority) *Subscriber {
+func (e *Event) SubscribeWithRecover(topic string, handler Handler, recoverFunc RecoverFunc, priority ...Priority) (*Subscriber, error) {
 	e.mu.Lock()
 	defer e.mu.Unlock()
+	if e.closed {
+		return nil, EventClosedError
+	}
 	e.counter++
 	level := PriorityNormal
 	if len(priority) > 0 {
@@ -77,10 +84,11 @@ func (e *Event) Subscribe(topic string, handler Handler, priority ...Priority) *
 		return make([]*handlerInfo, 0, 8)
 	})
 	h := &handlerInfo{
-		id:       e.counter,
-		topic:    topic,
-		handler:  handler,
-		priority: level,
+		id:          e.counter,
+		topic:       topic,
+		handler:     handler,
+		priority:    level,
+		recoverFunc: recoverFunc,
 	}
 	handlers := infos.([]*handlerInfo)
 	handlers = append(handlers, h)
@@ -90,7 +98,14 @@ func (e *Event) Subscribe(topic string, handler Handler, priority ...Priority) *
 		topic: topic,
 		event: e,
 		once:  sync.Once{},
-	}
+	}, nil
+}
+
+// Subscribe registers a handler function for a specific topic.
+// Handlers can be assigned a priority level, with higher priority handlers executed first.
+// Returns a Subscriber that can be used to unsubscribe the handler.
+func (e *Event) Subscribe(topic string, handler Handler, priority ...Priority) (*Subscriber, error) {
+	return e.SubscribeWithRecover(topic, handler, nil, priority...)
 }
 
 // UnSubscribe removes a handler from a topic by its ID.
@@ -120,48 +135,74 @@ func (e *Event) UnSubscribe(topic string, id int64) {
 			e.handlers.Remove(topic)
 		}
 	}
+}
+
+// executeHandlerWithRecover executes a handler function with a recover function for a topic.
+func (e *Event) executeHandlerWithRecover(info *handlerInfo, topic string, message any, async bool) {
+	handler := info.handler
+	recoverFunc := info.recoverFunc
+
+	wrapper := func() {
+		defer func() {
+			if err := recover(); err != nil {
+				recoverFunc(topic, message)
+			}
+		}()
+		handler(topic, message)
+	}
 
+	if async {
+		go wrapper()
+	} else {
+		wrapper()
+	}
+}
+
+// executeHandler executes a handler function for a topic.
+func (e *Event) executeHandler(handler Handler, topic string, message any, async bool) {
+	if async {
+		go handler(topic, message)
+	} else {
+		handler(topic, message)
+	}
 }
 
 // forEachHandler iterates through all handlers for a topic and executes them.
 // If async is true, handlers are executed in goroutines.
-func (e *Event) forEachHandler(topic string, message any, async bool) {
+func (e *Event) forEachHandler(topic string, message any, async bool) error {
+	if e.closed {
+		return EventClosedError
+	}
 	if m := e.handlers.Get(topic); m != nil {
 		m.(*gmap.TreeMap).IteratorDesc(func(key, value any) bool {
 			infos := value.([]*handlerInfo)
 			for _, info := range infos {
-				handler := info.handler
-				if async {
-					go handler(topic, message)
+				if info.recoverFunc != nil {
+					e.executeHandlerWithRecover(info, topic, message, async)
 				} else {
-					handler(topic, message)
+					e.executeHandler(info.handler, topic, message, async)
 				}
 			}
 			return true
 		})
 	}
+	return nil
 }
 
 // Publish sends a message to all handlers subscribed to a topic.
 // Handlers are executed asynchronously in separate goroutines.
-func (e *Event) Publish(topic string, message any) {
+func (e *Event) Publish(topic string, message any) error {
 	e.mu.RLock()
 	defer e.mu.RUnlock()
-	e.forEachHandler(topic, message, true)
+	return e.forEachHandler(topic, message, true)
 }
 
 // PublishSync sends a message to all handlers subscribed to a topic.
 // Handlers are executed synchronously in the current goroutine.
-func (e *Event) PublishSync(topic string, message any) {
+func (e *Event) PublishSync(topic string, message any) error {
 	e.mu.RLock()
 	defer e.mu.RUnlock()
-	e.forEachHandler(topic, message, false)
-}
-
-// SubscribeFunc is a convenience method that allows subscribing with a plain function.
-// It converts the function to a Handler type and subscribes it.
-func (e *Event) SubscribeFunc(topic string, f func(topic string, message any), priority ...Priority) *Subscriber {
-	return e.Subscribe(topic, f, priority...)
+	return e.forEachHandler(topic, message, false)
 }
 
 // Topics returns a list of all topics that currently have subscribers.
@@ -182,3 +223,24 @@ func (e *Event) SubscribersCount(topic string) int {
 	}
 	return count
 }
+
+// Clear removes all handlers and subscribers from the event manager.
+func (e *Event) Clear() {
+	e.mu.Lock()
+	defer e.mu.Unlock()
+	e.clear()
+}
+
+// clear removes all handlers and subscribers from the event manager.
+func (e *Event) clear() {
+	e.handlers.Clear()
+	e.counter = 0
+}
+
+// Close closes the event manager, preventing new subscribers from being added.
+func (e *Event) Close() {
+	e.mu.Lock()
+	defer e.mu.Unlock()
+	e.closed = true
+	e.clear()
+}