优化登录、离线、服务器重启等会话连接问题

gate/net/forward.go

@@ -22,26 +22,23 @@ import (
 )
 
 const (
-	ConnWaitToken = iota
-	ConnWaitLogin
-	ConnAlive
+	ConnEst = iota
+	ConnActive
 	ConnClose
 )
 
 // 发送消息到GS
 func (k *KcpConnectManager) recvMsgHandle(protoMsg *ProtoMsg, session *Session) {
 	userId := session.userId
-	headMeta := session.headMeta
 	connState := session.connState
 	if protoMsg.HeadMessage == nil {
 		logger.Error("recv null head msg: %v", protoMsg)
 	}
-	headMeta.seq = protoMsg.HeadMessage.ClientSequenceId
 	// gate本地处理的请求
 	switch protoMsg.CmdId {
 	case cmd.GetPlayerTokenReq:
 		// 获取玩家token请求
-		if connState != ConnWaitToken {
+		if connState != ConnEst {
 			return
 		}
 		getPlayerTokenReq := protoMsg.PayloadMessage.(*proto.GetPlayerTokenReq)
@@ -56,51 +53,9 @@ func (k *KcpConnectManager) recvMsgHandle(protoMsg *ProtoMsg, session *Session)
 		rsp.HeadMessage = k.getHeadMsg(protoMsg.HeadMessage.ClientSequenceId)
 		rsp.PayloadMessage = getPlayerTokenRsp
 		k.localMsgOutput <- rsp
-	case cmd.PlayerLoginReq:
-		// 玩家登录请求
-		if connState != ConnWaitLogin {
-			return
-		}
-		playerLoginReq := protoMsg.PayloadMessage.(*proto.PlayerLoginReq)
-		playerLoginRsp := k.playerLogin(playerLoginReq, session)
-		if playerLoginRsp == nil {
-			return
-		}
-		// 返回数据到客户端
-		rsp := new(ProtoMsg)
-		rsp.ConvId = protoMsg.ConvId
-		rsp.CmdId = cmd.PlayerLoginRsp
-		rsp.HeadMessage = k.getHeadMsg(protoMsg.HeadMessage.ClientSequenceId)
-		rsp.PayloadMessage = playerLoginRsp
-		k.localMsgOutput <- rsp
-		// 登录成功 通知GS初始化相关数据
-		gameMsg := new(mq.GameMsg)
-		gameMsg.UserId = userId
-		gameMsg.ClientSeq = headMeta.seq
-		k.messageQueue.SendToGs("1", &mq.NetMsg{
-			MsgType: mq.MsgTypeGame,
-			EventId: mq.UserLoginNotify,
-			GameMsg: gameMsg,
-		})
-		logger.Info("send to gs user login ok, ConvId: %v, UserId: %v", protoMsg.ConvId, gameMsg.UserId)
-	case cmd.SetPlayerBornDataReq:
-		// 玩家注册请求
-		if connState != ConnAlive {
-			return
-		}
-		gameMsg := new(mq.GameMsg)
-		gameMsg.UserId = userId
-		gameMsg.CmdId = cmd.SetPlayerBornDataReq
-		gameMsg.ClientSeq = protoMsg.HeadMessage.ClientSequenceId
-		gameMsg.PayloadMessage = protoMsg.PayloadMessage
-		k.messageQueue.SendToGs("1", &mq.NetMsg{
-			MsgType: mq.MsgTypeGame,
-			EventId: mq.UserRegNotify,
-			GameMsg: gameMsg,
-		})
 	case cmd.PlayerForceExitReq:
 		// 玩家退出游戏请求
-		if connState != ConnAlive {
+		if connState != ConnActive {
 			return
 		}
 		k.kcpEventInput <- &KcpEvent{
@@ -110,13 +65,9 @@ func (k *KcpConnectManager) recvMsgHandle(protoMsg *ProtoMsg, session *Session)
 		}
 	case cmd.PingReq:
 		// ping请求
-		if connState != ConnAlive {
-			return
-		}
 		pingReq := protoMsg.PayloadMessage.(*proto.PingReq)
 		logger.Debug("user ping req, data: %v", pingReq.String())
 		// 返回数据到客户端
-		// TODO 记录客户端最后一次ping时间做超时下线处理
 		pingRsp := new(proto.PingRsp)
 		pingRsp.ClientTime = pingReq.ClientTime
 		rsp := new(ProtoMsg)
@@ -125,32 +76,45 @@ func (k *KcpConnectManager) recvMsgHandle(protoMsg *ProtoMsg, session *Session)
 		rsp.HeadMessage = k.getHeadMsg(protoMsg.HeadMessage.ClientSequenceId)
 		rsp.PayloadMessage = pingRsp
 		k.localMsgOutput <- rsp
-		// 通知GS玩家客户端的本地时钟
-		gameMsg := new(mq.GameMsg)
-		gameMsg.UserId = userId
-		gameMsg.ClientTime = pingReq.ClientTime
-		k.messageQueue.SendToGs("1", &mq.NetMsg{
-			MsgType: mq.MsgTypeGame,
-			EventId: mq.ClientTimeNotify,
-			GameMsg: gameMsg,
-		})
-		// RTT
 		logger.Debug("convId: %v, RTO: %v, SRTT: %v, RTTVar: %v", protoMsg.ConvId, session.conn.GetRTO(), session.conn.GetSRTT(), session.conn.GetSRTTVar())
-		rtt := session.conn.GetSRTT()
-		// 通知GS玩家客户端往返时延
-		gameMsg = new(mq.GameMsg)
-		gameMsg.UserId = userId
-		gameMsg.ClientRtt = uint32(rtt)
-		k.messageQueue.SendToGs("1", &mq.NetMsg{
-			MsgType: mq.MsgTypeGame,
-			EventId: mq.ClientRttNotify,
-			GameMsg: gameMsg,
-		})
-	default:
-		// 未登录禁止访问
-		if connState != ConnAlive {
+		if connState != ConnActive {
 			return
 		}
+		// 通知GS玩家客户端的本地时钟
+		connCtrlMsg := new(mq.ConnCtrlMsg)
+		connCtrlMsg.UserId = userId
+		connCtrlMsg.ClientTime = pingReq.ClientTime
+		k.messageQueue.SendToGs("1", &mq.NetMsg{
+			MsgType:     mq.MsgTypeConnCtrl,
+			EventId:     mq.ClientTimeNotify,
+			ConnCtrlMsg: connCtrlMsg,
+		})
+		// 通知GS玩家客户端往返时延
+		rtt := session.conn.GetSRTT()
+		connCtrlMsg = new(mq.ConnCtrlMsg)
+		connCtrlMsg.UserId = userId
+		connCtrlMsg.ClientRtt = uint32(rtt)
+		k.messageQueue.SendToGs("1", &mq.NetMsg{
+			MsgType:     mq.MsgTypeConnCtrl,
+			EventId:     mq.ClientRttNotify,
+			ConnCtrlMsg: connCtrlMsg,
+		})
+	default:
+		if connState != ConnActive && !(protoMsg.CmdId == cmd.PlayerLoginReq || protoMsg.CmdId == cmd.SetPlayerBornDataReq) {
+			logger.Error("conn not active so drop packet, cmdId: %v, userId: %v, convId: %v", protoMsg.CmdId, userId, protoMsg.ConvId)
+			return
+		}
+		// 转发到GS
+		gameMsg := new(mq.GameMsg)
+		gameMsg.UserId = userId
+		gameMsg.CmdId = protoMsg.CmdId
+		gameMsg.ClientSeq = protoMsg.HeadMessage.ClientSequenceId
+		gameMsg.PayloadMessage = protoMsg.PayloadMessage
+		k.messageQueue.SendToGs("1", &mq.NetMsg{
+			MsgType: mq.MsgTypeGame,
+			EventId: mq.NormalMsg,
+			GameMsg: gameMsg,
+		})
 		// 转发到FIGHT
 		if protoMsg.CmdId == cmd.CombatInvocationsNotify {
 			gameMsg := new(mq.GameMsg)
@@ -164,17 +128,6 @@ func (k *KcpConnectManager) recvMsgHandle(protoMsg *ProtoMsg, session *Session)
 				GameMsg: gameMsg,
 			})
 		}
-		// 转发到GS
-		gameMsg := new(mq.GameMsg)
-		gameMsg.UserId = userId
-		gameMsg.CmdId = protoMsg.CmdId
-		gameMsg.ClientSeq = protoMsg.HeadMessage.ClientSequenceId
-		gameMsg.PayloadMessage = protoMsg.PayloadMessage
-		k.messageQueue.SendToGs("1", &mq.NetMsg{
-			MsgType: mq.MsgTypeGame,
-			EventId: mq.NormalMsg,
-			GameMsg: gameMsg,
-		})
 	}
 }
 
@@ -208,26 +161,40 @@ func (k *KcpConnectManager) sendMsgHandle() {
 		case protoMsg := <-k.localMsgOutput:
 			sendToClientFn(protoMsg)
 		case netMsg := <-k.messageQueue.GetNetMsg():
-			if netMsg.MsgType != mq.MsgTypeGame {
-				logger.Error("recv unknown msg type from game server, msg type: %v", netMsg.MsgType)
-				continue
+			switch netMsg.MsgType {
+			case mq.MsgTypeGame:
+				if netMsg.EventId != mq.NormalMsg {
+					logger.Error("recv unknown event from game server, event id: %v", netMsg.EventId)
+					continue
+				}
+				gameMsg := netMsg.GameMsg
+				convId, exist := userIdConvMap[gameMsg.UserId]
+				if !exist {
+					logger.Error("can not find convId by userId")
+					continue
+				}
+				protoMsg := new(ProtoMsg)
+				protoMsg.ConvId = convId
+				protoMsg.CmdId = gameMsg.CmdId
+				protoMsg.HeadMessage = k.getHeadMsg(gameMsg.ClientSeq)
+				protoMsg.PayloadMessage = gameMsg.PayloadMessage
+				sendToClientFn(protoMsg)
+			case mq.MsgTypeConnCtrl:
+				if netMsg.EventId != mq.KickPlayerNotify {
+					continue
+				}
+				connCtrlMsg := netMsg.ConnCtrlMsg
+				convId, exist := userIdConvMap[connCtrlMsg.KickUserId]
+				if !exist {
+					logger.Error("can not find convId by userId")
+					continue
+				}
+				k.kcpEventInput <- &KcpEvent{
+					ConvId:       convId,
+					EventId:      KcpConnForceClose,
+					EventMessage: connCtrlMsg.KickReason,
+				}
 			}
-			if netMsg.EventId != mq.NormalMsg {
-				logger.Error("recv unknown event from game server, event id: %v", netMsg.EventId)
-				continue
-			}
-			gameMsg := netMsg.GameMsg
-			convId, exist := userIdConvMap[gameMsg.UserId]
-			if !exist {
-				logger.Error("can not find convId by userId")
-				continue
-			}
-			protoMsg := new(ProtoMsg)
-			protoMsg.ConvId = convId
-			protoMsg.CmdId = gameMsg.CmdId
-			protoMsg.HeadMessage = k.getHeadMsg(gameMsg.ClientSeq)
-			protoMsg.PayloadMessage = gameMsg.PayloadMessage
-			sendToClientFn(protoMsg)
 		}
 	}
 }
