v0.0.1开发：增加玩家断开客服接口

bff/authbff/api/etc/authbff.yaml

@@ -5,7 +5,7 @@ Port: 10000
 #链路追踪
 Telemetry:
   Name: authbff-api
-  Endpoint: http://127.0.0.1:14268/api/traces
+  Endpoint: http://localhost:14268/api/traces
   Sampler: 1.0
   Batcher: jaeger
 

bff/cmdbff/api/desc/bean.api

@@ -46,8 +46,8 @@ type (
     }
 
     CsConnectPlayerReq {
-        PlayerId string `json:"player_id"`
         GameId string `json:"game_id"`
+        PlayerId string `json:"player_id"`
     }
 
     CsFetchHistoryChatReq {
@@ -62,8 +62,8 @@ type (
     }
 
     CsFetchHistoryMsgReq {
-        PlayerId string `json:"player_id"`
         GameId string `json:"game_id"`
+        PlayerId string `json:"player_id"`
         Page int64 `json:"page"`
         Limit int64 `json:"limit"`
     }
@@ -75,13 +75,13 @@ type (
     }
 
     CsFetchMsgReq {
-        PlayerId string `json:"player_id"`
         GameId string `json:"game_id"`
+        PlayerId string `json:"player_id"`
     }
 
     CsSendMsgReq {
-        PlayerId string `json:"player_id"`
         GameId string `json:"game_id"`
+        PlayerId string `json:"player_id"`
         Content string `json:"content"`
         Pic string `json:"pic"`
     }

bff/cmdbff/api/desc/cmdbff.api

@@ -28,6 +28,10 @@ service cmdbff {
 	@handler playerSendMsg
 	post /player/send-msg (PlayerSendMsgReq)
 	
+	@doc "玩家断开连接客服"
+	@handler playerDisconnect
+	post /player/disconnect
+	
 	@doc "客服获取玩家等待队列"
 	@handler csFetchPlayerQueue
 	post /cs/fetch-player-queue (CsFetchPlayerQueueReq) returns (CsFetchPlayerQueueResp)

bff/cmdbff/api/etc/cmdbff.yaml

@@ -4,7 +4,7 @@ Port: 10100
 
 Telemetry:
   Name: cmdbff-api
-  Endpoint: http://127.0.0.1:14268/api/traces
+  Endpoint: http://localhost:14268/api/traces
   Sampler: 1.0
   Batcher: jaeger
 

bff/cmdbff/api/internal/handler/playerdisconnecthandler.go

@@ -0,0 +1,17 @@
+package handler
+
+import (
+	"net/http"
+	"ylink/comm/result"
+
+	"ylink/bff/cmdbff/api/internal/logic"
+	"ylink/bff/cmdbff/api/internal/svc"
+)
+
+func playerDisconnectHandler(svcCtx *svc.ServiceContext) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		l := logic.NewPlayerDisconnectLogic(r.Context(), svcCtx)
+		err := l.PlayerDisconnect()
+		result.HttpResult(r, w, nil, err)
+	}
+}

bff/cmdbff/api/internal/handler/routes.go

@@ -27,6 +27,11 @@ func RegisterHandlers(server *rest.Server, serverCtx *svc.ServiceContext) {
 				Path:    "/player/send-msg",
 				Handler: playerSendMsgHandler(serverCtx),
 			},
+			{
+				Method:  http.MethodPost,
+				Path:    "/player/disconnect",
+				Handler: playerDisconnectHandler(serverCtx),
+			},
 			{
 				Method:  http.MethodPost,
 				Path:    "/cs/fetch-player-queue",

bff/cmdbff/api/internal/logic/csconnectplayerlogic.go

@@ -29,8 +29,8 @@ func (l *CsConnectPlayerLogic) CsConnectPlayer(req *types.CsConnectPlayerReq) er
 	csId := ctxdata.GetCsIdFromCtx(l.ctx)
 	_, err := l.svcCtx.CmdRpc.CsConnectPlayer(l.ctx, &cmd.CsConnectPlayerReq{
 		CsId:     csId,
-		PlayerId: req.PlayerId,
 		GameId:   req.GameId,
+		PlayerId: req.PlayerId,
 	})
 	if err != nil {
 		return err

bff/cmdbff/api/internal/logic/csfetchhistorymsglogic.go

@@ -29,8 +29,8 @@ func (l *CsFetchHistoryMsgLogic) CsFetchHistoryMsg(req *types.CsFetchHistoryMsgR
 	csId := ctxdata.GetCsIdFromCtx(l.ctx)
 	cmdResp, err := l.svcCtx.CmdRpc.CsFetchHistoryMsg(l.ctx, &cmd.CsFetchHistoryMsgReq{
 		CsId:     csId,
-		PlayerId: req.PlayerId,
 		GameId:   req.GameId,
+		PlayerId: req.PlayerId,
 		Page:     req.Page,
 		Limit:    req.Limit,
 	})

bff/cmdbff/api/internal/logic/cssendmsglogic.go

@@ -29,8 +29,8 @@ func (l *CsSendMsgLogic) CsSendMsg(req *types.CsSendMsgReq) error {
 	csId := ctxdata.GetCsIdFromCtx(l.ctx)
 	_, err := l.svcCtx.CmdRpc.CsSendMsg(l.ctx, &cmd.CsSendMsgReq{
 		CsId:     csId,
-		PlayerId: req.PlayerId,
 		GameId:   req.GameId,
+		PlayerId: req.PlayerId,
 		Content:  req.Content,
 		Pic:      req.Pic,
 	})

bff/cmdbff/api/internal/logic/playerdisconnectlogic.go

@@ -0,0 +1,34 @@
+package logic
+
+import (
+	"context"
+	"ylink/comm/ctxdata"
+	"ylink/core/cmd/rpc/cmd"
+
+	"github.com/zeromicro/go-zero/core/logx"
+	"ylink/bff/cmdbff/api/internal/svc"
+)
+
+type PlayerDisconnectLogic struct {
+	logx.Logger
+	ctx    context.Context
+	svcCtx *svc.ServiceContext
+}
+
+func NewPlayerDisconnectLogic(ctx context.Context, svcCtx *svc.ServiceContext) *PlayerDisconnectLogic {
+	return &PlayerDisconnectLogic{
+		Logger: logx.WithContext(ctx),
+		ctx:    ctx,
+		svcCtx: svcCtx,
+	}
+}
+
+func (l *PlayerDisconnectLogic) PlayerDisconnect() error {
+	gameId := ctxdata.GetGameIdFromCtx(l.ctx)
+	playerId := ctxdata.GetPlayerIdFromCtx(l.ctx)
+	_, err := l.svcCtx.CmdRpc.PlayerDisconnect(l.ctx, &cmd.PlayerDisconnectReq{
+		GameId:   gameId,
+		PlayerId: playerId,
+	})
+	return err
+}

bff/cmdbff/api/internal/logic/playerfetchcsinfologic.go

@@ -26,16 +26,14 @@ func NewPlayerFetchCsInfoLogic(ctx context.Context, svcCtx *svc.ServiceContext)
 }
 
 func (l *PlayerFetchCsInfoLogic) PlayerFetchCsInfo(req *types.PlayerFetchCsInfoReq) (resp *types.PlayerFetchCsInfoResp, err error) {
-	playerId := ctxdata.GetPlayerIdFromCtx(l.ctx)
 	gameId := ctxdata.GetGameIdFromCtx(l.ctx)
-	l.Logger.Infof("player id: %s", playerId)
+	playerId := ctxdata.GetPlayerIdFromCtx(l.ctx)
 	cmdResp, err := l.svcCtx.CmdRpc.PlayerFetchCsInfo(l.ctx, &cmd.PlayerFetchCsInfoReq{
-		PlayerId: playerId,
 		GameId:   gameId,
+		PlayerId: playerId,
 		CsId:     req.CsId,
 	})
 	if err != nil {
-		l.Logger.Info(err.Error())
 		return nil, err
 	}
 	return &types.PlayerFetchCsInfoResp{

bff/cmdbff/api/internal/logic/playerfetchhistorymsglogic.go

@@ -26,11 +26,11 @@ func NewPlayerFetchHistoryMsgLogic(ctx context.Context, svcCtx *svc.ServiceConte
 }
 
 func (l *PlayerFetchHistoryMsgLogic) PlayerFetchHistoryMsg(req *types.PlayerFetchHistoryMsgReq) (resp *types.PlayerFetchHistoryMsgResp, err error) {
-	playerId := ctxdata.GetPlayerIdFromCtx(l.ctx)
 	gameId := ctxdata.GetGameIdFromCtx(l.ctx)
+	playerId := ctxdata.GetPlayerIdFromCtx(l.ctx)
 	cmdResp, err := l.svcCtx.CmdRpc.PlayerFetchHistoryMsg(l.ctx, &pb.PlayerFetchHistoryMsgReq{
-		PlayerId: playerId,
 		GameId:   gameId,
+		PlayerId: playerId,
 		Page:     req.Page,
 		Limit:    req.Limit,
 	})

bff/cmdbff/api/internal/logic/playersendmsglogic.go

@@ -26,11 +26,11 @@ func NewPlayerSendMsgLogic(ctx context.Context, svcCtx *svc.ServiceContext) *Pla
 }
 
 func (l *PlayerSendMsgLogic) PlayerSendMsg(req *types.PlayerSendMsgReq) error {
-	playerId := ctxdata.GetPlayerIdFromCtx(l.ctx)
 	gameId := ctxdata.GetGameIdFromCtx(l.ctx)
+	playerId := ctxdata.GetPlayerIdFromCtx(l.ctx)
 	_, err := l.svcCtx.CmdRpc.PlayerSendMsg(l.ctx, &pb.PlayerSendMsgReq{
-		PlayerId: playerId,
 		GameId:   gameId,
+		PlayerId: playerId,
 		Content:  req.Content,
 		Pic:      req.Pic,
 	})

bff/cmdbff/api/internal/types/types.go

@@ -38,8 +38,8 @@ type CsFetchPlayerQueueResp struct {
 }
 
 type CsConnectPlayerReq struct {
-	PlayerId string `json:"player_id"`
 	GameId   string `json:"game_id"`
+	PlayerId string `json:"player_id"`
 }
 
 type CsFetchHistoryChatReq struct {
@@ -54,8 +54,8 @@ type CsFetchHistoryChatResp struct {
 }
 
 type CsFetchHistoryMsgReq struct {
-	PlayerId string `json:"player_id"`
 	GameId   string `json:"game_id"`
+	PlayerId string `json:"player_id"`
 	Page     int64  `json:"page"`
 	Limit    int64  `json:"limit"`
 }
@@ -67,13 +67,13 @@ type CsFetchHistoryMsgResp struct {
 }
 
 type CsFetchMsgReq struct {
-	PlayerId string `json:"player_id"`
 	GameId   string `json:"game_id"`
+	PlayerId string `json:"player_id"`
 }
 
 type CsSendMsgReq struct {
-	PlayerId string `json:"player_id"`
 	GameId   string `json:"game_id"`
+	PlayerId string `json:"player_id"`
 	Content  string `json:"content"`
 	Pic      string `json:"pic"`
 }

comm/globalkey/globalkey.go

@@ -5,6 +5,11 @@
 package globalkey
 
 const (
-	CONNECT_TYPE_PLAYER = 0
-	CONNECT_TYPE_CS     = 1
+	ConnectTypePlayer = 0
+	ConnectTypeCs     = 1
+)
+
+const (
+	EventRemoveTimeoutJob = "EventRemoveTimeoutJob"
+	EventUnsubscribeRmq   = "EventUnsubscribeRmq"
 )

comm/model/message.go

@@ -5,6 +5,7 @@
 package model
 
 const (
+	CMD_SEND_MESSAGE = 0
 	CMD_CHAT_TIMEOUT = 2001
 )
 
@@ -20,6 +21,9 @@ type KqMessage struct {
 }
 
 type KqCmdMessage struct {
-	Opt int64       `json:"opt"`
-	Ext interface{} `json:"ext"`
+	Opt        int64  `json:"opt"`
+	ReceiverId string `json:"receiver_id"`
+	GameId     string `json:"game_id"`
+	Uid        string `json:"uid"`
+	Ext        string `json:"ext"`
 }

comm/model/playerinfo.go

@@ -5,8 +5,8 @@
 package model
 
 type PlayerInfo struct {
-	PlayerId   string `json:"player_id"`
 	GameId     string `json:"game_id"`
+	PlayerId   string `json:"player_id"`
 	IsVip      int64  `json:"is_vip"`
 	CsId       string `json:"cs_id"`
 	ConnectTs  int64  `json:"connect_ts"`

core/cmd/rpc/cmd/cmd.go

@@ -23,6 +23,8 @@ type (
 	CsFetchPlayerQueueResp    = pb.CsFetchPlayerQueueResp
 	CsSendMsgReq              = pb.CsSendMsgReq
 	CsSendMsgResp             = pb.CsSendMsgResp
+	PlayerDisconnectReq       = pb.PlayerDisconnectReq
+	PlayerDisconnectResp      = pb.PlayerDisconnectResp
 	PlayerFetchCsInfoReq      = pb.PlayerFetchCsInfoReq
 	PlayerFetchCsInfoResp     = pb.PlayerFetchCsInfoResp
 	PlayerFetchHistoryMsgReq  = pb.PlayerFetchHistoryMsgReq
@@ -34,6 +36,7 @@ type (
 		PlayerFetchCsInfo(ctx context.Context, in *PlayerFetchCsInfoReq, opts ...grpc.CallOption) (*PlayerFetchCsInfoResp, error)
 		PlayerFetchHistoryMsg(ctx context.Context, in *PlayerFetchHistoryMsgReq, opts ...grpc.CallOption) (*PlayerFetchHistoryMsgResp, error)
 		PlayerSendMsg(ctx context.Context, in *PlayerSendMsgReq, opts ...grpc.CallOption) (*PlayerSendMsgResp, error)
+		PlayerDisconnect(ctx context.Context, in *PlayerDisconnectReq, opts ...grpc.CallOption) (*PlayerDisconnectResp, error)
 		CsFetchPlayerQueue(ctx context.Context, in *CsFetchPlayerQueueReq, opts ...grpc.CallOption) (*CsFetchPlayerQueueResp, error)
 		CsConnectPlayer(ctx context.Context, in *CsConnectPlayerReq, opts ...grpc.CallOption) (*CsConnectPlayerResp, error)
 		CsFetchHistoryChat(ctx context.Context, in *CsFetchHistoryChatReq, opts ...grpc.CallOption) (*CsFetchHistoryChatResp, error)
@@ -67,6 +70,11 @@ func (m *defaultCmd) PlayerSendMsg(ctx context.Context, in *PlayerSendMsgReq, op
 	return client.PlayerSendMsg(ctx, in, opts...)
 }
 
+func (m *defaultCmd) PlayerDisconnect(ctx context.Context, in *PlayerDisconnectReq, opts ...grpc.CallOption) (*PlayerDisconnectResp, error) {
+	client := pb.NewCmdClient(m.cli.Conn())
+	return client.PlayerDisconnect(ctx, in, opts...)
+}
+
 func (m *defaultCmd) CsFetchPlayerQueue(ctx context.Context, in *CsFetchPlayerQueueReq, opts ...grpc.CallOption) (*CsFetchPlayerQueueResp, error) {
 	client := pb.NewCmdClient(m.cli.Conn())
 	return client.CsFetchPlayerQueue(ctx, in, opts...)

core/cmd/rpc/etc/cmd.yaml

@@ -8,11 +8,11 @@ InnerRpcConf:
 
 Telemetry:
   Name: cmd-api
-  Endpoint: http://127.0.0.1:14268/api/traces
+  Endpoint: http://localhost:14268/api/traces
   Sampler: 1.0
   Batcher: jaeger
 
 KqMsgBoxProducerConf:
   Brokers:
-    - 127.0.0.1:9092
+    - localhost:9092
   Topic: send-box-topic

core/cmd/rpc/internal/logic/csconnectplayerlogic.go

@@ -27,8 +27,8 @@ func (l *CsConnectPlayerLogic) CsConnectPlayer(in *pb.CsConnectPlayerReq) (*pb.C
 	// 调用inner服务建立映射关系
 	_, err := l.svcCtx.InnerRpc.CsConnectPlayer(l.ctx, &inner.InnerCsConnectPlayerReq{
 		CsId:     in.CsId,
-		PlayerId: in.PlayerId,
 		GameId:   in.GameId,
+		PlayerId: in.PlayerId,
 	})
 	if err != nil {
 		return nil, err

core/cmd/rpc/internal/logic/playerdisconnectlogic.go

@@ -0,0 +1,32 @@
+package logic
+
+import (
+	"context"
+	"ylink/core/inner/rpc/inner"
+
+	"ylink/core/cmd/rpc/internal/svc"
+	"ylink/core/cmd/rpc/pb"
+
+	"github.com/zeromicro/go-zero/core/logx"
+)
+
+type PlayerDisconnectLogic struct {
+	ctx    context.Context
+	svcCtx *svc.ServiceContext
+	logx.Logger
+}
+
+func NewPlayerDisconnectLogic(ctx context.Context, svcCtx *svc.ServiceContext) *PlayerDisconnectLogic {
+	return &PlayerDisconnectLogic{
+		ctx:    ctx,
+		svcCtx: svcCtx,
+		Logger: logx.WithContext(ctx),
+	}
+}
+
+func (l *PlayerDisconnectLogic) PlayerDisconnect(in *pb.PlayerDisconnectReq) (*pb.PlayerDisconnectResp, error) {
+	if _, err := l.svcCtx.InnerRpc.PlayerDisconnect(l.ctx, &inner.InnerPlayerDisconnectReq{GameId: in.GameId, PlayerId: in.PlayerId}); err != nil {
+		return nil, err
+	}
+	return &pb.PlayerDisconnectResp{}, nil
+}

core/cmd/rpc/internal/server/cmdserver.go

@@ -37,6 +37,11 @@ func (s *CmdServer) PlayerSendMsg(ctx context.Context, in *pb.PlayerSendMsgReq)
 	return l.PlayerSendMsg(in)
 }
 
+func (s *CmdServer) PlayerDisconnect(ctx context.Context, in *pb.PlayerDisconnectReq) (*pb.PlayerDisconnectResp, error) {
+	l := logic.NewPlayerDisconnectLogic(ctx, s.svcCtx)
+	return l.PlayerDisconnect(in)
+}
+
 func (s *CmdServer) CsFetchPlayerQueue(ctx context.Context, in *pb.CsFetchPlayerQueueReq) (*pb.CsFetchPlayerQueueResp, error) {
 	l := logic.NewCsFetchPlayerQueueLogic(ctx, s.svcCtx)
 	return l.CsFetchPlayerQueue(in)

core/cmd/rpc/pb/cmd.proto

@@ -10,8 +10,8 @@ import "google/protobuf/struct.proto";
 Player Command Request Bean
  */
 message PlayerFetchCsInfoReq{
-  string player_id = 1;
-  string game_id = 2;
+  string game_id = 1;
+  string player_id = 2;
   string cs_id = 3;
 }
 
@@ -24,8 +24,8 @@ message PlayerFetchCsInfoResp{
 }
 
 message PlayerFetchHistoryMsgReq{
-  string player_id = 1;
-  string game_id = 2;
+  string game_id = 1;
+  string player_id = 2;
   int64 page = 3;
   int64 limit = 4;
 }
@@ -37,14 +37,21 @@ message PlayerFetchHistoryMsgResp{
 }
 
 message PlayerSendMsgReq{
-  string player_id = 1;
-  string game_id = 2;
+  string game_id = 1;
+  string player_id = 2;
   string content = 4;
   string pic = 5;
 }
 
 message PlayerSendMsgResp{}
 
+message PlayerDisconnectReq{
+  string game_id = 1;
+  string player_id = 2;
+}
+
+message PlayerDisconnectResp{}
+
 /**
 Cs Command Request Bean
  */
@@ -59,8 +66,8 @@ message CsFetchPlayerQueueResp{
 
 message CsConnectPlayerReq{
   string cs_id = 1;
-  string player_id = 2;
-  string game_id = 3;
+  string game_id = 2;
+  string player_id = 3;
 }
 
 message CsConnectPlayerResp{}
@@ -79,8 +86,8 @@ message CsFetchHistoryChatResp{
 
 message CsFetchHistoryMsgReq{
   string cs_id = 1;
-  string player_id = 2;
-  string game_id = 3;
+  string game_id = 2;
+  string player_id = 3;
   int64 page = 4;
   int64 limit = 5;
 }
@@ -94,8 +101,8 @@ message CsFetchHistoryMsgResp{
 
 message CsSendMsgReq{
   string cs_id = 1;
-  string player_id = 2;
-  string game_id = 3;
+  string game_id = 2;
+  string player_id = 3;
   string content = 4;
   string pic = 5;
 }
@@ -106,6 +113,7 @@ service Cmd {
   rpc playerFetchCsInfo (PlayerFetchCsInfoReq) returns (PlayerFetchCsInfoResp);
   rpc playerFetchHistoryMsg (PlayerFetchHistoryMsgReq) returns (PlayerFetchHistoryMsgResp);
   rpc playerSendMsg (PlayerSendMsgReq) returns (PlayerSendMsgResp);
+  rpc playerDisconnect (PlayerDisconnectReq) returns (PlayerDisconnectResp);
 
   rpc csFetchPlayerQueue (CsFetchPlayerQueueReq) returns (CsFetchPlayerQueueResp);
   rpc csConnectPlayer (CsConnectPlayerReq) returns (CsConnectPlayerResp);

core/cmd/rpc/pb/cmd_grpc.pb.go

@@ -25,6 +25,7 @@ type CmdClient interface {
 	PlayerFetchCsInfo(ctx context.Context, in *PlayerFetchCsInfoReq, opts ...grpc.CallOption) (*PlayerFetchCsInfoResp, error)
 	PlayerFetchHistoryMsg(ctx context.Context, in *PlayerFetchHistoryMsgReq, opts ...grpc.CallOption) (*PlayerFetchHistoryMsgResp, error)
 	PlayerSendMsg(ctx context.Context, in *PlayerSendMsgReq, opts ...grpc.CallOption) (*PlayerSendMsgResp, error)
+	PlayerDisconnect(ctx context.Context, in *PlayerDisconnectReq, opts ...grpc.CallOption) (*PlayerDisconnectResp, error)
 	CsFetchPlayerQueue(ctx context.Context, in *CsFetchPlayerQueueReq, opts ...grpc.CallOption) (*CsFetchPlayerQueueResp, error)
 	CsConnectPlayer(ctx context.Context, in *CsConnectPlayerReq, opts ...grpc.CallOption) (*CsConnectPlayerResp, error)
 	CsFetchHistoryChat(ctx context.Context, in *CsFetchHistoryChatReq, opts ...grpc.CallOption) (*CsFetchHistoryChatResp, error)
@@ -67,6 +68,15 @@ func (c *cmdClient) PlayerSendMsg(ctx context.Context, in *PlayerSendMsgReq, opt
 	return out, nil
 }
 
+func (c *cmdClient) PlayerDisconnect(ctx context.Context, in *PlayerDisconnectReq, opts ...grpc.CallOption) (*PlayerDisconnectResp, error) {
+	out := new(PlayerDisconnectResp)
+	err := c.cc.Invoke(ctx, "/pb.Cmd/playerDisconnect", in, out, opts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
 func (c *cmdClient) CsFetchPlayerQueue(ctx context.Context, in *CsFetchPlayerQueueReq, opts ...grpc.CallOption) (*CsFetchPlayerQueueResp, error) {
 	out := new(CsFetchPlayerQueueResp)
 	err := c.cc.Invoke(ctx, "/pb.Cmd/csFetchPlayerQueue", in, out, opts...)
@@ -119,6 +129,7 @@ type CmdServer interface {
 	PlayerFetchCsInfo(context.Context, *PlayerFetchCsInfoReq) (*PlayerFetchCsInfoResp, error)
 	PlayerFetchHistoryMsg(context.Context, *PlayerFetchHistoryMsgReq) (*PlayerFetchHistoryMsgResp, error)
 	PlayerSendMsg(context.Context, *PlayerSendMsgReq) (*PlayerSendMsgResp, error)
+	PlayerDisconnect(context.Context, *PlayerDisconnectReq) (*PlayerDisconnectResp, error)
 	CsFetchPlayerQueue(context.Context, *CsFetchPlayerQueueReq) (*CsFetchPlayerQueueResp, error)
 	CsConnectPlayer(context.Context, *CsConnectPlayerReq) (*CsConnectPlayerResp, error)
 	CsFetchHistoryChat(context.Context, *CsFetchHistoryChatReq) (*CsFetchHistoryChatResp, error)
@@ -140,6 +151,9 @@ func (UnimplementedCmdServer) PlayerFetchHistoryMsg(context.Context, *PlayerFetc
 func (UnimplementedCmdServer) PlayerSendMsg(context.Context, *PlayerSendMsgReq) (*PlayerSendMsgResp, error) {
 	return nil, status.Errorf(codes.Unimplemented, "method PlayerSendMsg not implemented")
 }
+func (UnimplementedCmdServer) PlayerDisconnect(context.Context, *PlayerDisconnectReq) (*PlayerDisconnectResp, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method PlayerDisconnect not implemented")
+}
 func (UnimplementedCmdServer) CsFetchPlayerQueue(context.Context, *CsFetchPlayerQueueReq) (*CsFetchPlayerQueueResp, error) {
 	return nil, status.Errorf(codes.Unimplemented, "method CsFetchPlayerQueue not implemented")
 }
@@ -222,6 +236,24 @@ func _Cmd_PlayerSendMsg_Handler(srv interface{}, ctx context.Context, dec func(i
 	return interceptor(ctx, in, info, handler)
 }
 
+func _Cmd_PlayerDisconnect_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(PlayerDisconnectReq)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(CmdServer).PlayerDisconnect(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: "/pb.Cmd/playerDisconnect",
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(CmdServer).PlayerDisconnect(ctx, req.(*PlayerDisconnectReq))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
 func _Cmd_CsFetchPlayerQueue_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
 	in := new(CsFetchPlayerQueueReq)
 	if err := dec(in); err != nil {
@@ -331,6 +363,10 @@ var Cmd_ServiceDesc = grpc.ServiceDesc{
 			MethodName: "playerSendMsg",
 			Handler:    _Cmd_PlayerSendMsg_Handler,
 		},
+		{
+			MethodName: "playerDisconnect",
+			Handler:    _Cmd_PlayerDisconnect_Handler,
+		},
 		{
 			MethodName: "csFetchPlayerQueue",
 			Handler:    _Cmd_CsFetchPlayerQueue_Handler,

core/inner/rpc/etc/inner.yaml

@@ -3,29 +3,29 @@ ListenOn: 0.0.0.0:10500
 
 Telemetry:
   Name: inner-rpc
-  Endpoint: http://127.0.0.1:14268/api/traces
+  Endpoint: http://localhost:14268/api/traces
   Sampler: 1.0
   Batcher: jaeger
 
 KqMsgBoxConsumerConf:
   Brokers:
-    - 127.0.0.1:9092
+    - localhost:9092
   Topic: send-box-topic
   GroupId: inner-rpc
 
 KqMsgBoxProducerConf:
   Brokers:
-    - 127.0.0.1:9092
+    - localhost:9092
   Topic: recv-box-topic
 
 KqDbBoxProducerConf:
   Brokers:
-    - 127.0.0.1:9092
+    - localhost:9092
   Topic: db-box-topic
 
 KqCmdBoxProducerConf:
   Brokers:
-    - 127.0.0.1:9092
+    - localhost:9092
   Topic: cmd-box-topic
 
 MockFolder: /Users/suyghur/Develop/yyxx/backend/ylink/core/inner/rpc/mock

core/inner/rpc/inner/inner.go

@@ -17,6 +17,8 @@ type (
 	InnerCsConnectPlayerResp    = pb.InnerCsConnectPlayerResp
 	InnerCsFetchPlayerQueueReq  = pb.InnerCsFetchPlayerQueueReq
 	InnerCsFetchPlayerQueueResp = pb.InnerCsFetchPlayerQueueResp
+	InnerPlayerDisconnectReq    = pb.InnerPlayerDisconnectReq
+	InnerPlayerDisconnectResp   = pb.InnerPlayerDisconnectResp
 	InnerPlayerFetchCsInfoReq   = pb.InnerPlayerFetchCsInfoReq
 	InnerPlayerFetchCsInfoResp  = pb.InnerPlayerFetchCsInfoResp
 	NotifyUserStatusReq         = pb.NotifyUserStatusReq
@@ -24,6 +26,7 @@ type (
 
 	Inner interface {
 		PlayerFetchCsInfo(ctx context.Context, in *InnerPlayerFetchCsInfoReq, opts ...grpc.CallOption) (*InnerPlayerFetchCsInfoResp, error)
+		PlayerDisconnect(ctx context.Context, in *InnerPlayerDisconnectReq, opts ...grpc.CallOption) (*InnerPlayerDisconnectResp, error)
 		CsFetchPlayerQueue(ctx context.Context, in *InnerCsFetchPlayerQueueReq, opts ...grpc.CallOption) (*InnerCsFetchPlayerQueueResp, error)
 		CsConnectPlayer(ctx context.Context, in *InnerCsConnectPlayerReq, opts ...grpc.CallOption) (*InnerCsConnectPlayerResp, error)
 		NotifyUserOnline(ctx context.Context, in *NotifyUserStatusReq, opts ...grpc.CallOption) (*NotifyUserStatusResp, error)
@@ -46,6 +49,11 @@ func (m *defaultInner) PlayerFetchCsInfo(ctx context.Context, in *InnerPlayerFet
 	return client.PlayerFetchCsInfo(ctx, in, opts...)
 }
 
+func (m *defaultInner) PlayerDisconnect(ctx context.Context, in *InnerPlayerDisconnectReq, opts ...grpc.CallOption) (*InnerPlayerDisconnectResp, error) {
+	client := pb.NewInnerClient(m.cli.Conn())
+	return client.PlayerDisconnect(ctx, in, opts...)
+}
+
 func (m *defaultInner) CsFetchPlayerQueue(ctx context.Context, in *InnerCsFetchPlayerQueueReq, opts ...grpc.CallOption) (*InnerCsFetchPlayerQueueResp, error) {
 	client := pb.NewInnerClient(m.cli.Conn())
 	return client.CsFetchPlayerQueue(ctx, in, opts...)

core/inner/rpc/internal/ext/buskey.go

@@ -1,7 +0,0 @@
-//@File     buskey.go
-//@Time     2022/06/07
-//@Author   #Suyghur,
-
-package ext
-
-const EVENT_REMOVE_TIMEOUT_JOB = "EVENT_REMOVE_TIMEOUT_JOB"

core/inner/rpc/internal/logic/csconnectplayerlogic.go

@@ -5,10 +5,13 @@ import (
 	"github.com/bytedance/sonic"
 	"github.com/gookit/event"
 	treemap "github.com/liyue201/gostl/ds/map"
+	"github.com/pkg/errors"
 	"github.com/robfig/cron/v3"
 	"github.com/zeromicro/go-zero/core/logx"
 	"time"
+	"ylink/comm/globalkey"
 	"ylink/comm/model"
+	"ylink/comm/result"
 	"ylink/core/inner/rpc/internal/ext"
 	"ylink/core/inner/rpc/internal/svc"
 	"ylink/core/inner/rpc/pb"
@@ -31,6 +34,10 @@ func NewCsConnectPlayerLogic(ctx context.Context, svcCtx *svc.ServiceContext) *C
 func (l *CsConnectPlayerLogic) CsConnectPlayer(in *pb.InnerCsConnectPlayerReq) (*pb.InnerCsConnectPlayerResp, error) {
 
 	playerInfo := ext.GetOnlinePlayerInfo(in.GameId, in.PlayerId)
+
+	if playerInfo == nil {
+		return nil, errors.Wrapf(result.NewErrMsg("The player is not connected"), "")
+	}
 	playerInfo.CsId = in.CsId
 	playerInfo.DequeueTs = time.Now().Unix()
 
@@ -63,16 +70,18 @@ func (l *CsConnectPlayerLogic) CsConnectPlayer(in *pb.InnerCsConnectPlayerReq) (
 			timeoutTs = time.Now().Unix() - playerInfo.LastChatTs
 		}
 		if timeoutTs >= 300 {
-			_ = event.MustFire(ext.EVENT_REMOVE_TIMEOUT_JOB, event.M{"entry_id": entryId})
+			// 释放计时器任务
+			_ = event.MustFire(globalkey.EventRemoveTimeoutJob, event.M{"entry_id": entryId})
 			l.Logger.Infof("trigger timeout event, remove cron job, entry id: %d", entryId)
 
-			// 发下线command
-			//ext, _ := sonic.Marshal(playerInfo)
+			// 发踢下线的command指令
 			message, _ := sonic.MarshalString(&model.KqCmdMessage{
-				Opt: model.CMD_CHAT_TIMEOUT,
-				Ext: playerInfo,
+				Opt:        model.CMD_CHAT_TIMEOUT,
+				ReceiverId: in.GameId + "_" + in.PlayerId,
+				GameId:     in.GameId,
+				Uid:        in.PlayerId,
 			})
-			l.svcCtx.KqCmdBoxProducer.SendMessage(l.ctx, message, in.PlayerId)
+			l.svcCtx.KqCmdBoxProducer.SendMessage(l.ctx, message, in.GameId+"_"+in.PlayerId)
 		}
 	})
 

core/inner/rpc/internal/logic/csfetchplayerqueuelogic.go

@@ -47,8 +47,8 @@ func (l *CsFetchPlayerQueueLogic) CsFetchPlayerQueue(in *pb.InnerCsFetchPlayerQu
 	for node := ext.WaitingList.FrontNode(); node != nil && index < queueLen; node = node.Next() {
 		info := node.Value.(*model.PlayerInfo)
 		queue[index] = map[string]interface{}{
-			"player_id": info.PlayerId,
 			"game_id":   info.GameId,
+			"player_id": info.PlayerId,
 			"wait_time": time.Now().Unix() - info.EnqueueTs,
 		}
 		index += 1

core/inner/rpc/internal/logic/notifyuserofflinelogic.go

@@ -31,7 +31,7 @@ func NewNotifyUserOfflineLogic(ctx context.Context, svcCtx *svc.ServiceContext)
 
 func (l *NotifyUserOfflineLogic) NotifyUserOffline(in *pb.NotifyUserStatusReq) (*pb.NotifyUserStatusResp, error) {
 	switch in.Type {
-	case globalkey.CONNECT_TYPE_PLAYER:
+	case globalkey.ConnectTypePlayer:
 		// 修改玩家在线状态
 		if ext.GameOnlinePlayerMap.Contains(in.GameId) {
 			// 有则取出玩家
@@ -50,7 +50,7 @@ func (l *NotifyUserOfflineLogic) NotifyUserOffline(in *pb.NotifyUserStatusReq) (
 				break
 			}
 		}
-	case globalkey.CONNECT_TYPE_CS:
+	case globalkey.ConnectTypeCs:
 		// 修改客服在线状态
 		if csInfo := ext.GetCsInfo(in.Uid); csInfo != nil {
 			csInfo.OnlineStatus = 0

core/inner/rpc/internal/logic/notifyuseronlinelogic.go

@@ -32,7 +32,7 @@ func NewNotifyUserOnlineLogic(ctx context.Context, svcCtx *svc.ServiceContext) *
 
 func (l *NotifyUserOnlineLogic) NotifyUserOnline(in *pb.NotifyUserStatusReq) (*pb.NotifyUserStatusResp, error) {
 	switch in.Type {
-	case globalkey.CONNECT_TYPE_PLAYER:
+	case globalkey.ConnectTypePlayer:
 		// 修改玩家在线状态
 		if ext.GameOnlinePlayerMap.Contains(in.GameId) {
 			// 有则取出玩家的map
@@ -48,8 +48,8 @@ func (l *NotifyUserOnlineLogic) NotifyUserOnline(in *pb.NotifyUserStatusReq) (*p
 					// 不是vip
 					ts := time.Now().Unix()
 					playerInfo := model.PlayerInfo{
-						PlayerId:  in.Uid,
 						GameId:    in.GameId,
+						PlayerId:  in.Uid,
 						ConnectTs: ts,
 						EnqueueTs: ts,
 					}
@@ -59,7 +59,6 @@ func (l *NotifyUserOnlineLogic) NotifyUserOnline(in *pb.NotifyUserStatusReq) (*p
 					l.Logger.Infof("enqueue waiting list: %s", ext.WaitingList.String())
 				}
 			}
-			l.Logger.Infof("111111")
 		} else {
 			onlinePlayerMap := treemap.New(treemap.WithGoroutineSafe())
 			// 判断是不是vip玩家
@@ -70,8 +69,8 @@ func (l *NotifyUserOnlineLogic) NotifyUserOnline(in *pb.NotifyUserStatusReq) (*p
 				// 不是vip
 				ts := time.Now().Unix()
 				playerInfo := model.PlayerInfo{
-					PlayerId:  in.Uid,
 					GameId:    in.GameId,
+					PlayerId:  in.Uid,
 					ConnectTs: ts,
 					EnqueueTs: ts,
 				}
@@ -81,9 +80,8 @@ func (l *NotifyUserOnlineLogic) NotifyUserOnline(in *pb.NotifyUserStatusReq) (*p
 				l.Logger.Infof("enqueue waiting list: %s", ext.WaitingList.String())
 			}
 			ext.GameOnlinePlayerMap.Insert(in.GameId, onlinePlayerMap)
-			l.Logger.Infof("22222")
 		}
-	case globalkey.CONNECT_TYPE_CS:
+	case globalkey.ConnectTypeCs:
 		if csInfo := ext.GetCsInfo(in.Uid); csInfo != nil {
 			csInfo.OnlineStatus = 1
 		} else {

core/inner/rpc/internal/logic/playerdisconnectlogic.go

@@ -0,0 +1,30 @@
+package logic
+
+import (
+	"context"
+
+	"ylink/core/inner/rpc/internal/svc"
+	"ylink/core/inner/rpc/pb"
+
+	"github.com/zeromicro/go-zero/core/logx"
+)
+
+type PlayerDisconnectLogic struct {
+	ctx    context.Context
+	svcCtx *svc.ServiceContext
+	logx.Logger
+}
+
+func NewPlayerDisconnectLogic(ctx context.Context, svcCtx *svc.ServiceContext) *PlayerDisconnectLogic {
+	return &PlayerDisconnectLogic{
+		ctx:    ctx,
+		svcCtx: svcCtx,
+		Logger: logx.WithContext(ctx),
+	}
+}
+
+func (l *PlayerDisconnectLogic) PlayerDisconnect(in *pb.InnerPlayerFetchCsInfoReq) (*pb.InnerPlayerFetchCsInfoResp, error) {
+	// todo: add your logic here and delete this line
+
+	return &pb.InnerPlayerFetchCsInfoResp{}, nil
+}

core/inner/rpc/internal/logic/playerfetchcsinfologic.go

@@ -2,9 +2,7 @@ package logic
 
 import (
 	"context"
-	treemap "github.com/liyue201/gostl/ds/map"
 	"github.com/pkg/errors"
-	"ylink/comm/model"
 	"ylink/comm/result"
 	"ylink/core/inner/rpc/internal/ext"
 	"ylink/core/inner/rpc/internal/svc"
@@ -28,11 +26,9 @@ func NewPlayerFetchCsInfoLogic(ctx context.Context, svcCtx *svc.ServiceContext)
 }
 
 func (l *PlayerFetchCsInfoLogic) PlayerFetchCsInfo(in *pb.InnerPlayerFetchCsInfoReq) (*pb.InnerPlayerFetchCsInfoResp, error) {
-	if ext.GameConnectedMap.Contains(in.GameId) {
-		playerConnMap := ext.GameConnectedMap.Get(in.GameId).(*treemap.Map)
-		csId := playerConnMap.Get(in.PlayerId).(string)
-		if ext.CsInfoMap.Contains(csId) {
-			csInfo := ext.CsInfoMap.Get(csId).(model.CsInfo)
+	if playerInfo := ext.GetConnectedPlayerInfo(in.GameId, in.PlayerId); playerInfo != nil {
+		// 玩家已连接
+		if csInfo := ext.GetCsInfo(playerInfo.CsId); csInfo != nil {
 			return &pb.InnerPlayerFetchCsInfoResp{
 				CsId:         csInfo.CsId,
 				CsNickname:   csInfo.CsNickname,
@@ -41,6 +37,8 @@ func (l *PlayerFetchCsInfoLogic) PlayerFetchCsInfo(in *pb.InnerPlayerFetchCsInfo
 				OnlineStatus: csInfo.OnlineStatus,
 			}, nil
 		}
+		return nil, errors.Wrap(result.NewErrMsg("Customer service information does not exist"), "")
 	}
-	return nil, errors.Wrap(result.NewErrMsg("Customer service information does not exist"), "")
+	return nil, errors.Wrap(result.NewErrMsg("The player is not connected"), "")
+
 }

core/inner/rpc/internal/server/innerserver.go

@@ -27,6 +27,11 @@ func (s *InnerServer) PlayerFetchCsInfo(ctx context.Context, in *pb.InnerPlayerF
 	return l.PlayerFetchCsInfo(in)
 }
 
+func (s *InnerServer) PlayerDisconnect(ctx context.Context, in *pb.InnerPlayerDisconnectReq) (*pb.InnerPlayerDisconnectResp, error) {
+	l := logic.NewPlayerDisconnectLogic(ctx, s.svcCtx)
+	return l.PlayerDisconnect(in)
+}
+
 func (s *InnerServer) CsFetchPlayerQueue(ctx context.Context, in *pb.InnerCsFetchPlayerQueueReq) (*pb.InnerCsFetchPlayerQueueResp, error) {
 	l := logic.NewCsFetchPlayerQueueLogic(ctx, s.svcCtx)
 	return l.CsFetchPlayerQueue(in)

core/inner/rpc/internal/svc/servicecontext.go

@@ -11,6 +11,7 @@ import (
 	"github.com/zeromicro/go-zero/core/logx"
 	"go.opentelemetry.io/otel/attribute"
 	"io/ioutil"
+	"ylink/comm/globalkey"
 	"ylink/comm/kafka"
 	"ylink/comm/model"
 	"ylink/comm/trace"
@@ -109,9 +110,10 @@ func (s *ServiceContext) subscribe() {
 	go s.KqMsgConsumerGroup.RegisterHandleAndConsumer(s)
 
 	// 注册事件
-	event.On(ext.EVENT_REMOVE_TIMEOUT_JOB, event.ListenerFunc(func(e event.Event) error {
+	event.On(globalkey.EventRemoveTimeoutJob, event.ListenerFunc(func(e event.Event) error {
 		logx.Info("on event remove timeout job...")
 		entryId := e.Get("entry_id").(cron.EntryID)
+
 		s.TimeoutCron.Remove(entryId)
 		return nil
 	}), event.High)

core/inner/rpc/pb/inner.pb.go

@@ -28,8 +28,8 @@ type InnerPlayerFetchCsInfoReq struct {
 	sizeCache     protoimpl.SizeCache
 	unknownFields protoimpl.UnknownFields
 
-	PlayerId string `protobuf:"bytes,1,opt,name=player_id,json=playerId,proto3" json:"player_id,omitempty"`
-	GameId   string `protobuf:"bytes,2,opt,name=game_id,json=gameId,proto3" json:"game_id,omitempty"`
+	GameId   string `protobuf:"bytes,1,opt,name=game_id,json=gameId,proto3" json:"game_id,omitempty"`
+	PlayerId string `protobuf:"bytes,2,opt,name=player_id,json=playerId,proto3" json:"player_id,omitempty"`
 	CsId     string `protobuf:"bytes,3,opt,name=cs_id,json=csId,proto3" json:"cs_id,omitempty"`
 }
 
@@ -65,16 +65,16 @@ func (*InnerPlayerFetchCsInfoReq) Descriptor() ([]byte, []int) {
 	return file_pb_inner_proto_rawDescGZIP(), []int{0}
 }
 
-func (x *InnerPlayerFetchCsInfoReq) GetPlayerId() string {
+func (x *InnerPlayerFetchCsInfoReq) GetGameId() string {
 	if x != nil {
-		return x.PlayerId
+		return x.GameId
 	}
 	return ""
 }
 
-func (x *InnerPlayerFetchCsInfoReq) GetGameId() string {
+func (x *InnerPlayerFetchCsInfoReq) GetPlayerId() string {
 	if x != nil {
-		return x.GameId
+		return x.PlayerId
 	}
 	return ""
 }
@@ -165,6 +165,99 @@ func (x *InnerPlayerFetchCsInfoResp) GetOnlineStatus() int64 {
 	return 0
 }
 
+type InnerPlayerDisconnectReq struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	PlayerId string `protobuf:"bytes,1,opt,name=player_id,json=playerId,proto3" json:"player_id,omitempty"`
+	GameId   string `protobuf:"bytes,2,opt,name=game_id,json=gameId,proto3" json:"game_id,omitempty"`
+}
+
+func (x *InnerPlayerDisconnectReq) Reset() {
+	*x = InnerPlayerDisconnectReq{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_pb_inner_proto_msgTypes[2]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *InnerPlayerDisconnectReq) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*InnerPlayerDisconnectReq) ProtoMessage() {}
+
+func (x *InnerPlayerDisconnectReq) ProtoReflect() protoreflect.Message {
+	mi := &file_pb_inner_proto_msgTypes[2]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use InnerPlayerDisconnectReq.ProtoReflect.Descriptor instead.
+func (*InnerPlayerDisconnectReq) Descriptor() ([]byte, []int) {
+	return file_pb_inner_proto_rawDescGZIP(), []int{2}
+}
+
+func (x *InnerPlayerDisconnectReq) GetPlayerId() string {
+	if x != nil {
+		return x.PlayerId
+	}
+	return ""
+}
+
+func (x *InnerPlayerDisconnectReq) GetGameId() string {
+	if x != nil {
+		return x.GameId
+	}
+	return ""
+}
+
+type InnerPlayerDisconnectResp struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+}
+
+func (x *InnerPlayerDisconnectResp) Reset() {
+	*x = InnerPlayerDisconnectResp{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_pb_inner_proto_msgTypes[3]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *InnerPlayerDisconnectResp) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*InnerPlayerDisconnectResp) ProtoMessage() {}
+
+func (x *InnerPlayerDisconnectResp) ProtoReflect() protoreflect.Message {
+	mi := &file_pb_inner_proto_msgTypes[3]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use InnerPlayerDisconnectResp.ProtoReflect.Descriptor instead.
+func (*InnerPlayerDisconnectResp) Descriptor() ([]byte, []int) {
+	return file_pb_inner_proto_rawDescGZIP(), []int{3}
+}
+
 //*
 //Cs Command Request Bean
 type InnerCsFetchPlayerQueueReq struct {
@@ -178,7 +271,7 @@ type InnerCsFetchPlayerQueueReq struct {
 func (x *InnerCsFetchPlayerQueueReq) Reset() {
 	*x = InnerCsFetchPlayerQueueReq{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_pb_inner_proto_msgTypes[2]
+		mi := &file_pb_inner_proto_msgTypes[4]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -191,7 +284,7 @@ func (x *InnerCsFetchPlayerQueueReq) String() string {
 func (*InnerCsFetchPlayerQueueReq) ProtoMessage() {}
 
 func (x *InnerCsFetchPlayerQueueReq) ProtoReflect() protoreflect.Message {
-	mi := &file_pb_inner_proto_msgTypes[2]
+	mi := &file_pb_inner_proto_msgTypes[4]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -204,7 +297,7 @@ func (x *InnerCsFetchPlayerQueueReq) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use InnerCsFetchPlayerQueueReq.ProtoReflect.Descriptor instead.
 func (*InnerCsFetchPlayerQueueReq) Descriptor() ([]byte, []int) {
-	return file_pb_inner_proto_rawDescGZIP(), []int{2}
+	return file_pb_inner_proto_rawDescGZIP(), []int{4}
 }
 
 func (x *InnerCsFetchPlayerQueueReq) GetLimit() int64 {
@@ -226,7 +319,7 @@ type InnerCsFetchPlayerQueueResp struct {
 func (x *InnerCsFetchPlayerQueueResp) Reset() {
 	*x = InnerCsFetchPlayerQueueResp{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_pb_inner_proto_msgTypes[3]
+		mi := &file_pb_inner_proto_msgTypes[5]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -239,7 +332,7 @@ func (x *InnerCsFetchPlayerQueueResp) String() string {
 func (*InnerCsFetchPlayerQueueResp) ProtoMessage() {}
 
 func (x *InnerCsFetchPlayerQueueResp) ProtoReflect() protoreflect.Message {
-	mi := &file_pb_inner_proto_msgTypes[3]
+	mi := &file_pb_inner_proto_msgTypes[5]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -252,7 +345,7 @@ func (x *InnerCsFetchPlayerQueueResp) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use InnerCsFetchPlayerQueueResp.ProtoReflect.Descriptor instead.
 func (*InnerCsFetchPlayerQueueResp) Descriptor() ([]byte, []int) {
-	return file_pb_inner_proto_rawDescGZIP(), []int{3}
+	return file_pb_inner_proto_rawDescGZIP(), []int{5}
 }
 
 func (x *InnerCsFetchPlayerQueueResp) GetTotal() int64 {
@@ -275,14 +368,14 @@ type InnerCsConnectPlayerReq struct {
 	unknownFields protoimpl.UnknownFields
 
 	CsId     string `protobuf:"bytes,1,opt,name=cs_id,json=csId,proto3" json:"cs_id,omitempty"`
-	PlayerId string `protobuf:"bytes,2,opt,name=player_id,json=playerId,proto3" json:"player_id,omitempty"`
-	GameId   string `protobuf:"bytes,3,opt,name=game_id,json=gameId,proto3" json:"game_id,omitempty"`
+	GameId   string `protobuf:"bytes,2,opt,name=game_id,json=gameId,proto3" json:"game_id,omitempty"`
+	PlayerId string `protobuf:"bytes,3,opt,name=player_id,json=playerId,proto3" json:"player_id,omitempty"`
 }
 
 func (x *InnerCsConnectPlayerReq) Reset() {
 	*x = InnerCsConnectPlayerReq{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_pb_inner_proto_msgTypes[4]
+		mi := &file_pb_inner_proto_msgTypes[6]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -295,7 +388,7 @@ func (x *InnerCsConnectPlayerReq) String() string {
 func (*InnerCsConnectPlayerReq) ProtoMessage() {}
 
 func (x *InnerCsConnectPlayerReq) ProtoReflect() protoreflect.Message {
-	mi := &file_pb_inner_proto_msgTypes[4]
+	mi := &file_pb_inner_proto_msgTypes[6]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -308,7 +401,7 @@ func (x *InnerCsConnectPlayerReq) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use InnerCsConnectPlayerReq.ProtoReflect.Descriptor instead.
 func (*InnerCsConnectPlayerReq) Descriptor() ([]byte, []int) {
-	return file_pb_inner_proto_rawDescGZIP(), []int{4}
+	return file_pb_inner_proto_rawDescGZIP(), []int{6}
 }
 
 func (x *InnerCsConnectPlayerReq) GetCsId() string {
@@ -318,16 +411,16 @@ func (x *InnerCsConnectPlayerReq) GetCsId() string {
 	return ""
 }
 
-func (x *InnerCsConnectPlayerReq) GetPlayerId() string {
+func (x *InnerCsConnectPlayerReq) GetGameId() string {
 	if x != nil {
-		return x.PlayerId
+		return x.GameId
 	}
 	return ""
 }
 
-func (x *InnerCsConnectPlayerReq) GetGameId() string {
+func (x *InnerCsConnectPlayerReq) GetPlayerId() string {
 	if x != nil {
-		return x.GameId
+		return x.PlayerId
 	}
 	return ""
 }
@@ -341,7 +434,7 @@ type InnerCsConnectPlayerResp struct {
 func (x *InnerCsConnectPlayerResp) Reset() {
 	*x = InnerCsConnectPlayerResp{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_pb_inner_proto_msgTypes[5]
+		mi := &file_pb_inner_proto_msgTypes[7]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -354,7 +447,7 @@ func (x *InnerCsConnectPlayerResp) String() string {
 func (*InnerCsConnectPlayerResp) ProtoMessage() {}
 
 func (x *InnerCsConnectPlayerResp) ProtoReflect() protoreflect.Message {
-	mi := &file_pb_inner_proto_msgTypes[5]
+	mi := &file_pb_inner_proto_msgTypes[7]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -367,7 +460,7 @@ func (x *InnerCsConnectPlayerResp) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use InnerCsConnectPlayerResp.ProtoReflect.Descriptor instead.
 func (*InnerCsConnectPlayerResp) Descriptor() ([]byte, []int) {
-	return file_pb_inner_proto_rawDescGZIP(), []int{5}
+	return file_pb_inner_proto_rawDescGZIP(), []int{7}
 }
 
 type NotifyUserStatusReq struct {
@@ -383,7 +476,7 @@ type NotifyUserStatusReq struct {
 func (x *NotifyUserStatusReq) Reset() {
 	*x = NotifyUserStatusReq{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_pb_inner_proto_msgTypes[6]
+		mi := &file_pb_inner_proto_msgTypes[8]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -396,7 +489,7 @@ func (x *NotifyUserStatusReq) String() string {
 func (*NotifyUserStatusReq) ProtoMessage() {}
 
 func (x *NotifyUserStatusReq) ProtoReflect() protoreflect.Message {
-	mi := &file_pb_inner_proto_msgTypes[6]
+	mi := &file_pb_inner_proto_msgTypes[8]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -409,7 +502,7 @@ func (x *NotifyUserStatusReq) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use NotifyUserStatusReq.ProtoReflect.Descriptor instead.
 func (*NotifyUserStatusReq) Descriptor() ([]byte, []int) {
-	return file_pb_inner_proto_rawDescGZIP(), []int{6}
+	return file_pb_inner_proto_rawDescGZIP(), []int{8}
 }
 
 func (x *NotifyUserStatusReq) GetType() int64 {
@@ -442,7 +535,7 @@ type NotifyUserStatusResp struct {
 func (x *NotifyUserStatusResp) Reset() {
 	*x = NotifyUserStatusResp{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_pb_inner_proto_msgTypes[7]
+		mi := &file_pb_inner_proto_msgTypes[9]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -455,7 +548,7 @@ func (x *NotifyUserStatusResp) String() string {
 func (*NotifyUserStatusResp) ProtoMessage() {}
 
 func (x *NotifyUserStatusResp) ProtoReflect() protoreflect.Message {
-	mi := &file_pb_inner_proto_msgTypes[7]
+	mi := &file_pb_inner_proto_msgTypes[9]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -468,7 +561,7 @@ func (x *NotifyUserStatusResp) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use NotifyUserStatusResp.ProtoReflect.Descriptor instead.
 func (*NotifyUserStatusResp) Descriptor() ([]byte, []int) {
-	return file_pb_inner_proto_rawDescGZIP(), []int{7}
+	return file_pb_inner_proto_rawDescGZIP(), []int{9}
 }
 
 var File_pb_inner_proto protoreflect.FileDescriptor
@@ -479,10 +572,10 @@ var file_pb_inner_proto_rawDesc = []byte{
 	0x74, 0x6f, 0x62, 0x75, 0x66, 0x2f, 0x73, 0x74, 0x72, 0x75, 0x63, 0x74, 0x2e, 0x70, 0x72, 0x6f,
 	0x74, 0x6f, 0x22, 0x66, 0x0a, 0x19, 0x49, 0x6e, 0x6e, 0x65, 0x72, 0x50, 0x6c, 0x61, 0x79, 0x65,
 	0x72, 0x46, 0x65, 0x74, 0x63, 0x68, 0x43, 0x73, 0x49, 0x6e, 0x66, 0x6f, 0x52, 0x65, 0x71, 0x12,
-	0x1b, 0x0a, 0x09, 0x70, 0x6c, 0x61, 0x79, 0x65, 0x72, 0x5f, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01,
-	0x28, 0x09, 0x52, 0x08, 0x70, 0x6c, 0x61, 0x79, 0x65, 0x72, 0x49, 0x64, 0x12, 0x17, 0x0a, 0x07,
-	0x67, 0x61, 0x6d, 0x65, 0x5f, 0x69, 0x64, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x06, 0x67,
-	0x61, 0x6d, 0x65, 0x49, 0x64, 0x12, 0x13, 0x0a, 0x05, 0x63, 0x73, 0x5f, 0x69, 0x64, 0x18, 0x03,
+	0x17, 0x0a, 0x07, 0x67, 0x61, 0x6d, 0x65, 0x5f, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09,
+	0x52, 0x06, 0x67, 0x61, 0x6d, 0x65, 0x49, 0x64, 0x12, 0x1b, 0x0a, 0x09, 0x70, 0x6c, 0x61, 0x79,
+	0x65, 0x72, 0x5f, 0x69, 0x64, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x70, 0x6c, 0x61,
+	0x79, 0x65, 0x72, 0x49, 0x64, 0x12, 0x13, 0x0a, 0x05, 0x63, 0x73, 0x5f, 0x69, 0x64, 0x18, 0x03,
 	0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x63, 0x73, 0x49, 0x64, 0x22, 0xbe, 0x01, 0x0a, 0x1a, 0x49,
 	0x6e, 0x6e, 0x65, 0x72, 0x50, 0x6c, 0x61, 0x79, 0x65, 0x72, 0x46, 0x65, 0x74, 0x63, 0x68, 0x43,
 	0x73, 0x49, 0x6e, 0x66, 0x6f, 0x52, 0x65, 0x73, 0x70, 0x12, 0x13, 0x0a, 0x05, 0x63, 0x73, 0x5f,
@@ -495,37 +588,49 @@ var file_pb_inner_proto_rawDesc = []byte{
 	0x75, 0x72, 0x65, 0x18, 0x04, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0b, 0x63, 0x73, 0x53, 0x69, 0x67,
 	0x6e, 0x61, 0x74, 0x75, 0x72, 0x65, 0x12, 0x23, 0x0a, 0x0d, 0x6f, 0x6e, 0x6c, 0x69, 0x6e, 0x65,
 	0x5f, 0x73, 0x74, 0x61, 0x74, 0x75, 0x73, 0x18, 0x05, 0x20, 0x01, 0x28, 0x03, 0x52, 0x0c, 0x6f,
-	0x6e, 0x6c, 0x69, 0x6e, 0x65, 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x22, 0x32, 0x0a, 0x1a, 0x49,
-	0x6e, 0x6e, 0x65, 0x72, 0x43, 0x73, 0x46, 0x65, 0x74, 0x63, 0x68, 0x50, 0x6c, 0x61, 0x79, 0x65,
-	0x72, 0x51, 0x75, 0x65, 0x75, 0x65, 0x52, 0x65, 0x71, 0x12, 0x14, 0x0a, 0x05, 0x6c, 0x69, 0x6d,
-	0x69, 0x74, 0x18, 0x01, 0x20, 0x01, 0x28, 0x03, 0x52, 0x05, 0x6c, 0x69, 0x6d, 0x69, 0x74, 0x22,
-	0x63, 0x0a, 0x1b, 0x49, 0x6e, 0x6e, 0x65, 0x72, 0x43, 0x73, 0x46, 0x65, 0x74, 0x63, 0x68, 0x50,
-	0x6c, 0x61, 0x79, 0x65, 0x72, 0x51, 0x75, 0x65, 0x75, 0x65, 0x52, 0x65, 0x73, 0x70, 0x12, 0x14,
-	0x0a, 0x05, 0x74, 0x6f, 0x74, 0x61, 0x6c, 0x18, 0x01, 0x20, 0x01, 0x28, 0x03, 0x52, 0x05, 0x74,
-	0x6f, 0x74, 0x61, 0x6c, 0x12, 0x2e, 0x0a, 0x04, 0x6c, 0x69, 0x73, 0x74, 0x18, 0x02, 0x20, 0x01,
-	0x28, 0x0b, 0x32, 0x1a, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74,
-	0x6f, 0x62, 0x75, 0x66, 0x2e, 0x4c, 0x69, 0x73, 0x74, 0x56, 0x61, 0x6c, 0x75, 0x65, 0x52, 0x04,
-	0x6c, 0x69, 0x73, 0x74, 0x22, 0x64, 0x0a, 0x17, 0x49, 0x6e, 0x6e, 0x65, 0x72, 0x43, 0x73, 0x43,
-	0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x50, 0x6c, 0x61, 0x79, 0x65, 0x72, 0x52, 0x65, 0x71, 0x12,
-	0x13, 0x0a, 0x05, 0x63, 0x73, 0x5f, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04,
-	0x63, 0x73, 0x49, 0x64, 0x12, 0x1b, 0x0a, 0x09, 0x70, 0x6c, 0x61, 0x79, 0x65, 0x72, 0x5f, 0x69,
-	0x64, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x70, 0x6c, 0x61, 0x79, 0x65, 0x72, 0x49,
-	0x64, 0x12, 0x17, 0x0a, 0x07, 0x67, 0x61, 0x6d, 0x65, 0x5f, 0x69, 0x64, 0x18, 0x03, 0x20, 0x01,
-	0x28, 0x09, 0x52, 0x06, 0x67, 0x61, 0x6d, 0x65, 0x49, 0x64, 0x22, 0x1a, 0x0a, 0x18, 0x49, 0x6e,
-	0x6e, 0x65, 0x72, 0x43, 0x73, 0x43, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x50, 0x6c, 0x61, 0x79,
-	0x65, 0x72, 0x52, 0x65, 0x73, 0x70, 0x22, 0x54, 0x0a, 0x13, 0x4e, 0x6f, 0x74, 0x69, 0x66, 0x79,
-	0x55, 0x73, 0x65, 0x72, 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x52, 0x65, 0x71, 0x12, 0x12, 0x0a,
-	0x04, 0x74, 0x79, 0x70, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x03, 0x52, 0x04, 0x74, 0x79, 0x70,
-	0x65, 0x12, 0x10, 0x0a, 0x03, 0x75, 0x69, 0x64, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03,
-	0x75, 0x69, 0x64, 0x12, 0x17, 0x0a, 0x07, 0x67, 0x61, 0x6d, 0x65, 0x5f, 0x69, 0x64, 0x18, 0x03,
-	0x20, 0x01, 0x28, 0x09, 0x52, 0x06, 0x67, 0x61, 0x6d, 0x65, 0x49, 0x64, 0x22, 0x16, 0x0a, 0x14,
-	0x4e, 0x6f, 0x74, 0x69, 0x66, 0x79, 0x55, 0x73, 0x65, 0x72, 0x53, 0x74, 0x61, 0x74, 0x75, 0x73,
-	0x52, 0x65, 0x73, 0x70, 0x32, 0x8f, 0x03, 0x0a, 0x05, 0x49, 0x6e, 0x6e, 0x65, 0x72, 0x12, 0x52,
-	0x0a, 0x11, 0x70, 0x6c, 0x61, 0x79, 0x65, 0x72, 0x46, 0x65, 0x74, 0x63, 0x68, 0x43, 0x73, 0x49,
-	0x6e, 0x66, 0x6f, 0x12, 0x1d, 0x2e, 0x70, 0x62, 0x2e, 0x49, 0x6e, 0x6e, 0x65, 0x72, 0x50, 0x6c,
-	0x61, 0x79, 0x65, 0x72, 0x46, 0x65, 0x74, 0x63, 0x68, 0x43, 0x73, 0x49, 0x6e, 0x66, 0x6f, 0x52,
-	0x65, 0x71, 0x1a, 0x1e, 0x2e, 0x70, 0x62, 0x2e, 0x49, 0x6e, 0x6e, 0x65, 0x72, 0x50, 0x6c, 0x61,
+	0x6e, 0x6c, 0x69, 0x6e, 0x65, 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x22, 0x50, 0x0a, 0x18, 0x49,
+	0x6e, 0x6e, 0x65, 0x72, 0x50, 0x6c, 0x61, 0x79, 0x65, 0x72, 0x44, 0x69, 0x73, 0x63, 0x6f, 0x6e,
+	0x6e, 0x65, 0x63, 0x74, 0x52, 0x65, 0x71, 0x12, 0x1b, 0x0a, 0x09, 0x70, 0x6c, 0x61, 0x79, 0x65,
+	0x72, 0x5f, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x70, 0x6c, 0x61, 0x79,
+	0x65, 0x72, 0x49, 0x64, 0x12, 0x17, 0x0a, 0x07, 0x67, 0x61, 0x6d, 0x65, 0x5f, 0x69, 0x64, 0x18,
+	0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x06, 0x67, 0x61, 0x6d, 0x65, 0x49, 0x64, 0x22, 0x1b, 0x0a,
+	0x19, 0x49, 0x6e, 0x6e, 0x65, 0x72, 0x50, 0x6c, 0x61, 0x79, 0x65, 0x72, 0x44, 0x69, 0x73, 0x63,
+	0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x52, 0x65, 0x73, 0x70, 0x22, 0x32, 0x0a, 0x1a, 0x49, 0x6e,
+	0x6e, 0x65, 0x72, 0x43, 0x73, 0x46, 0x65, 0x74, 0x63, 0x68, 0x50, 0x6c, 0x61, 0x79, 0x65, 0x72,
+	0x51, 0x75, 0x65, 0x75, 0x65, 0x52, 0x65, 0x71, 0x12, 0x14, 0x0a, 0x05, 0x6c, 0x69, 0x6d, 0x69,
+	0x74, 0x18, 0x01, 0x20, 0x01, 0x28, 0x03, 0x52, 0x05, 0x6c, 0x69, 0x6d, 0x69, 0x74, 0x22, 0x63,
+	0x0a, 0x1b, 0x49, 0x6e, 0x6e, 0x65, 0x72, 0x43, 0x73, 0x46, 0x65, 0x74, 0x63, 0x68, 0x50, 0x6c,
+	0x61, 0x79, 0x65, 0x72, 0x51, 0x75, 0x65, 0x75, 0x65, 0x52, 0x65, 0x73, 0x70, 0x12, 0x14, 0x0a,
+	0x05, 0x74, 0x6f, 0x74, 0x61, 0x6c, 0x18, 0x01, 0x20, 0x01, 0x28, 0x03, 0x52, 0x05, 0x74, 0x6f,
+	0x74, 0x61, 0x6c, 0x12, 0x2e, 0x0a, 0x04, 0x6c, 0x69, 0x73, 0x74, 0x18, 0x02, 0x20, 0x01, 0x28,
+	0x0b, 0x32, 0x1a, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f,
+	0x62, 0x75, 0x66, 0x2e, 0x4c, 0x69, 0x73, 0x74, 0x56, 0x61, 0x6c, 0x75, 0x65, 0x52, 0x04, 0x6c,
+	0x69, 0x73, 0x74, 0x22, 0x64, 0x0a, 0x17, 0x49, 0x6e, 0x6e, 0x65, 0x72, 0x43, 0x73, 0x43, 0x6f,
+	0x6e, 0x6e, 0x65, 0x63, 0x74, 0x50, 0x6c, 0x61, 0x79, 0x65, 0x72, 0x52, 0x65, 0x71, 0x12, 0x13,
+	0x0a, 0x05, 0x63, 0x73, 0x5f, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x63,
+	0x73, 0x49, 0x64, 0x12, 0x17, 0x0a, 0x07, 0x67, 0x61, 0x6d, 0x65, 0x5f, 0x69, 0x64, 0x18, 0x02,
+	0x20, 0x01, 0x28, 0x09, 0x52, 0x06, 0x67, 0x61, 0x6d, 0x65, 0x49, 0x64, 0x12, 0x1b, 0x0a, 0x09,
+	0x70, 0x6c, 0x61, 0x79, 0x65, 0x72, 0x5f, 0x69, 0x64, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52,
+	0x08, 0x70, 0x6c, 0x61, 0x79, 0x65, 0x72, 0x49, 0x64, 0x22, 0x1a, 0x0a, 0x18, 0x49, 0x6e, 0x6e,
+	0x65, 0x72, 0x43, 0x73, 0x43, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x50, 0x6c, 0x61, 0x79, 0x65,
+	0x72, 0x52, 0x65, 0x73, 0x70, 0x22, 0x54, 0x0a, 0x13, 0x4e, 0x6f, 0x74, 0x69, 0x66, 0x79, 0x55,
+	0x73, 0x65, 0x72, 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x52, 0x65, 0x71, 0x12, 0x12, 0x0a, 0x04,
+	0x74, 0x79, 0x70, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x03, 0x52, 0x04, 0x74, 0x79, 0x70, 0x65,
+	0x12, 0x10, 0x0a, 0x03, 0x75, 0x69, 0x64, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x75,
+	0x69, 0x64, 0x12, 0x17, 0x0a, 0x07, 0x67, 0x61, 0x6d, 0x65, 0x5f, 0x69, 0x64, 0x18, 0x03, 0x20,
+	0x01, 0x28, 0x09, 0x52, 0x06, 0x67, 0x61, 0x6d, 0x65, 0x49, 0x64, 0x22, 0x16, 0x0a, 0x14, 0x4e,
+	0x6f, 0x74, 0x69, 0x66, 0x79, 0x55, 0x73, 0x65, 0x72, 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x52,
+	0x65, 0x73, 0x70, 0x32, 0xe0, 0x03, 0x0a, 0x05, 0x49, 0x6e, 0x6e, 0x65, 0x72, 0x12, 0x52, 0x0a,
+	0x11, 0x70, 0x6c, 0x61, 0x79, 0x65, 0x72, 0x46, 0x65, 0x74, 0x63, 0x68, 0x43, 0x73, 0x49, 0x6e,
+	0x66, 0x6f, 0x12, 0x1d, 0x2e, 0x70, 0x62, 0x2e, 0x49, 0x6e, 0x6e, 0x65, 0x72, 0x50, 0x6c, 0x61,
 	0x79, 0x65, 0x72, 0x46, 0x65, 0x74, 0x63, 0x68, 0x43, 0x73, 0x49, 0x6e, 0x66, 0x6f, 0x52, 0x65,
+	0x71, 0x1a, 0x1e, 0x2e, 0x70, 0x62, 0x2e, 0x49, 0x6e, 0x6e, 0x65, 0x72, 0x50, 0x6c, 0x61, 0x79,
+	0x65, 0x72, 0x46, 0x65, 0x74, 0x63, 0x68, 0x43, 0x73, 0x49, 0x6e, 0x66, 0x6f, 0x52, 0x65, 0x73,
+	0x70, 0x12, 0x4f, 0x0a, 0x10, 0x70, 0x6c, 0x61, 0x79, 0x65, 0x72, 0x44, 0x69, 0x73, 0x63, 0x6f,
+	0x6e, 0x6e, 0x65, 0x63, 0x74, 0x12, 0x1c, 0x2e, 0x70, 0x62, 0x2e, 0x49, 0x6e, 0x6e, 0x65, 0x72,
+	0x50, 0x6c, 0x61, 0x79, 0x65, 0x72, 0x44, 0x69, 0x73, 0x63, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74,
+	0x52, 0x65, 0x71, 0x1a, 0x1d, 0x2e, 0x70, 0x62, 0x2e, 0x49, 0x6e, 0x6e, 0x65, 0x72, 0x50, 0x6c,
+	0x61, 0x79, 0x65, 0x72, 0x44, 0x69, 0x73, 0x63, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x52, 0x65,
 	0x73, 0x70, 0x12, 0x55, 0x0a, 0x12, 0x63, 0x73, 0x46, 0x65, 0x74, 0x63, 0x68, 0x50, 0x6c, 0x61,
 	0x79, 0x65, 0x72, 0x51, 0x75, 0x65, 0x75, 0x65, 0x12, 0x1e, 0x2e, 0x70, 0x62, 0x2e, 0x49, 0x6e,
 	0x6e, 0x65, 0x72, 0x43, 0x73, 0x46, 0x65, 0x74, 0x63, 0x68, 0x50, 0x6c, 0x61, 0x79, 0x65, 0x72,
@@ -561,35 +666,39 @@ func file_pb_inner_proto_rawDescGZIP() []byte {
 	return file_pb_inner_proto_rawDescData
 }
 
-var file_pb_inner_proto_msgTypes = make([]protoimpl.MessageInfo, 8)
+var file_pb_inner_proto_msgTypes = make([]protoimpl.MessageInfo, 10)
 var file_pb_inner_proto_goTypes = []interface{}{
 	(*InnerPlayerFetchCsInfoReq)(nil),   // 0: pb.InnerPlayerFetchCsInfoReq
 	(*InnerPlayerFetchCsInfoResp)(nil),  // 1: pb.InnerPlayerFetchCsInfoResp
-	(*InnerCsFetchPlayerQueueReq)(nil),  // 2: pb.InnerCsFetchPlayerQueueReq
-	(*InnerCsFetchPlayerQueueResp)(nil), // 3: pb.InnerCsFetchPlayerQueueResp
-	(*InnerCsConnectPlayerReq)(nil),     // 4: pb.InnerCsConnectPlayerReq
-	(*InnerCsConnectPlayerResp)(nil),    // 5: pb.InnerCsConnectPlayerResp
-	(*NotifyUserStatusReq)(nil),         // 6: pb.NotifyUserStatusReq
-	(*NotifyUserStatusResp)(nil),        // 7: pb.NotifyUserStatusResp
-	(*structpb.ListValue)(nil),          // 8: google.protobuf.ListValue
+	(*InnerPlayerDisconnectReq)(nil),    // 2: pb.InnerPlayerDisconnectReq
+	(*InnerPlayerDisconnectResp)(nil),   // 3: pb.InnerPlayerDisconnectResp
+	(*InnerCsFetchPlayerQueueReq)(nil),  // 4: pb.InnerCsFetchPlayerQueueReq
+	(*InnerCsFetchPlayerQueueResp)(nil), // 5: pb.InnerCsFetchPlayerQueueResp
+	(*InnerCsConnectPlayerReq)(nil),     // 6: pb.InnerCsConnectPlayerReq
+	(*InnerCsConnectPlayerResp)(nil),    // 7: pb.InnerCsConnectPlayerResp
+	(*NotifyUserStatusReq)(nil),         // 8: pb.NotifyUserStatusReq
+	(*NotifyUserStatusResp)(nil),        // 9: pb.NotifyUserStatusResp
+	(*structpb.ListValue)(nil),          // 10: google.protobuf.ListValue
 }
 var file_pb_inner_proto_depIdxs = []int32{
-	8, // 0: pb.InnerCsFetchPlayerQueueResp.list:type_name -> google.protobuf.ListValue
-	0, // 1: pb.Inner.playerFetchCsInfo:input_type -> pb.InnerPlayerFetchCsInfoReq
-	2, // 2: pb.Inner.csFetchPlayerQueue:input_type -> pb.InnerCsFetchPlayerQueueReq
-	4, // 3: pb.Inner.csConnectPlayer:input_type -> pb.InnerCsConnectPlayerReq
-	6, // 4: pb.Inner.notifyUserOnline:input_type -> pb.NotifyUserStatusReq
-	6, // 5: pb.Inner.notifyUserOffline:input_type -> pb.NotifyUserStatusReq
-	1, // 6: pb.Inner.playerFetchCsInfo:output_type -> pb.InnerPlayerFetchCsInfoResp
-	3, // 7: pb.Inner.csFetchPlayerQueue:output_type -> pb.InnerCsFetchPlayerQueueResp
-	5, // 8: pb.Inner.csConnectPlayer:output_type -> pb.InnerCsConnectPlayerResp
-	7, // 9: pb.Inner.notifyUserOnline:output_type -> pb.NotifyUserStatusResp
-	7, // 10: pb.Inner.notifyUserOffline:output_type -> pb.NotifyUserStatusResp
-	6, // [6:11] is the sub-list for method output_type
-	1, // [1:6] is the sub-list for method input_type
-	1, // [1:1] is the sub-list for extension type_name
-	1, // [1:1] is the sub-list for extension extendee
-	0, // [0:1] is the sub-list for field type_name
+	10, // 0: pb.InnerCsFetchPlayerQueueResp.list:type_name -> google.protobuf.ListValue
+	0,  // 1: pb.Inner.playerFetchCsInfo:input_type -> pb.InnerPlayerFetchCsInfoReq
+	2,  // 2: pb.Inner.playerDisconnect:input_type -> pb.InnerPlayerDisconnectReq
+	4,  // 3: pb.Inner.csFetchPlayerQueue:input_type -> pb.InnerCsFetchPlayerQueueReq
+	6,  // 4: pb.Inner.csConnectPlayer:input_type -> pb.InnerCsConnectPlayerReq
+	8,  // 5: pb.Inner.notifyUserOnline:input_type -> pb.NotifyUserStatusReq
+	8,  // 6: pb.Inner.notifyUserOffline:input_type -> pb.NotifyUserStatusReq
+	1,  // 7: pb.Inner.playerFetchCsInfo:output_type -> pb.InnerPlayerFetchCsInfoResp
+	3,  // 8: pb.Inner.playerDisconnect:output_type -> pb.InnerPlayerDisconnectResp
+	5,  // 9: pb.Inner.csFetchPlayerQueue:output_type -> pb.InnerCsFetchPlayerQueueResp
+	7,  // 10: pb.Inner.csConnectPlayer:output_type -> pb.InnerCsConnectPlayerResp
+	9,  // 11: pb.Inner.notifyUserOnline:output_type -> pb.NotifyUserStatusResp
+	9,  // 12: pb.Inner.notifyUserOffline:output_type -> pb.NotifyUserStatusResp
+	7,  // [7:13] is the sub-list for method output_type
+	1,  // [1:7] is the sub-list for method input_type
+	1,  // [1:1] is the sub-list for extension type_name
+	1,  // [1:1] is the sub-list for extension extendee
+	0,  // [0:1] is the sub-list for field type_name
 }
 
 func init() { file_pb_inner_proto_init() }
@@ -623,7 +732,7 @@ func file_pb_inner_proto_init() {
 			}
 		}
 		file_pb_inner_proto_msgTypes[2].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*InnerCsFetchPlayerQueueReq); i {
+			switch v := v.(*InnerPlayerDisconnectReq); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -635,7 +744,7 @@ func file_pb_inner_proto_init() {
 			}
 		}
 		file_pb_inner_proto_msgTypes[3].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*InnerCsFetchPlayerQueueResp); i {
+			switch v := v.(*InnerPlayerDisconnectResp); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -647,7 +756,7 @@ func file_pb_inner_proto_init() {
 			}
 		}
 		file_pb_inner_proto_msgTypes[4].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*InnerCsConnectPlayerReq); i {
+			switch v := v.(*InnerCsFetchPlayerQueueReq); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -659,7 +768,7 @@ func file_pb_inner_proto_init() {
 			}
 		}
 		file_pb_inner_proto_msgTypes[5].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*InnerCsConnectPlayerResp); i {
+			switch v := v.(*InnerCsFetchPlayerQueueResp); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -671,7 +780,7 @@ func file_pb_inner_proto_init() {
 			}
 		}
 		file_pb_inner_proto_msgTypes[6].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*NotifyUserStatusReq); i {
+			switch v := v.(*InnerCsConnectPlayerReq); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -683,6 +792,30 @@ func file_pb_inner_proto_init() {
 			}
 		}
 		file_pb_inner_proto_msgTypes[7].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*InnerCsConnectPlayerResp); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_pb_inner_proto_msgTypes[8].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*NotifyUserStatusReq); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_pb_inner_proto_msgTypes[9].Exporter = func(v interface{}, i int) interface{} {
 			switch v := v.(*NotifyUserStatusResp); i {
 			case 0:
 				return &v.state
@@ -701,7 +834,7 @@ func file_pb_inner_proto_init() {
 			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
 			RawDescriptor: file_pb_inner_proto_rawDesc,
 			NumEnums:      0,
-			NumMessages:   8,
+			NumMessages:   10,
 			NumExtensions: 0,
 			NumServices:   1,
 		},

core/inner/rpc/pb/inner.proto

@@ -10,8 +10,8 @@ import "google/protobuf/struct.proto";
 Player Command Request Bean
  */
 message InnerPlayerFetchCsInfoReq{
-  string player_id = 1;
-  string game_id = 2;
+  string game_id = 1;
+  string player_id = 2;
   string cs_id = 3;
 }
 
@@ -23,6 +23,13 @@ message InnerPlayerFetchCsInfoResp{
   int64 online_status = 5;
 }
 
+message InnerPlayerDisconnectReq{
+  string player_id = 1;
+  string game_id = 2;
+}
+
+message InnerPlayerDisconnectResp{}
+
 /**
 Cs Command Request Bean
  */
@@ -37,8 +44,8 @@ message InnerCsFetchPlayerQueueResp{
 
 message InnerCsConnectPlayerReq{
   string cs_id = 1;
-  string player_id = 2;
-  string game_id = 3;
+  string game_id = 2;
+  string player_id = 3;
 }
 
 message InnerCsConnectPlayerResp{}
@@ -54,6 +61,7 @@ message NotifyUserStatusResp{}
 
 service Inner {
   rpc playerFetchCsInfo (InnerPlayerFetchCsInfoReq) returns (InnerPlayerFetchCsInfoResp);
+  rpc playerDisconnect (InnerPlayerDisconnectReq) returns (InnerPlayerDisconnectResp);
 
   rpc csFetchPlayerQueue (InnerCsFetchPlayerQueueReq) returns (InnerCsFetchPlayerQueueResp);
   rpc csConnectPlayer (InnerCsConnectPlayerReq) returns (InnerCsConnectPlayerResp);

core/inner/rpc/pb/inner_grpc.pb.go

@@ -23,6 +23,7 @@ const _ = grpc.SupportPackageIsVersion7
 // For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream.
 type InnerClient interface {
 	PlayerFetchCsInfo(ctx context.Context, in *InnerPlayerFetchCsInfoReq, opts ...grpc.CallOption) (*InnerPlayerFetchCsInfoResp, error)
+	PlayerDisconnect(ctx context.Context, in *InnerPlayerDisconnectReq, opts ...grpc.CallOption) (*InnerPlayerDisconnectResp, error)
 	CsFetchPlayerQueue(ctx context.Context, in *InnerCsFetchPlayerQueueReq, opts ...grpc.CallOption) (*InnerCsFetchPlayerQueueResp, error)
 	CsConnectPlayer(ctx context.Context, in *InnerCsConnectPlayerReq, opts ...grpc.CallOption) (*InnerCsConnectPlayerResp, error)
 	NotifyUserOnline(ctx context.Context, in *NotifyUserStatusReq, opts ...grpc.CallOption) (*NotifyUserStatusResp, error)
@@ -46,6 +47,15 @@ func (c *innerClient) PlayerFetchCsInfo(ctx context.Context, in *InnerPlayerFetc
 	return out, nil
 }
 
+func (c *innerClient) PlayerDisconnect(ctx context.Context, in *InnerPlayerDisconnectReq, opts ...grpc.CallOption) (*InnerPlayerDisconnectResp, error) {
+	out := new(InnerPlayerDisconnectResp)
+	err := c.cc.Invoke(ctx, "/pb.Inner/playerDisconnect", in, out, opts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
 func (c *innerClient) CsFetchPlayerQueue(ctx context.Context, in *InnerCsFetchPlayerQueueReq, opts ...grpc.CallOption) (*InnerCsFetchPlayerQueueResp, error) {
 	out := new(InnerCsFetchPlayerQueueResp)
 	err := c.cc.Invoke(ctx, "/pb.Inner/csFetchPlayerQueue", in, out, opts...)
@@ -87,6 +97,7 @@ func (c *innerClient) NotifyUserOffline(ctx context.Context, in *NotifyUserStatu
 // for forward compatibility
 type InnerServer interface {
 	PlayerFetchCsInfo(context.Context, *InnerPlayerFetchCsInfoReq) (*InnerPlayerFetchCsInfoResp, error)
+	PlayerDisconnect(context.Context, *InnerPlayerDisconnectReq) (*InnerPlayerDisconnectResp, error)
 	CsFetchPlayerQueue(context.Context, *InnerCsFetchPlayerQueueReq) (*InnerCsFetchPlayerQueueResp, error)
 	CsConnectPlayer(context.Context, *InnerCsConnectPlayerReq) (*InnerCsConnectPlayerResp, error)
 	NotifyUserOnline(context.Context, *NotifyUserStatusReq) (*NotifyUserStatusResp, error)
@@ -101,6 +112,9 @@ type UnimplementedInnerServer struct {
 func (UnimplementedInnerServer) PlayerFetchCsInfo(context.Context, *InnerPlayerFetchCsInfoReq) (*InnerPlayerFetchCsInfoResp, error) {
 	return nil, status.Errorf(codes.Unimplemented, "method PlayerFetchCsInfo not implemented")
 }
+func (UnimplementedInnerServer) PlayerDisconnect(context.Context, *InnerPlayerDisconnectReq) (*InnerPlayerDisconnectResp, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method PlayerDisconnect not implemented")
+}
 func (UnimplementedInnerServer) CsFetchPlayerQueue(context.Context, *InnerCsFetchPlayerQueueReq) (*InnerCsFetchPlayerQueueResp, error) {
 	return nil, status.Errorf(codes.Unimplemented, "method CsFetchPlayerQueue not implemented")
 }
@@ -144,6 +158,24 @@ func _Inner_PlayerFetchCsInfo_Handler(srv interface{}, ctx context.Context, dec
 	return interceptor(ctx, in, info, handler)
 }
 
+func _Inner_PlayerDisconnect_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(InnerPlayerDisconnectReq)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(InnerServer).PlayerDisconnect(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: "/pb.Inner/playerDisconnect",
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(InnerServer).PlayerDisconnect(ctx, req.(*InnerPlayerDisconnectReq))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
 func _Inner_CsFetchPlayerQueue_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
 	in := new(InnerCsFetchPlayerQueueReq)
 	if err := dec(in); err != nil {
@@ -227,6 +259,10 @@ var Inner_ServiceDesc = grpc.ServiceDesc{
 			MethodName: "playerFetchCsInfo",
 			Handler:    _Inner_PlayerFetchCsInfo_Handler,
 		},
+		{
+			MethodName: "playerDisconnect",
+			Handler:    _Inner_PlayerDisconnect_Handler,
+		},
 		{
 			MethodName: "csFetchPlayerQueue",
 			Handler:    _Inner_CsFetchPlayerQueue_Handler,

docker-compose-env.yml

@@ -115,7 +115,7 @@ services:
     ports:
       - "9092:9092"
     environment:
-      - "KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://127.0.0.1:9092"
+      - "KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092"
       - "KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092 "
       - "KAFKA_ADVERTISED_HOST_NAME=kafka"
       - "KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181"

flowsrv/rpc/etc/flowsrv.yaml

@@ -3,7 +3,7 @@ ListenOn: 0.0.0.0:10200
 
 Telemetry:
   Name: flowsrv-api
-  Endpoint: http://127.0.0.1:14268/api/traces
+  Endpoint: http://localhost:14268/api/traces
   Sampler: 1.0
   Batcher: jaeger
 
@@ -19,18 +19,18 @@ InnerRpcConf:
 
 KqMsgBoxConsumerConf:
   Brokers:
-    - 127.0.0.1:9092
+    - localhost:9092
   Topic: recv-box-topic
   GroupId: flowsrv-api
 
 KqCmdBoxConsumerConf:
   Brokers:
-    - 127.0.0.1:9092
+    - localhost:9092
   Topic: cmd-box-topic
   GroupId: flowsrv-api
 
 Redis:
-  Host: 127.0.0.1:6379
+  Host: localhost:6379
   Type: node
   Pass: ylink
 

flowsrv/rpc/internal/logic/connectlogic.go

@@ -53,7 +53,7 @@ func (l *ConnectLogic) Connect(in *pb.CommandReq, stream pb.Flowsrv_ConnectServe
 	}
 
 	var flowId string
-	if in.Type == globalkey.CONNECT_TYPE_PLAYER {
+	if in.Type == globalkey.ConnectTypePlayer {
 		flowId = gameId + "_" + uid
 	} else {
 		flowId = uid
@@ -93,7 +93,7 @@ func (l *ConnectLogic) checkAuth(in *pb.CommandReq) (string, string, error) {
 	if token.Valid {
 		//将获取的token中的Claims强转为MapClaims
 		claims, _ := token.Claims.(jwt.MapClaims)
-		if in.Type == globalkey.CONNECT_TYPE_PLAYER {
+		if in.Type == globalkey.ConnectTypePlayer {
 			uid = claims[jwtkey.PlayerId].(string)
 			gameId = claims[jwtkey.GameId].(string)
 		} else {

flowsrv/rpc/internal/logic/disconnectlogic.go

@@ -1,98 +0,0 @@
-package logic
-
-import (
-	"context"
-	"github.com/golang-jwt/jwt/v4"
-	"github.com/pkg/errors"
-	"ylink/comm/globalkey"
-	"ylink/comm/jwtkey"
-	"ylink/comm/result"
-	"ylink/core/inner/rpc/inner"
-	"ylink/flowsrv/rpc/internal/mgr"
-	"ylink/flowsrv/rpc/internal/svc"
-	"ylink/flowsrv/rpc/pb"
-
-	"github.com/zeromicro/go-zero/core/logx"
-)
-
-type DisconnectLogic struct {
-	ctx    context.Context
-	svcCtx *svc.ServiceContext
-	logx.Logger
-}
-
-func NewDisconnectLogic(ctx context.Context, svcCtx *svc.ServiceContext) *DisconnectLogic {
-	return &DisconnectLogic{
-		ctx:    ctx,
-		svcCtx: svcCtx,
-		Logger: logx.WithContext(ctx),
-	}
-}
-
-func (l *DisconnectLogic) Disconnect(in *pb.CommandReq) (*pb.CommandResp, error) {
-	uid, gameId, err := l.checkAuth(in)
-	if err != nil {
-		return &pb.CommandResp{
-			Code: result.TokenParseError,
-			Msg:  err.Error(),
-			Data: nil,
-		}, err
-	}
-	_, err = l.svcCtx.InnerRpc.NotifyUserOffline(l.ctx, &inner.NotifyUserStatusReq{
-		Type:   in.Type,
-		Uid:    uid,
-		GameId: gameId,
-	})
-	if err != nil {
-		return &pb.CommandResp{
-			Code: result.ServerCommonError,
-			Msg:  err.Error(),
-			Data: nil,
-		}, err
-	}
-
-	var flowId string
-	if in.Type == globalkey.CONNECT_TYPE_PLAYER {
-		flowId = gameId + "_" + uid
-	} else {
-		flowId = uid
-	}
-	mgr.GetFlowMgrInstance().UnRegister(flowId)
-
-	return &pb.CommandResp{
-		Code: result.Ok,
-		Msg:  "success",
-		Data: nil,
-	}, nil
-}
-
-func (l *DisconnectLogic) checkAuth(in *pb.CommandReq) (string, string, error) {
-	token, err := jwt.Parse(in.AccessToken, func(token *jwt.Token) (i interface{}, err error) {
-		return []byte(l.svcCtx.Config.JwtAuth.AccessSecret), nil
-	})
-
-	uid := ""
-	gameId := ""
-	if token.Valid {
-		//将获取的token中的Claims强转为MapClaims
-		claims, _ := token.Claims.(jwt.MapClaims)
-		if in.Type == globalkey.CONNECT_TYPE_PLAYER {
-			uid = claims[jwtkey.PlayerId].(string)
-			gameId = claims[jwtkey.GameId].(string)
-		} else {
-			uid = claims[jwtkey.CsId].(string)
-		}
-		return uid, gameId, nil
-	} else if ve, ok := err.(*jwt.ValidationError); ok {
-		if ve.Errors&jwt.ValidationErrorMalformed != 0 {
-			return uid, gameId, errors.Wrap(result.NewErrCode(result.TokenParseError), "")
-		} else if ve.Errors&(jwt.ValidationErrorExpired|jwt.ValidationErrorNotValidYet) != 0 {
-			// Token is either expired or not active yet
-			return uid, gameId, errors.Wrap(result.NewErrCode(result.TokenExpireError), "")
-		} else {
-			return uid, gameId, errors.Wrap(result.NewErrCode(result.TokenParseError), "")
-		}
-	} else {
-		return uid, gameId, errors.Wrap(result.NewErrCode(result.TokenParseError), "")
-	}
-}

flowsrv/rpc/internal/mgr/flowmgr.go

@@ -44,7 +44,7 @@ func (manager *flowManager) registerFlow(flow *model.Flow) {
 		select {
 		case <-flow.Stream.Context().Done():
 			if manager.Has(flow.FlowId) {
-				flow.Logger.Infof("flowstream was disconnected abnormally")
+				flow.Logger.Infof("stream was disconnected abnormally")
 				manager.UnRegister(flow.FlowId)
 				flow.SvcCtx.InnerRpc.NotifyUserOffline(flow.Ctx, &inner.NotifyUserStatusReq{
 					Type:   flow.Type,
@@ -94,7 +94,6 @@ func (manager *flowManager) UnRegister(flowId string) {
 	if manager.flowMap.Contains(flowId) {
 		flow := manager.Get(flowId)
 		close(flow.Message)
-		//flow.EndRmq <- 0
 		manager.flowMap.Erase(flowId)
 	}
 }

flowsrv/rpc/internal/svc/servicecontext.go

@@ -5,6 +5,7 @@ import (
 	"github.com/Shopify/sarama"
 	"github.com/bytedance/sonic"
 	"github.com/go-redis/redis/v8"
+	"github.com/gookit/event"
 	"github.com/zeromicro/go-zero/core/logx"
 	gozerotrace "github.com/zeromicro/go-zero/core/trace"
 	"github.com/zeromicro/go-zero/zrpc"
@@ -13,6 +14,7 @@ import (
 	"go.opentelemetry.io/otel/propagation"
 	oteltrace "go.opentelemetry.io/otel/trace"
 	"net/http"
+	"ylink/comm/globalkey"
 	"ylink/comm/kafka"
 	"ylink/comm/model"
 	"ylink/comm/trace"
@@ -128,15 +130,14 @@ func (s *ServiceContext) handleCommand(sess sarama.ConsumerGroupSession, msg *sa
 			return
 		}
 		trace.StartTrace(ctx, "FlowsrvServer.handleCommand.PushMessage", func(ctx context.Context) {
-			logx.WithContext(ctx).Infof("recv command: %v", message)
 
-			//// 投递到receiver_id对应的redis队列暂存
-			//intCmd := s.RedisClient.LPush(ctx, message.ReceiverId, string(msg.Value))
-			//if size, err := intCmd.Result(); err != nil {
-			//	logx.WithContext(ctx).Errorf("push message rmq err %v", err)
-			//} else {
-			//	logx.WithContext(ctx).Infof("current rmq size: %d", size)
-			//}
+			// 投递到receiver_id对应的redis队列暂存
+			intCmd := s.RedisClient.LPush(ctx, message.ReceiverId, string(msg.Value))
+			if size, err := intCmd.Result(); err != nil {
+				logx.WithContext(ctx).Errorf("push message rmq err %v", err)
+			} else {
+				logx.WithContext(ctx).Infof("current rmq size: %d", size)
+			}
 
 			sess.MarkMessage(msg, "")
 		}, attribute.String("msg.key", string(msg.Key)))
@@ -146,4 +147,10 @@ func (s *ServiceContext) handleCommand(sess sarama.ConsumerGroupSession, msg *sa
 func (s *ServiceContext) subscribe() {
 	go s.MessageConsumerGroup.RegisterHandleAndConsumer(s)
 	go s.CommandConsumerGroup.RegisterHandleAndConsumer(s)
+
+	// 注册事件
+	event.On(globalkey.EventUnsubscribeRmq, event.ListenerFunc(func(e event.Event) error {
+
+		return nil
+	}), event.High)
 }

flowsrv/rpc/pb/flowsrv.proto

@@ -20,5 +20,4 @@ message CommandResp {
 
 service Flowsrv {
   rpc connect(CommandReq) returns (stream CommandResp);
-  rpc disconnect(CommandReq) returns (CommandResp);
 }