other: server 异步消息回调将不再使用 MessageTypeSystem，更改为 MessageTypeAsyncCallback

server/message.go

@@ -22,17 +22,21 @@ const (
 	// MessageTypeAsync 异步消息类型
 	MessageTypeAsync
 
+	// MessageTypeAsyncCallback 异步回调消息类型
+	MessageTypeAsyncCallback
+
 	// MessageTypeSystem 系统消息类型
 	MessageTypeSystem
 )
 
 var messageNames = map[MessageType]string{
-	MessageTypePacket: "MessageTypePacket",
+	MessageTypePacket:        "MessageTypePacket",
-	MessageTypeError:  "MessageTypeError",
+	MessageTypeError:         "MessageTypeError",
-	MessageTypeCross:  "MessageTypeCross",
+	MessageTypeCross:         "MessageTypeCross",
-	MessageTypeTicker: "MessageTypeTicker",
+	MessageTypeTicker:        "MessageTypeTicker",
-	MessageTypeAsync:  "MessageTypeAsync",
+	MessageTypeAsync:         "MessageTypeAsync",
-	MessageTypeSystem: "MessageTypeSystem",
+	MessageTypeAsyncCallback: "MessageTypeAsyncCallback",
+	MessageTypeSystem:        "MessageTypeSystem",
 }
 
 const (
@@ -146,6 +150,8 @@ func PushTickerMessage(srv *Server, caller func(), mark ...any) {
 //   - 异步消息将在服务器的异步消息队列中进行处理，处理完成 caller 的阻塞操作后，将会通过系统消息执行 callback 函数
 //   - callback 函数将在异步消息处理完成后进行调用，无论过程是否产生 err，都将被执行，允许为 nil
 //   - 需要注意的是，为了避免并发问题，caller 函数请仅处理阻塞操作，其他操作应该在 callback 函数中进行
+//
+// 在通过 WithShunt 使用分流服务器时，异步消息不会转换到分流通道中进行处理。依旧需要注意上方第三条
 func PushAsyncMessage(srv *Server, caller func() error, callback func(err error), mark ...any) {
 	msg := srv.messagePool.Get()
 	msg.t = MessageTypeAsync

server/server.go

@@ -545,7 +545,7 @@ func (slf *Server) pushMessage(message *Message) {
 	if slf.isShutdown.Load() {
 		return
 	}
-	if slf.shuntChannels != nil && (message.t == MessageTypePacket) {
+	if slf.shuntChannels != nil && message.t == MessageTypePacket {
 		conn := message.attrs[0].(*Conn)
 		channelGuid, allowToCreate := slf.shuntMatcher(conn)
 		channel, exist := slf.shuntChannels.GetExist(channelGuid)
@@ -681,15 +681,22 @@ func (slf *Server) dispatchMessage(msg *Message) {
 			}()
 			err := handle()
 			if cb && callback != nil {
-				PushSystemMessage(slf, func() {
+				acm := slf.messagePool.Get()
-					callback(err)
+				acm.t = MessageTypeAsyncCallback
-				}, "AsyncCallback")
+				if len(attrs) > 2 {
+					acm.attrs = append([]any{func() { callback(err) }}, attrs[2:]...)
+				} else {
+					acm.attrs = []any{func() { callback(err) }}
+				}
+				slf.pushMessage(acm)
 			} else if err != nil {
 				log.Error("Server", log.String("MessageType", messageNames[msg.t]), log.Any("error", err), log.String("stack", string(debug.Stack())))
 			}
 		}); err != nil {
 			panic(err)
 		}
+	case MessageTypeAsyncCallback:
+		attrs[0].(func())()
 	case MessageTypeSystem:
 		attrs[0].(func())()
 	default: