fix: 修复 server.Server 部分事件中发生 panic 导致程序退出的问题

This commit is contained in:
kercylan98 2023-07-14 12:40:13 +08:00
parent 10fcb54322
commit 0215d9ff8c
3 changed files with 75 additions and 34 deletions

View File

@ -6,6 +6,7 @@ import (
"github.com/kercylan98/minotaur/utils/runtimes"
"go.uber.org/zap"
"reflect"
"runtime/debug"
"sync"
"time"
)
@ -45,9 +46,11 @@ func (slf *event) RegStopEvent(handle StopEventHandle) {
}
func (slf *event) OnStopEvent() {
PushSystemMessage(slf.Server, func() {
for _, handle := range slf.stopEventHandles {
handle(slf.Server)
}
})
}
// RegConsoleCommandEvent 控制台收到指令时将立即执行被注册的事件处理函数
@ -69,6 +72,7 @@ func (slf *event) RegConsoleCommandEvent(command string, handle ConsoleCommandEv
}
func (slf *event) OnConsoleCommandEvent(command string) {
PushSystemMessage(slf.Server, func() {
handles, exist := slf.consoleCommandEventHandles[command]
if !exist {
switch command {
@ -83,6 +87,7 @@ func (slf *event) OnConsoleCommandEvent(command string) {
handle(slf.Server)
}
}
})
}
// RegStartBeforeEvent 在服务器初始化完成启动前立刻执行被注册的事件处理函数
@ -92,6 +97,12 @@ func (slf *event) RegStartBeforeEvent(handle StartBeforeEventHandle) {
}
func (slf *event) OnStartBeforeEvent() {
defer func() {
if err := recover(); err != nil {
log.Error("Server", zap.String("OnStartBeforeEvent", fmt.Sprintf("%v", err)))
debug.PrintStack()
}
}()
for _, handle := range slf.startBeforeEventHandles {
handle(slf.Server)
}
@ -104,9 +115,11 @@ func (slf *event) RegStartFinishEvent(handle StartFinishEventHandle) {
}
func (slf *event) OnStartFinishEvent() {
PushSystemMessage(slf.Server, func() {
for _, handle := range slf.startFinishEventHandles {
handle(slf.Server)
}
})
}
// RegConnectionClosedEvent 在连接关闭后将立刻执行被注册的事件处理函数
@ -119,11 +132,13 @@ func (slf *event) RegConnectionClosedEvent(handle ConnectionClosedEventHandle) {
}
func (slf *event) OnConnectionClosedEvent(conn *Conn, err any) {
PushSystemMessage(slf.Server, func() {
for _, handle := range slf.connectionClosedEventHandles {
handle(slf.Server, conn, err)
}
conn.Close()
slf.Server.online.Delete(conn.GetID())
})
}
// RegConnectionOpenedEvent 在连接打开后将立刻执行被注册的事件处理函数
@ -136,10 +151,12 @@ func (slf *event) RegConnectionOpenedEvent(handle ConnectionOpenedEventHandle) {
}
func (slf *event) OnConnectionOpenedEvent(conn *Conn) {
PushSystemMessage(slf.Server, func() {
slf.Server.online.Set(conn.GetID(), conn)
for _, handle := range slf.connectionOpenedEventHandles {
handle(slf.Server, conn)
}
})
}
// RegConnectionReceivePacketEvent 在接收到数据包时将立刻执行被注册的事件处理函数
@ -176,9 +193,11 @@ func (slf *event) RegMessageErrorEvent(handle MessageErrorEventHandle) {
}
func (slf *event) OnMessageErrorEvent(message *Message, err error) {
PushSystemMessage(slf.Server, func() {
for _, handle := range slf.messageErrorEventHandles {
handle(slf.Server, message, err)
}
})
}
// RegMessageLowExecEvent 在处理消息缓慢时将立即执行被注册的事件处理函数
@ -188,9 +207,11 @@ func (slf *event) RegMessageLowExecEvent(handle MessageLowExecEventHandle) {
}
func (slf *event) OnMessageLowExecEvent(message *Message, cost time.Duration) {
PushSystemMessage(slf.Server, func() {
for _, handle := range slf.messageLowExecEventHandles {
handle(slf.Server, message, cost)
}
})
}
func (slf *event) check() {

View File

@ -21,6 +21,9 @@ const (
// MessageTypeAsync 异步消息类型
MessageTypeAsync
// MessageTypeSystem 系统消息类型
MessageTypeSystem
)
var messageNames = map[MessageType]string{
@ -29,6 +32,7 @@ var messageNames = map[MessageType]string{
MessageTypeCross: "MessageTypeCross",
MessageTypeTicker: "MessageTypeTicker",
MessageTypeAsync: "MessageTypeAsync",
MessageTypeSystem: "MessageTypeSystem",
}
const (
@ -146,6 +150,14 @@ func PushAsyncMessage(srv *Server, caller func() error, callback func(err error)
srv.pushMessage(msg)
}
// 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 设置消息可视化函数
// - 消息可视化将在慢消息等情况用于打印,使用自定消息可视化函数可以便于开发者进行调试
// - 默认的消息可视化函数将直接返回消息的字符串表示

View File

@ -114,6 +114,7 @@ func (slf *Server) Run(addr string) error {
slf.event.check()
slf.addr = addr
var protoAddr = fmt.Sprintf("%s://%s", slf.network, slf.addr)
var messageInitFinish = make(chan struct{}, 1)
var connectionInitHandle = func(callback func()) {
slf.messagePool = synchronization.NewPool[*Message](slf.messagePoolSize,
func() *Message {
@ -132,6 +133,7 @@ func (slf *Server) Run(addr string) error {
go callback()
}
go func() {
messageInitFinish <- struct{}{}
for message := range slf.messageChannel {
slf.dispatchMessage(message)
}
@ -312,6 +314,10 @@ func (slf *Server) Run(addr string) error {
return ErrCanNotSupportNetwork
}
<-messageInitFinish
close(messageInitFinish)
messageInitFinish = nil
fmt.Println("messageInitFinish")
if slf.multiple == nil {
log.Info("Server", zap.String(serverMark, "===================================================================="))
log.Info("Server", zap.String(serverMark, "RunningInfo"),
@ -568,6 +574,8 @@ func (slf *Server) dispatchMessage(msg *Message) {
}); err != nil {
panic(err)
}
case MessageTypeSystem:
attrs[0].(func())()
default:
log.Warn("Server", zap.String("not support message type", msg.t.String()))
}