@@ -276,15 +243,16 @@ func (k *KcpConnectManager) getPlayerToken(req *proto.GetPlayerTokenReq, session
 	oldSession := k.GetSessionByUserId(tokenVerifyRsp.PlayerID)
 	if oldSession != nil {
 		// 顶号
+		kickFinishNotifyChan := make(chan bool)
 		k.kcpEventInput <- &KcpEvent{
 			ConvId:       oldSession.conn.GetConv(),
-			EventId:      KcpConnForceClose,
-			EventMessage: uint32(kcp.EnetServerRelogin),
+			EventId:      KcpConnRelogin,
+			EventMessage: kickFinishNotifyChan,
 		}
+		<-kickFinishNotifyChan
 	}
 	// 关联玩家uid和连接信息
 	session.userId = tokenVerifyRsp.PlayerID
-	session.connState = ConnWaitLogin
 	k.SetSession(session, session.conn.GetConv(), session.userId)
 	k.createSessionChan <- session
 	// 返回响应
@@ -344,7 +312,6 @@ func (k *KcpConnectManager) getPlayerToken(req *proto.GetPlayerTokenReq, session
 		timeRand := random.GetTimeRand()
 		serverSeedUint64 := timeRand.Uint64()
 		session.seed = serverSeedUint64
-		session.changeXorKey = true
 		seedUint64 := serverSeedUint64 ^ clientSeedUint64
 		seedBuf := new(bytes.Buffer)
 		err = binary.Write(seedBuf, binary.BigEndian, seedUint64)
@@ -369,27 +336,3 @@ func (k *KcpConnectManager) getPlayerToken(req *proto.GetPlayerTokenReq, session
 	}
 	return rsp
 }
-
-func (k *KcpConnectManager) playerLogin(req *proto.PlayerLoginReq, session *Session) (rsp *proto.PlayerLoginRsp) {
-	logger.Debug("player login, info: %v", req.String())
-	// TODO 验证token
-	session.connState = ConnAlive
-	// 返回响应
-	rsp = new(proto.PlayerLoginRsp)
-	rsp.IsUseAbilityHash = true
-	rsp.AbilityHashCode = -228935105
-	rsp.GameBiz = "hk4e_cn"
-	rsp.IsScOpen = false
-	rsp.RegisterCps = "taptap"
-	rsp.CountryCode = "CN"
-	rsp.Birthday = "2000-01-01"
-	rsp.TotalTickTime = 1185941.871788
-	rsp.ClientDataVersion = k.regionCurr.RegionInfo.ClientDataVersion
-	rsp.ClientSilenceDataVersion = k.regionCurr.RegionInfo.ClientSilenceDataVersion
-	rsp.ClientMd5 = k.regionCurr.RegionInfo.ClientDataMd5
-	rsp.ClientSilenceMd5 = k.regionCurr.RegionInfo.ClientSilenceDataMd5
-	rsp.ResVersionConfig = k.regionCurr.RegionInfo.ResVersionConfig
-	rsp.ClientVersionSuffix = k.regionCurr.RegionInfo.ClientVersionSuffix
-	rsp.ClientSilenceVersionSuffix = k.regionCurr.RegionInfo.ClientSilenceVersionSuffix
-	return rsp
-}

gate/net/kcp_connect_manager.go

@@ -16,7 +16,6 @@ import (
 	"hk4e/pkg/logger"
 	"hk4e/pkg/random"
 	"hk4e/protocol/cmd"
-	"hk4e/protocol/proto"
 )
 
 const PacketFreqLimit = 1000
@@ -35,7 +34,6 @@ type KcpConnectManager struct {
 	destroySessionChan chan *Session
 	// 密钥相关
 	dispatchKey  []byte
-	regionCurr   *proto.QueryCurrRegionHttpRsp
 	signRsaKey   []byte
 	encRsaKeyMap map[string][]byte
 }
@@ -58,9 +56,6 @@ func NewKcpConnectManager(messageQueue *mq.MessageQueue) (r *KcpConnectManager)
 func (k *KcpConnectManager) Start() {
 	// 读取密钥相关文件
 	k.signRsaKey, k.encRsaKeyMap, _ = region.LoadRsaKey()
-	// region
-	regionCurr, _, _ := region.InitRegion(config.CONF.Hk4e.KcpAddr, config.CONF.Hk4e.KcpPort)
-	k.regionCurr = regionCurr
 	// key
 	dispatchEc2bSeedRsp, err := httpclient.Get[controller.DispatchEc2bSeedRsp]("http://127.0.0.1:8080/dispatch/ec2b/seed", "")
 	if err != nil {
@@ -89,6 +84,11 @@ func (k *KcpConnectManager) Start() {
 	go k.acceptHandle(listener)
 }
 
+func (k *KcpConnectManager) Stop() {
+	k.closeAllKcpConn()
+	time.Sleep(time.Second * 3)
+}
+
 func (k *KcpConnectManager) acceptHandle(listener *kcp.Listener) {
 	logger.Debug("accept handle start")
 	for {
@@ -107,16 +107,13 @@ func (k *KcpConnectManager) acceptHandle(listener *kcp.Listener) {
 		logger.Debug("client connect, convId: %v", convId)
 		kcpRawSendChan := make(chan *ProtoMsg, 1000)
 		session := &Session{
-			conn:      conn,
-			connState: ConnWaitToken,
-			userId:    0,
-			headMeta: &ClientHeadMeta{
-				seq: 0,
-			},
-			kcpRawSendChan: kcpRawSendChan,
-			seed:           0,
-			xorKey:         k.dispatchKey,
-			changeXorKey:   false,
+			conn:            conn,
+			connState:       ConnEst,
+			userId:          0,
+			kcpRawSendChan:  kcpRawSendChan,
+			seed:            0,
+			xorKey:          k.dispatchKey,
+			changeXorKeyFin: false,
 		}
 		go k.recvHandle(session)
 		go k.sendHandle(session)
@@ -203,24 +200,19 @@ func (k *KcpConnectManager) enetHandle(listener *kcp.Listener) {
 	}
 }
 
-type ClientHeadMeta struct {
-	seq uint32
-}
-
 type Session struct {
-	conn           *kcp.UDPSession
-	connState      uint8
-	userId         uint32
-	headMeta       *ClientHeadMeta
-	kcpRawSendChan chan *ProtoMsg
-	seed           uint64
-	xorKey         []byte
-	changeXorKey   bool
+	conn            *kcp.UDPSession
+	connState       uint8
+	userId          uint32
+	kcpRawSendChan  chan *ProtoMsg
+	seed            uint64
+	xorKey          []byte
+	changeXorKeyFin bool
 }
 
+// 接收
 func (k *KcpConnectManager) recvHandle(session *Session) {
 	logger.Debug("recv handle start")
-	// 接收
 	conn := session.conn
 	convId := conn.GetConv()
 	pktFreqLimitCounter := 0
@@ -234,14 +226,6 @@ func (k *KcpConnectManager) recvHandle(session *Session) {
 			k.closeKcpConn(session, kcp.EnetServerKick)
 			break
 		}
-		if session.changeXorKey {
-			session.changeXorKey = false
-			keyBlock := random.NewKeyBlock(session.seed)
-			xorKey := keyBlock.XorKey()
-			key := make([]byte, 4096)
-			copy(key, xorKey[:])
-			session.xorKey = key
-		}
 		// 收包频率限制
 		pktFreqLimitCounter++
 		now := time.Now().UnixNano()
@@ -267,9 +251,9 @@ func (k *KcpConnectManager) recvHandle(session *Session) {
 	}
 }
 
+// 发送
 func (k *KcpConnectManager) sendHandle(session *Session) {
 	logger.Debug("send handle start")
-	// 发送
 	conn := session.conn
 	convId := conn.GetConv()
 	for {
@@ -292,9 +276,33 @@ func (k *KcpConnectManager) sendHandle(session *Session) {
 			k.closeKcpConn(session, kcp.EnetServerKick)
 			break
 		}
+		if session.changeXorKeyFin == false && protoMsg.CmdId == cmd.GetPlayerTokenRsp {
+			logger.Debug("change session xor key, convId: %v", convId)
+			session.changeXorKeyFin = true
+			keyBlock := random.NewKeyBlock(session.seed)
+			xorKey := keyBlock.XorKey()
+			key := make([]byte, 4096)
+			copy(key, xorKey[:])
+			session.xorKey = key
+		}
+		if protoMsg.CmdId == cmd.PlayerLoginRsp {
+			logger.Debug("session active, convId: %v", convId)
+			session.connState = ConnActive
+		}
 	}
 }
 
+func (k *KcpConnectManager) forceCloseKcpConn(convId uint64, reason uint32) {
+	// 强制关闭某个连接
+	session := k.GetSessionByConvId(convId)
+	if session == nil {
+		logger.Error("session not exist, convId: %v", convId)
+		return
+	}
+	k.closeKcpConn(session, reason)
+	logger.Info("conn has been force close, convId: %v", convId)
+}
+
 func (k *KcpConnectManager) closeKcpConn(session *Session, enetType uint32) {
 	if session.connState == ConnClose {
 		return
@@ -330,19 +338,14 @@ func (k *KcpConnectManager) closeKcpConn(session *Session, enetType uint32) {
 }
 
 func (k *KcpConnectManager) closeAllKcpConn() {
-	closeConnList := make([]*kcp.UDPSession, 0)
+	sessionList := make([]*Session, 0)
 	k.sessionMapLock.RLock()
-	for _, v := range k.sessionConvIdMap {
-		closeConnList = append(closeConnList, v.conn)
+	for _, session := range k.sessionConvIdMap {
+		sessionList = append(sessionList, session)
 	}
 	k.sessionMapLock.RUnlock()
-	for _, v := range closeConnList {
-		// 关闭连接
-		v.SendEnetNotify(&kcp.Enet{
-			ConnType: kcp.ConnEnetFin,
-			EnetType: kcp.EnetServerShutdown,
-		})
-		_ = v.Close()
+	for _, session := range sessionList {
+		k.closeKcpConn(session, kcp.EnetServerShutdown)
 	}
 }
 

gate/net/kcp_event.go

@@ -11,6 +11,7 @@ const (
 	KcpConnForceClose = iota
 	KcpAllConnForceClose
 	KcpGateOpenState
+	KcpConnRelogin
 	KcpConnCloseNotify
 	KcpConnEstNotify
 	KcpConnAddrChangeNotify
@@ -38,23 +39,12 @@ func (k *KcpConnectManager) eventHandle() {
 		logger.Info("kcp manager recv event, ConvId: %v, EventId: %v, EventMessage Type: %v", event.ConvId, event.EventId, reflect.TypeOf(event.EventMessage))
 		switch event.EventId {
 		case KcpConnForceClose:
-			// 强制关闭某个连接
-			session := k.GetSessionByConvId(event.ConvId)
-			if session == nil {
-				logger.Error("session not exist, convId: %v", event.ConvId)
-				continue
-			}
 			reason, ok := event.EventMessage.(uint32)
 			if !ok {
 				logger.Error("event KcpConnForceClose msg type error")
-				continue
+				return
 			}
-			session.conn.SendEnetNotify(&kcp.Enet{
-				ConnType: kcp.ConnEnetFin,
-				EnetType: reason,
-			})
-			_ = session.conn.Close()
-			logger.Info("conn has been force close, convId: %v", event.ConvId)
+			k.forceCloseKcpConn(event.ConvId, reason)
 		case KcpAllConnForceClose:
 			// 强制关闭所有连接
 			k.closeAllKcpConn()
@@ -70,6 +60,14 @@ func (k *KcpConnectManager) eventHandle() {
 			if openState == false {
 				k.closeAllKcpConn()
 			}
+		case KcpConnRelogin:
+			kickFinishNotifyChan, ok := event.EventMessage.(chan bool)
+			if !ok {
+				logger.Error("event KcpConnRelogin msg type error")
+				continue
+			}
+			k.forceCloseKcpConn(event.ConvId, kcp.EnetServerRelogin)
+			kickFinishNotifyChan <- true
 		}
 	}
 }