package server import ( "context" "fmt" "github.com/kercylan98/minotaur/utils/hash" "github.com/kercylan98/minotaur/utils/super" "reflect" ) const ( // MessageTypePacket 数据包消息类型:该类型的数据将被发送到 ConnectionReceivePacketEvent 进行处理 MessageTypePacket MessageType = iota // MessageTypeError 错误消息类型:根据不同的错误状态,将交由 Server 进行统一处理 MessageTypeError // MessageTypeCross 跨服消息类型:将被推送到跨服的 Cross 实现中进行处理 MessageTypeCross // MessageTypeTicker 定时器消息类型 MessageTypeTicker // MessageTypeAsync 异步消息类型 MessageTypeAsync // MessageTypeAsyncCallback 异步回调消息类型 MessageTypeAsyncCallback // MessageTypeSystem 系统消息类型 MessageTypeSystem ) var messageNames = map[MessageType]string{ MessageTypePacket: "MessageTypePacket", MessageTypeError: "MessageTypeError", MessageTypeCross: "MessageTypeCross", MessageTypeTicker: "MessageTypeTicker", MessageTypeAsync: "MessageTypeAsync", MessageTypeAsyncCallback: "MessageTypeAsyncCallback", MessageTypeSystem: "MessageTypeSystem", } const ( MessageErrorActionNone MessageErrorAction = iota + 1 // 错误消息类型操作:将不会被进行任何特殊处理,仅进行日志输出 MessageErrorActionShutdown // 错误消息类型操作:当接收到该类型的操作时,服务器将执行 Server.shutdown 函数 ) var messageErrorActionNames = map[MessageErrorAction]string{ MessageErrorActionNone: "None", MessageErrorActionShutdown: "Shutdown", } var ( messagePacketVisualization = func(packet []byte) string { return string(packet) } ) type ( // MessageType 消息类型 MessageType byte // MessageErrorAction 错误消息类型操作 MessageErrorAction byte ) // HasMessageType 检查是否存在指定的消息类型 func HasMessageType(mt MessageType) bool { return hash.Exist(messageNames, mt) } func (slf MessageErrorAction) String() string { return messageErrorActionNames[slf] } // Message 服务器消息 type Message struct { t MessageType // 消息类型 attrs []any // 消息属性 } // MessageType 返回消息类型 func (slf *Message) MessageType() MessageType { return slf.t } // AttrsString 返回消息属性的字符串表示 func (slf *Message) AttrsString() string { var attrs = make([]any, 0, len(slf.attrs)) for _, attr := range slf.attrs { if conn, hit := attr.(*Conn); hit { attrs = append(attrs, conn.GetID()) continue } if tof := reflect.TypeOf(attr); tof.Kind() == reflect.Func { attrs = append(attrs, tof.String()) continue } attrs = append(attrs, attr) } if len(attrs) == 0 { return "NoneAttr" } return string(super.MarshalJSON(attrs)) } // String 返回消息的字符串表示 func (slf *Message) String() string { var attrs = make([]any, 0, len(slf.attrs)) for _, attr := range slf.attrs { if reflect.TypeOf(attr).Kind() == reflect.Func { continue } attrs = append(attrs, attr) } return fmt.Sprintf("[%s] %s", slf.t, slf.AttrsString()) } // String 返回消息类型的字符串表示 func (slf MessageType) String() string { return messageNames[slf] } // GetPacketMessageAttrs 获取消息中的数据包属性 func (slf *Message) GetPacketMessageAttrs() (conn *Conn, packet []byte) { conn = slf.attrs[0].(*Conn) packet = slf.attrs[1].([]byte) return } // PushPacketMessage 向特定服务器中推送 MessageTypePacket 消息 func PushPacketMessage(srv *Server, conn *Conn, wst int, packet []byte, mark ...any) { msg := srv.messagePool.Get() msg.t = MessageTypePacket msg.attrs = append([]any{&Conn{ctx: context.WithValue(conn.ctx, contextKeyWST, wst), connection: conn.connection}, packet}, mark...) srv.pushMessage(msg) } // GetErrorMessageAttrs 获取消息中的错误属性 func (slf *Message) GetErrorMessageAttrs() (err error, action MessageErrorAction) { err = slf.attrs[0].(error) action = slf.attrs[1].(MessageErrorAction) return } // PushErrorMessage 向特定服务器中推送 MessageTypeError 消息 func PushErrorMessage(srv *Server, err error, action MessageErrorAction, mark ...any) { msg := srv.messagePool.Get() msg.t = MessageTypeError msg.attrs = append([]any{err, action}, mark...) srv.pushMessage(msg) } // GetCrossMessageAttrs 获取消息中的跨服属性 func (slf *Message) GetCrossMessageAttrs() (serverId string, packet []byte) { serverId = slf.attrs[0].(string) packet = slf.attrs[1].([]byte) return } // PushCrossMessage 向特定服务器中推送 MessageTypeCross 消息 func PushCrossMessage(srv *Server, crossName string, serverId string, packet []byte, mark ...any) { if serverId == srv.id { msg := srv.messagePool.Get() msg.t = MessageTypeCross msg.attrs = append([]any{serverId, packet}, mark...) srv.pushMessage(msg) } else { if len(srv.cross) == 0 { return } cross, exist := srv.cross[crossName] if !exist { return } _ = cross.PushMessage(serverId, packet) } } // GetTickerMessageAttrs 获取消息中的定时器属性 func (slf *Message) GetTickerMessageAttrs() (caller func()) { caller = slf.attrs[0].(func()) return } // PushTickerMessage 向特定服务器中推送 MessageTypeTicker 消息 func PushTickerMessage(srv *Server, caller func(), mark ...any) { srv.messageLock.RLock() if srv.messagePool == nil { srv.messageLock.RUnlock() return } srv.messageLock.RUnlock() msg := srv.messagePool.Get() msg.t = MessageTypeTicker msg.attrs = append([]any{caller}, mark...) srv.pushMessage(msg) } // GetAsyncMessageAttrs 获取消息中的异步消息属性 func (slf *Message) GetAsyncMessageAttrs() (caller func() error, callback func(err error), hasCallback bool) { caller = slf.attrs[0].(func() error) callback, hasCallback = slf.attrs[1].(func(err error)) return } // PushAsyncMessage 向特定服务器中推送 MessageTypeAsync 消息 // - 异步消息将在服务器的异步消息队列中进行处理,处理完成 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 msg.attrs = append([]any{caller, callback}, mark...) srv.pushMessage(msg) } // GetSystemMessageAttrs 获取消息中的系统消息属性 func (slf *Message) GetSystemMessageAttrs() (handle func()) { handle = slf.attrs[0].(func()) return } // PushSystemMessage 向特定服务器中推送 MessageTypeSystem 消息 func PushSystemMessage(srv *Server, handle func(), mark ...any) { msg := srv.messagePool.Get() msg.t = MessageTypeSystem msg.attrs = append([]any{handle}, mark...) srv.pushMessage(msg) } // SetMessagePacketVisualizer 设置消息可视化函数 // - 消息可视化将在慢消息等情况用于打印,使用自定消息可视化函数可以便于开发者进行调试 // - 默认的消息可视化函数将直接返回消息的字符串表示 func SetMessagePacketVisualizer(handle func(packet []byte) string) { messagePacketVisualization = handle }