v0.0.1开发：玩家等待队列功能开发

comm/kafka/producer.go

@@ -21,8 +21,6 @@ type Producer struct {
 }
 
 func NewKafkaProducer(addr []string, topic string) *Producer {
-	logx.Infof("brokers: %v", addr)
-	logx.Infof("topic: %s", topic)
 	p := Producer{}
 	p.config = sarama.NewConfig()
 	// Whether to enable the successes channel to be notified after the message is sent successfully
@@ -47,6 +45,7 @@ func NewKafkaProducer(addr []string, topic string) *Producer {
 }
 
 func (p *Producer) SendMessage(ctx context.Context, m string, key ...string) (partition int32, offset int64, err error) {
+	logx.WithContext(ctx).Infof("send msg to kafka, msg: %s", m)
 	traceId := ctxdata.GetTraceIdFromCtx(ctx)
 	msg := &sarama.ProducerMessage{}
 	msg.Headers = []sarama.RecordHeader{{

comm/model/playerinfo.go

@@ -0,0 +1,11 @@
+//@File     playerinfo.go
+//@Time     2022/05/19
+//@Author   #Suyghur,
+
+package model
+
+type PlayerWaitingInfo struct {
+	PlayerId    string `json:"player_id"`
+	GameId      string `json:"game_id"`
+	EnqueueTime int64  `json:"enqueue_time"`
+}

core/inner/rpc/inner/inner.go

@@ -17,19 +17,17 @@ type (
 	InnerCsConnectPlayerResp    = pb.InnerCsConnectPlayerResp
 	InnerCsFetchPlayerQueueReq  = pb.InnerCsFetchPlayerQueueReq
 	InnerCsFetchPlayerQueueResp = pb.InnerCsFetchPlayerQueueResp
-	InnerPlayerDisconnectReq    = pb.InnerPlayerDisconnectReq
-	InnerPlayerDisconnectResp   = pb.InnerPlayerDisconnectResp
 	InnerPlayerFetchCsInfoReq   = pb.InnerPlayerFetchCsInfoReq
 	InnerPlayerFetchCsInfoResp  = pb.InnerPlayerFetchCsInfoResp
-	UpdateUserStatusReq         = pb.UpdateUserStatusReq
-	UpdateUserStatusResp        = pb.UpdateUserStatusResp
+	NotifyUserStatusReq         = pb.NotifyUserStatusReq
+	NotifyUserStatusResp        = pb.NotifyUserStatusResp
 
 	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)
-		UpdateUserStatus(ctx context.Context, in *UpdateUserStatusReq, opts ...grpc.CallOption) (*UpdateUserStatusResp, error)
+		NotifyUserOnline(ctx context.Context, in *NotifyUserStatusReq, opts ...grpc.CallOption) (*NotifyUserStatusResp, error)
+		NotifyUserOffline(ctx context.Context, in *NotifyUserStatusReq, opts ...grpc.CallOption) (*NotifyUserStatusResp, error)
 	}
 
 	defaultInner struct {
@@ -48,11 +46,6 @@ 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...)
@@ -63,7 +56,12 @@ func (m *defaultInner) CsConnectPlayer(ctx context.Context, in *InnerCsConnectPl
 	return client.CsConnectPlayer(ctx, in, opts...)
 }
 
-func (m *defaultInner) UpdateUserStatus(ctx context.Context, in *UpdateUserStatusReq, opts ...grpc.CallOption) (*UpdateUserStatusResp, error) {
+func (m *defaultInner) NotifyUserOnline(ctx context.Context, in *NotifyUserStatusReq, opts ...grpc.CallOption) (*NotifyUserStatusResp, error) {
+	client := pb.NewInnerClient(m.cli.Conn())
+	return client.NotifyUserOnline(ctx, in, opts...)
+}
+
+func (m *defaultInner) NotifyUserOffline(ctx context.Context, in *NotifyUserStatusReq, opts ...grpc.CallOption) (*NotifyUserStatusResp, error) {
 	client := pb.NewInnerClient(m.cli.Conn())
-	return client.UpdateUserStatus(ctx, in, opts...)
+	return client.NotifyUserOffline(ctx, in, opts...)
 }

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

@@ -5,13 +5,13 @@
 package ext
 
 import (
-	treemap "github.com/liyue201/gostl/ds/map"
-	"github.com/liyue201/gostl/ds/set"
+	"github.com/liyue201/gostl/ds/list/simplelist"
+	"github.com/liyue201/gostl/ds/map"
 )
 
 var (
 	Game2PlayerMap     *treemap.Map
 	CsMap              *treemap.Map
 	Game2PlayerStatMap *treemap.Map
-	CsStatSet          *set.Set
+	WaitingQueue       *simplelist.List
 )

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

@@ -4,7 +4,10 @@ import (
 	"context"
 	"github.com/pkg/errors"
 	"google.golang.org/protobuf/types/known/structpb"
+	"time"
+	"ylink/comm/model"
 	"ylink/comm/result"
+	"ylink/core/inner/rpc/internal/ext"
 	"ylink/core/inner/rpc/internal/svc"
 	"ylink/core/inner/rpc/pb"
 
@@ -26,20 +29,32 @@ func NewCsFetchPlayerQueueLogic(ctx context.Context, svcCtx *svc.ServiceContext)
 }
 
 func (l *CsFetchPlayerQueueLogic) CsFetchPlayerQueue(in *pb.InnerCsFetchPlayerQueueReq) (*pb.InnerCsFetchPlayerQueueResp, error) {
-	// todo: 查询等待用户的队列
-
-	list, err := structpb.NewList([]interface{}{
-		map[string]interface{}{
-			"player_id": "player1111",
-			"game_id":   "game1231",
-			"wait_time": 1000,
-		},
-		map[string]interface{}{
-			"player_id": "player2222",
-			"game_id":   "game1231",
-			"wait_time": 10,
-		},
-	})
+	queueLen := int64(ext.WaitingQueue.Len())
+	if queueLen == 0 {
+		// 等待队列为空直接返回
+		return &pb.InnerCsFetchPlayerQueueResp{
+			List: nil,
+		}, nil
+	}
+
+	var index int64 = 0
+	if in.Limit != 0 && in.Limit < queueLen {
+		queueLen = in.Limit
+	}
+
+	queue := make([]interface{}, queueLen)
+
+	for node := ext.WaitingQueue.FrontNode(); node != nil && index < queueLen; node = node.Next() {
+		info := node.Value.(*model.PlayerWaitingInfo)
+		queue[index] = map[string]interface{}{
+			"player_id": info.PlayerId,
+			"game_id":   info.GameId,
+			"wait_time": time.Now().Unix() - info.EnqueueTime,
+		}
+		index += 1
+	}
+
+	list, err := structpb.NewList(queue)
 	if err != nil {
 		return nil, errors.Wrap(result.NewErrMsg("fetch player wait queue error"), "")
 	}

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

@@ -5,6 +5,7 @@ import (
 	"github.com/liyue201/gostl/ds/set"
 	"github.com/pkg/errors"
 	"ylink/comm/globalkey"
+	"ylink/comm/model"
 	"ylink/comm/result"
 	"ylink/core/inner/rpc/internal/ext"
 
@@ -14,21 +15,21 @@ import (
 	"github.com/zeromicro/go-zero/core/logx"
 )
 
-type UpdateUserStatusLogic struct {
+type NotifyUserOfflineLogic struct {
 	ctx    context.Context
 	svcCtx *svc.ServiceContext
 	logx.Logger
 }
 
-func NewUpdateUserStatusLogic(ctx context.Context, svcCtx *svc.ServiceContext) *UpdateUserStatusLogic {
-	return &UpdateUserStatusLogic{
+func NewNotifyUserOfflineLogic(ctx context.Context, svcCtx *svc.ServiceContext) *NotifyUserOfflineLogic {
+	return &NotifyUserOfflineLogic{
 		ctx:    ctx,
 		svcCtx: svcCtx,
 		Logger: logx.WithContext(ctx),
 	}
 }
 
-func (l *UpdateUserStatusLogic) UpdateUserStatus(in *pb.UpdateUserStatusReq) (*pb.UpdateUserStatusResp, error) {
+func (l *NotifyUserOfflineLogic) NotifyUserOffline(in *pb.NotifyUserStatusReq) (*pb.NotifyUserStatusResp, error) {
 	switch in.Type {
 	case globalkey.CONNECT_TYPE_PLAYER:
 		// 修改玩家在线状态
@@ -38,24 +39,25 @@ func (l *UpdateUserStatusLogic) UpdateUserStatus(in *pb.UpdateUserStatusReq) (*p
 			if playerStatSet.Contains(in.Uid) {
 				// 有则清除，代表下线
 				playerStatSet.Erase(in.Uid)
-			} else {
-				playerStatSet.Insert(in.Uid)
 			}
-		} else {
-			playerStatSet := set.New()
-			playerStatSet.Insert(in.Uid)
-			ext.Game2PlayerStatMap.Insert(in.GameId, playerStatSet)
 		}
+
+		for n := ext.WaitingQueue.FrontNode(); n != nil; n = n.Next() {
+			info := n.Value.(*model.PlayerWaitingInfo)
+			if info.GameId == in.GameId && info.PlayerId == in.Uid {
+				l.Logger.Infof("remove the player from the queue, game_id: %s, player_id: %s", in.GameId, in.Uid)
+				ext.WaitingQueue.Remove(nil, n)
+				break
+			}
+		}
+		l.Logger.Infof("waiting queue size: %d", ext.WaitingQueue.Len())
+		l.Logger.Infof("waiting queue: %s", ext.WaitingQueue.String())
 	case globalkey.CONNECT_TYPE_CS:
 		// 修改客服在线状态
-		if ext.CsStatSet.Contains(in.Uid) {
-			// 有则清除，代表下线
-			ext.CsStatSet.Erase(in.Uid)
-		} else {
-			ext.CsStatSet.Insert(in.Uid)
-		}
+		csInfo := ext.CsMap.Get(in.Uid).(*model.CsInfo)
+		csInfo.OnlineStatus = 0
 	default:
 		return nil, errors.Wrap(result.NewErrMsg("no such user type"), "")
 	}
-	return &pb.UpdateUserStatusResp{}, nil
+	return &pb.NotifyUserStatusResp{}, nil
 }

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

@@ -0,0 +1,76 @@
+package logic
+
+import (
+	"context"
+	treemap "github.com/liyue201/gostl/ds/map"
+	"github.com/liyue201/gostl/ds/set"
+	"github.com/pkg/errors"
+	"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"
+
+	"github.com/zeromicro/go-zero/core/logx"
+)
+
+type NotifyUserOnlineLogic struct {
+	ctx    context.Context
+	svcCtx *svc.ServiceContext
+	logx.Logger
+}
+
+func NewNotifyUserOnlineLogic(ctx context.Context, svcCtx *svc.ServiceContext) *NotifyUserOnlineLogic {
+	return &NotifyUserOnlineLogic{
+		ctx:    ctx,
+		svcCtx: svcCtx,
+		Logger: logx.WithContext(ctx),
+	}
+}
+
+func (l *NotifyUserOnlineLogic) NotifyUserOnline(in *pb.NotifyUserStatusReq) (*pb.NotifyUserStatusResp, error) {
+	switch in.Type {
+	case globalkey.CONNECT_TYPE_PLAYER:
+		// 修改玩家在线状态
+		if ext.Game2PlayerStatMap.Contains(in.GameId) {
+			// 有则取出玩家的set
+			playerStatSet := ext.Game2PlayerStatMap.Get(in.GameId).(*set.Set)
+			if !playerStatSet.Contains(in.Uid) {
+				playerStatSet.Insert(in.Uid)
+			}
+		} else {
+			playerStatSet := set.New()
+			playerStatSet.Insert(in.Uid)
+			ext.Game2PlayerStatMap.Insert(in.GameId, playerStatSet)
+		}
+
+		// 判断是否有专属客服，没有则放入等待队列
+		if ext.Game2PlayerMap.Contains(in.GameId) {
+			p2cMap := ext.Game2PlayerMap.Get(in.GameId).(*treemap.Map)
+			if !p2cMap.Contains(in.Uid) {
+				ext.WaitingQueue.PushBack(&model.PlayerWaitingInfo{
+					PlayerId:    in.Uid,
+					GameId:      in.GameId,
+					EnqueueTime: time.Now().Unix(),
+				})
+			}
+		} else {
+			ext.WaitingQueue.PushBack(&model.PlayerWaitingInfo{
+				PlayerId:    in.Uid,
+				GameId:      in.GameId,
+				EnqueueTime: time.Now().Unix(),
+			})
+		}
+		l.Logger.Infof("enqueue waiting list: %s", ext.WaitingQueue.String())
+	case globalkey.CONNECT_TYPE_CS:
+		// 修改客服在线状态
+		csInfo := ext.CsMap.Get(in.Uid).(*model.CsInfo)
+		csInfo.OnlineStatus = 1
+	default:
+		return nil, errors.Wrap(result.NewErrMsg("no such user type"), "")
+	}
+	return &pb.NotifyUserStatusResp{}, nil
+}

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

@@ -1,29 +0,0 @@
-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.InnerPlayerDisconnectReq) (*pb.InnerPlayerDisconnectResp, error) {
-	// todo: 修改玩家在线状态
-
-	return &pb.InnerPlayerDisconnectResp{}, nil
-}

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

@@ -5,15 +5,6 @@ package server
 
 import (
 	"context"
-	"encoding/json"
-	"github.com/Shopify/sarama"
-	treemap "github.com/liyue201/gostl/ds/map"
-	"github.com/zeromicro/go-zero/core/logx"
-	"go.opentelemetry.io/otel/attribute"
-	"ylink/comm/kafka"
-	"ylink/comm/model"
-	"ylink/comm/trace"
-	"ylink/core/inner/rpc/internal/ext"
 
 	"ylink/core/inner/rpc/internal/logic"
 	"ylink/core/inner/rpc/internal/svc"
@@ -23,24 +14,12 @@ import (
 type InnerServer struct {
 	svcCtx *svc.ServiceContext
 	pb.UnimplementedInnerServer
-	ConsumerGroup *kafka.ConsumerGroup
 }
 
 func NewInnerServer(svcCtx *svc.ServiceContext) *InnerServer {
-	server := &InnerServer{
+	return &InnerServer{
 		svcCtx: svcCtx,
-		ConsumerGroup: kafka.NewConsumerGroup(&kafka.ConsumerGroupConfig{
-			KafkaVersion:   sarama.V1_0_0_0,
-			OffsetsInitial: sarama.OffsetNewest,
-			IsReturnErr:    false,
-		},
-			svcCtx.Config.KqMsgBoxConsumerConf.Brokers,
-			[]string{svcCtx.Config.KqMsgBoxConsumerConf.Topic},
-			svcCtx.Config.KqMsgBoxConsumerConf.GroupId),
 	}
-	server.subscribe()
-	return server
-
 }
 
 func (s *InnerServer) PlayerFetchCsInfo(ctx context.Context, in *pb.InnerPlayerFetchCsInfoReq) (*pb.InnerPlayerFetchCsInfoResp, error) {
@@ -48,11 +27,6 @@ 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)
@@ -63,55 +37,12 @@ func (s *InnerServer) CsConnectPlayer(ctx context.Context, in *pb.InnerCsConnect
 	return l.CsConnectPlayer(in)
 }
 
-func (s *InnerServer) UpdateUserStatus(ctx context.Context, in *pb.UpdateUserStatusReq) (*pb.UpdateUserStatusResp, error) {
-	l := logic.NewUpdateUserStatusLogic(ctx, s.svcCtx)
-	return l.UpdateUserStatus(in)
-}
-
-func (s *InnerServer) Setup(_ sarama.ConsumerGroupSession) error {
-	return nil
-}
-
-func (s *InnerServer) Cleanup(_ sarama.ConsumerGroupSession) error {
-	return nil
-}
-
-func (s *InnerServer) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
-	for msg := range claim.Messages() {
-		if msg.Topic == s.svcCtx.Config.KqMsgBoxConsumerConf.Topic {
-			s.handleMessage(sess, msg)
-		}
-	}
-	return nil
-}
-
-func (s *InnerServer) handleMessage(sess sarama.ConsumerGroupSession, msg *sarama.ConsumerMessage) {
-	traceId := kafka.GetTraceFromHeader(msg.Headers)
-	if len(traceId) == 0 {
-		return
-	}
-	trace.RunOnTracing(traceId, func(ctx context.Context) {
-		var message model.KqMessage
-		if err := json.Unmarshal(msg.Value, &message); err != nil {
-			logx.WithContext(ctx).Errorf("unmarshal msg error: %v", err)
-			return
-		}
-		trace.StartTrace(ctx, "InnerServer.handleMessage.SendMessage", func(ctx context.Context) {
-			if len(message.ReceiverId) == 0 || message.ReceiverId == "" {
-				// 玩家发的消息
-				p2cMap := ext.Game2PlayerMap.Get(message.GameId).(*treemap.Map)
-				message.ReceiverId = p2cMap.Get(message.SenderId).(string)
-				logx.WithContext(ctx).Infof("receiver: %s", message.ReceiverId)
-				kMsg, _ := json.Marshal(message)
-				s.svcCtx.KqMsgBoxProducer.SendMessage(ctx, string(kMsg), message.ReceiverId)
-			} else {
-				s.svcCtx.KqMsgBoxProducer.SendMessage(ctx, string(msg.Value), message.ReceiverId)
-			}
-			sess.MarkMessage(msg, "")
-		}, attribute.String("msg.key", string(msg.Key)))
-	})
+func (s *InnerServer) NotifyUserOnline(ctx context.Context, in *pb.NotifyUserStatusReq) (*pb.NotifyUserStatusResp, error) {
+	l := logic.NewNotifyUserOnlineLogic(ctx, s.svcCtx)
+	return l.NotifyUserOnline(in)
 }
 
-func (s *InnerServer) subscribe() {
-	go s.ConsumerGroup.RegisterHandleAndConsumer(s)
+func (s *InnerServer) NotifyUserOffline(ctx context.Context, in *pb.NotifyUserStatusReq) (*pb.NotifyUserStatusResp, error) {
+	l := logic.NewNotifyUserOfflineLogic(ctx, s.svcCtx)
+	return l.NotifyUserOffline(in)
 }

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

@@ -1,10 +1,16 @@
 package svc
 
 import (
+	"context"
+	"encoding/json"
+	"github.com/Shopify/sarama"
+	"github.com/liyue201/gostl/ds/list/simplelist"
 	treemap "github.com/liyue201/gostl/ds/map"
-	"github.com/liyue201/gostl/ds/set"
+	"github.com/zeromicro/go-zero/core/logx"
+	"go.opentelemetry.io/otel/attribute"
 	"ylink/comm/kafka"
 	"ylink/comm/model"
+	"ylink/comm/trace"
 	"ylink/core/inner/rpc/internal/config"
 	"ylink/core/inner/rpc/internal/ext"
 )
@@ -12,23 +18,113 @@ import (
 type ServiceContext struct {
 	Config           config.Config
 	KqMsgBoxProducer *kafka.Producer
+	ConsumerGroup    *kafka.ConsumerGroup
 }
 
 func NewServiceContext(c config.Config) *ServiceContext {
 	fetchCsCenterInfo()
-	ext.Game2PlayerStatMap = treemap.New(treemap.WithGoroutineSafe())
-	ext.CsStatSet = set.New(set.WithGoroutineSafe())
-	return &ServiceContext{
+	svcCtx := &ServiceContext{
 		Config:           c,
 		KqMsgBoxProducer: kafka.NewKafkaProducer(c.KqMsgBoxProducerConf.Brokers, c.KqMsgBoxProducerConf.Topic),
+		ConsumerGroup: kafka.NewConsumerGroup(&kafka.ConsumerGroupConfig{
+			KafkaVersion:   sarama.V1_0_0_0,
+			OffsetsInitial: sarama.OffsetNewest,
+			IsReturnErr:    false,
+		},
+			c.KqMsgBoxConsumerConf.Brokers,
+			[]string{c.KqMsgBoxConsumerConf.Topic},
+			c.KqMsgBoxConsumerConf.GroupId),
+	}
+	go svcCtx.subscribe()
+	go fetchCsCenterInfo()
+	return svcCtx
+}
+
+func (s *ServiceContext) Setup(_ sarama.ConsumerGroupSession) error {
+	return nil
+}
+
+func (s *ServiceContext) Cleanup(_ sarama.ConsumerGroupSession) error {
+	return nil
+}
+
+func (s *ServiceContext) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
+	for msg := range claim.Messages() {
+		if msg.Topic == s.Config.KqMsgBoxConsumerConf.Topic {
+			s.handleMessage(sess, msg)
+		}
+	}
+	return nil
+}
+
+func (s *ServiceContext) handleMessage(sess sarama.ConsumerGroupSession, msg *sarama.ConsumerMessage) {
+	traceId := kafka.GetTraceFromHeader(msg.Headers)
+	if len(traceId) == 0 {
+		return
 	}
+	trace.RunOnTracing(traceId, func(ctx context.Context) {
+		var message model.KqMessage
+		if err := json.Unmarshal(msg.Value, &message); err != nil {
+			logx.WithContext(ctx).Errorf("unmarshal msg error: %v", err)
+			return
+		}
+		logx.WithContext(ctx).Infof("handle message: %s", msg.Value)
+		trace.StartTrace(ctx, "InnerServer.handleMessage.SendMessage", func(ctx context.Context) {
+			if len(message.ReceiverId) == 0 || message.ReceiverId == "" {
+				// 玩家发的消息
+				p2cMap := ext.Game2PlayerMap.Get(message.GameId).(*treemap.Map)
+				message.ReceiverId = p2cMap.Get(message.SenderId).(string)
+				logx.WithContext(ctx).Infof("receiver: %s", message.ReceiverId)
+				kMsg, _ := json.Marshal(message)
+				s.KqMsgBoxProducer.SendMessage(ctx, string(kMsg), message.ReceiverId)
+			} else {
+				s.KqMsgBoxProducer.SendMessage(ctx, string(msg.Value), message.ReceiverId)
+			}
+			sess.MarkMessage(msg, "")
+		}, attribute.String("msg.key", string(msg.Key)))
+	})
+}
+
+func (s *ServiceContext) subscribe() {
+	go s.ConsumerGroup.RegisterHandleAndConsumer(s)
 }
 
 func fetchCsCenterInfo() {
 	// mock info
+	ext.Game2PlayerStatMap = treemap.New(treemap.WithGoroutineSafe())
+	ext.WaitingQueue = simplelist.New()
 	mockInfo()
 }
 
+func loadGameList() {
+
+}
+
+func loadCsInfo() {
+	ext.CsMap = treemap.New(treemap.WithGoroutineSafe())
+	ext.CsMap.Insert("cs_1231", &model.CsInfo{
+		CsId:         "cs_1231",
+		CsNickname:   "客服1231",
+		CsAvatarUrl:  "https://www.baidu.com",
+		CsSignature:  "我是客服1231",
+		OnlineStatus: 0,
+	})
+	ext.CsMap.Insert("cs_1111", &model.CsInfo{
+		CsId:         "cs_1111",
+		CsNickname:   "客服1111",
+		CsAvatarUrl:  "https://www.baidu.com",
+		CsSignature:  "我是客服1111",
+		OnlineStatus: 0,
+	})
+	ext.CsMap.Insert("cs_2222", &model.CsInfo{
+		CsId:         "cs_2222",
+		CsNickname:   "客服2222",
+		CsAvatarUrl:  "https://www.baidu.com",
+		CsSignature:  "我是客服2222",
+		OnlineStatus: 0,
+	})
+}
+
 func mockInfo() {
 	ext.Game2PlayerMap = treemap.New(treemap.WithGoroutineSafe())
 	ext.CsMap = treemap.New(treemap.WithGoroutineSafe())
@@ -51,14 +147,14 @@ func mockInfo() {
 		CsNickname:   "客服1231",
 		CsAvatarUrl:  "https://www.baidu.com",
 		CsSignature:  "我是客服1231",
-		OnlineStatus: 1,
+		OnlineStatus: 0,
 	})
 	ext.CsMap.Insert("cs_1111", &model.CsInfo{
 		CsId:         "cs_1111",
 		CsNickname:   "客服1111",
 		CsAvatarUrl:  "https://www.baidu.com",
 		CsSignature:  "我是客服1111",
-		OnlineStatus: 1,
+		OnlineStatus: 0,
 	})
 	ext.CsMap.Insert("cs_2222", &model.CsInfo{
 		CsId:         "cs_2222",

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

@@ -165,99 +165,6 @@ 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 {
@@ -271,7 +178,7 @@ type InnerCsFetchPlayerQueueReq struct {
 func (x *InnerCsFetchPlayerQueueReq) Reset() {
 	*x = InnerCsFetchPlayerQueueReq{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_pb_inner_proto_msgTypes[4]
+		mi := &file_pb_inner_proto_msgTypes[2]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -284,7 +191,7 @@ func (x *InnerCsFetchPlayerQueueReq) String() string {
 func (*InnerCsFetchPlayerQueueReq) ProtoMessage() {}
 
 func (x *InnerCsFetchPlayerQueueReq) ProtoReflect() protoreflect.Message {
-	mi := &file_pb_inner_proto_msgTypes[4]
+	mi := &file_pb_inner_proto_msgTypes[2]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -297,7 +204,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{4}
+	return file_pb_inner_proto_rawDescGZIP(), []int{2}
 }
 
 func (x *InnerCsFetchPlayerQueueReq) GetLimit() int64 {
@@ -319,7 +226,7 @@ type InnerCsFetchPlayerQueueResp struct {
 func (x *InnerCsFetchPlayerQueueResp) Reset() {
 	*x = InnerCsFetchPlayerQueueResp{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_pb_inner_proto_msgTypes[5]
+		mi := &file_pb_inner_proto_msgTypes[3]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -332,7 +239,7 @@ func (x *InnerCsFetchPlayerQueueResp) String() string {
 func (*InnerCsFetchPlayerQueueResp) ProtoMessage() {}
 
 func (x *InnerCsFetchPlayerQueueResp) ProtoReflect() protoreflect.Message {
-	mi := &file_pb_inner_proto_msgTypes[5]
+	mi := &file_pb_inner_proto_msgTypes[3]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -345,7 +252,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{5}
+	return file_pb_inner_proto_rawDescGZIP(), []int{3}
 }
 
 func (x *InnerCsFetchPlayerQueueResp) GetTotal() int64 {
@@ -375,7 +282,7 @@ type InnerCsConnectPlayerReq struct {
 func (x *InnerCsConnectPlayerReq) Reset() {
 	*x = InnerCsConnectPlayerReq{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_pb_inner_proto_msgTypes[6]
+		mi := &file_pb_inner_proto_msgTypes[4]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -388,7 +295,7 @@ func (x *InnerCsConnectPlayerReq) String() string {
 func (*InnerCsConnectPlayerReq) ProtoMessage() {}
 
 func (x *InnerCsConnectPlayerReq) ProtoReflect() protoreflect.Message {
-	mi := &file_pb_inner_proto_msgTypes[6]
+	mi := &file_pb_inner_proto_msgTypes[4]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -401,7 +308,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{6}
+	return file_pb_inner_proto_rawDescGZIP(), []int{4}
 }
 
 func (x *InnerCsConnectPlayerReq) GetCsId() string {
@@ -434,7 +341,7 @@ type InnerCsConnectPlayerResp struct {
 func (x *InnerCsConnectPlayerResp) Reset() {
 	*x = InnerCsConnectPlayerResp{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_pb_inner_proto_msgTypes[7]
+		mi := &file_pb_inner_proto_msgTypes[5]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -447,7 +354,7 @@ func (x *InnerCsConnectPlayerResp) String() string {
 func (*InnerCsConnectPlayerResp) ProtoMessage() {}
 
 func (x *InnerCsConnectPlayerResp) ProtoReflect() protoreflect.Message {
-	mi := &file_pb_inner_proto_msgTypes[7]
+	mi := &file_pb_inner_proto_msgTypes[5]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -460,10 +367,10 @@ func (x *InnerCsConnectPlayerResp) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use InnerCsConnectPlayerResp.ProtoReflect.Descriptor instead.
 func (*InnerCsConnectPlayerResp) Descriptor() ([]byte, []int) {
-	return file_pb_inner_proto_rawDescGZIP(), []int{7}
+	return file_pb_inner_proto_rawDescGZIP(), []int{5}
 }
 
-type UpdateUserStatusReq struct {
+type NotifyUserStatusReq struct {
 	state         protoimpl.MessageState
 	sizeCache     protoimpl.SizeCache
 	unknownFields protoimpl.UnknownFields
@@ -473,23 +380,23 @@ type UpdateUserStatusReq struct {
 	GameId string `protobuf:"bytes,3,opt,name=game_id,json=gameId,proto3" json:"game_id,omitempty"`
 }
 
-func (x *UpdateUserStatusReq) Reset() {
-	*x = UpdateUserStatusReq{}
+func (x *NotifyUserStatusReq) Reset() {
+	*x = NotifyUserStatusReq{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_pb_inner_proto_msgTypes[8]
+		mi := &file_pb_inner_proto_msgTypes[6]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
 }
 
-func (x *UpdateUserStatusReq) String() string {
+func (x *NotifyUserStatusReq) String() string {
 	return protoimpl.X.MessageStringOf(x)
 }
 
-func (*UpdateUserStatusReq) ProtoMessage() {}
+func (*NotifyUserStatusReq) ProtoMessage() {}
 
-func (x *UpdateUserStatusReq) ProtoReflect() protoreflect.Message {
-	mi := &file_pb_inner_proto_msgTypes[8]
+func (x *NotifyUserStatusReq) ProtoReflect() protoreflect.Message {
+	mi := &file_pb_inner_proto_msgTypes[6]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -500,55 +407,55 @@ func (x *UpdateUserStatusReq) ProtoReflect() protoreflect.Message {
 	return mi.MessageOf(x)
 }
 
-// Deprecated: Use UpdateUserStatusReq.ProtoReflect.Descriptor instead.
-func (*UpdateUserStatusReq) Descriptor() ([]byte, []int) {
-	return file_pb_inner_proto_rawDescGZIP(), []int{8}
+// Deprecated: Use NotifyUserStatusReq.ProtoReflect.Descriptor instead.
+func (*NotifyUserStatusReq) Descriptor() ([]byte, []int) {
+	return file_pb_inner_proto_rawDescGZIP(), []int{6}
 }
 
-func (x *UpdateUserStatusReq) GetType() int64 {
+func (x *NotifyUserStatusReq) GetType() int64 {
 	if x != nil {
 		return x.Type
 	}
 	return 0
 }
 
-func (x *UpdateUserStatusReq) GetUid() string {
+func (x *NotifyUserStatusReq) GetUid() string {
 	if x != nil {
 		return x.Uid
 	}
 	return ""
 }
 
-func (x *UpdateUserStatusReq) GetGameId() string {
+func (x *NotifyUserStatusReq) GetGameId() string {
 	if x != nil {
 		return x.GameId
 	}
 	return ""
 }
 
-type UpdateUserStatusResp struct {
+type NotifyUserStatusResp struct {
 	state         protoimpl.MessageState
 	sizeCache     protoimpl.SizeCache
 	unknownFields protoimpl.UnknownFields
 }
 
-func (x *UpdateUserStatusResp) Reset() {
-	*x = UpdateUserStatusResp{}
+func (x *NotifyUserStatusResp) Reset() {
+	*x = NotifyUserStatusResp{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_pb_inner_proto_msgTypes[9]
+		mi := &file_pb_inner_proto_msgTypes[7]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
 }
 
-func (x *UpdateUserStatusResp) String() string {
+func (x *NotifyUserStatusResp) String() string {
 	return protoimpl.X.MessageStringOf(x)
 }
 
-func (*UpdateUserStatusResp) ProtoMessage() {}
+func (*NotifyUserStatusResp) ProtoMessage() {}
 
-func (x *UpdateUserStatusResp) ProtoReflect() protoreflect.Message {
-	mi := &file_pb_inner_proto_msgTypes[9]
+func (x *NotifyUserStatusResp) ProtoReflect() protoreflect.Message {
+	mi := &file_pb_inner_proto_msgTypes[7]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -559,9 +466,9 @@ func (x *UpdateUserStatusResp) ProtoReflect() protoreflect.Message {
 	return mi.MessageOf(x)
 }
 
-// Deprecated: Use UpdateUserStatusResp.ProtoReflect.Descriptor instead.
-func (*UpdateUserStatusResp) Descriptor() ([]byte, []int) {
-	return file_pb_inner_proto_rawDescGZIP(), []int{9}
+// Deprecated: Use NotifyUserStatusResp.ProtoReflect.Descriptor instead.
+func (*NotifyUserStatusResp) Descriptor() ([]byte, []int) {
+	return file_pb_inner_proto_rawDescGZIP(), []int{7}
 }
 
 var File_pb_inner_proto protoreflect.FileDescriptor
@@ -588,49 +495,37 @@ 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, 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, 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, 0x55, 0x70, 0x64, 0x61, 0x74, 0x65, 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, 0x55,
-	0x70, 0x64, 0x61, 0x74, 0x65, 0x55, 0x73, 0x65, 0x72, 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x52,
-	0x65, 0x73, 0x70, 0x32, 0x98, 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,
+	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,
 	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,
@@ -641,12 +536,17 @@ var file_pb_inner_proto_rawDesc = []byte{
 	0x62, 0x2e, 0x49, 0x6e, 0x6e, 0x65, 0x72, 0x43, 0x73, 0x43, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74,
 	0x50, 0x6c, 0x61, 0x79, 0x65, 0x72, 0x52, 0x65, 0x71, 0x1a, 0x1c, 0x2e, 0x70, 0x62, 0x2e, 0x49,
 	0x6e, 0x6e, 0x65, 0x72, 0x43, 0x73, 0x43, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x50, 0x6c, 0x61,
-	0x79, 0x65, 0x72, 0x52, 0x65, 0x73, 0x70, 0x12, 0x45, 0x0a, 0x10, 0x75, 0x70, 0x64, 0x61, 0x74,
-	0x65, 0x55, 0x73, 0x65, 0x72, 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x12, 0x17, 0x2e, 0x70, 0x62,
-	0x2e, 0x55, 0x70, 0x64, 0x61, 0x74, 0x65, 0x55, 0x73, 0x65, 0x72, 0x53, 0x74, 0x61, 0x74, 0x75,
-	0x73, 0x52, 0x65, 0x71, 0x1a, 0x18, 0x2e, 0x70, 0x62, 0x2e, 0x55, 0x70, 0x64, 0x61, 0x74, 0x65,
-	0x55, 0x73, 0x65, 0x72, 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x52, 0x65, 0x73, 0x70, 0x42, 0x06,
-	0x5a, 0x04, 0x2e, 0x2f, 0x70, 0x62, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+	0x79, 0x65, 0x72, 0x52, 0x65, 0x73, 0x70, 0x12, 0x45, 0x0a, 0x10, 0x6e, 0x6f, 0x74, 0x69, 0x66,
+	0x79, 0x55, 0x73, 0x65, 0x72, 0x4f, 0x6e, 0x6c, 0x69, 0x6e, 0x65, 0x12, 0x17, 0x2e, 0x70, 0x62,
+	0x2e, 0x4e, 0x6f, 0x74, 0x69, 0x66, 0x79, 0x55, 0x73, 0x65, 0x72, 0x53, 0x74, 0x61, 0x74, 0x75,
+	0x73, 0x52, 0x65, 0x71, 0x1a, 0x18, 0x2e, 0x70, 0x62, 0x2e, 0x4e, 0x6f, 0x74, 0x69, 0x66, 0x79,
+	0x55, 0x73, 0x65, 0x72, 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x52, 0x65, 0x73, 0x70, 0x12, 0x46,
+	0x0a, 0x11, 0x6e, 0x6f, 0x74, 0x69, 0x66, 0x79, 0x55, 0x73, 0x65, 0x72, 0x4f, 0x66, 0x66, 0x6c,
+	0x69, 0x6e, 0x65, 0x12, 0x17, 0x2e, 0x70, 0x62, 0x2e, 0x4e, 0x6f, 0x74, 0x69, 0x66, 0x79, 0x55,
+	0x73, 0x65, 0x72, 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x52, 0x65, 0x71, 0x1a, 0x18, 0x2e, 0x70,
+	0x62, 0x2e, 0x4e, 0x6f, 0x74, 0x69, 0x66, 0x79, 0x55, 0x73, 0x65, 0x72, 0x53, 0x74, 0x61, 0x74,
+	0x75, 0x73, 0x52, 0x65, 0x73, 0x70, 0x42, 0x06, 0x5a, 0x04, 0x2e, 0x2f, 0x70, 0x62, 0x62, 0x06,
+	0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
 }
 
 var (
@@ -661,37 +561,35 @@ func file_pb_inner_proto_rawDescGZIP() []byte {
 	return file_pb_inner_proto_rawDescData
 }
 
-var file_pb_inner_proto_msgTypes = make([]protoimpl.MessageInfo, 10)
+var file_pb_inner_proto_msgTypes = make([]protoimpl.MessageInfo, 8)
 var file_pb_inner_proto_goTypes = []interface{}{
 	(*InnerPlayerFetchCsInfoReq)(nil),   // 0: pb.InnerPlayerFetchCsInfoReq
 	(*InnerPlayerFetchCsInfoResp)(nil),  // 1: pb.InnerPlayerFetchCsInfoResp
-	(*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
-	(*UpdateUserStatusReq)(nil),         // 8: pb.UpdateUserStatusReq
-	(*UpdateUserStatusResp)(nil),        // 9: pb.UpdateUserStatusResp
-	(*structpb.ListValue)(nil),          // 10: google.protobuf.ListValue
+	(*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
 }
 var file_pb_inner_proto_depIdxs = []int32{
-	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.updateUserStatus:input_type -> pb.UpdateUserStatusReq
-	1,  // 6: pb.Inner.playerFetchCsInfo:output_type -> pb.InnerPlayerFetchCsInfoResp
-	3,  // 7: pb.Inner.playerDisconnect:output_type -> pb.InnerPlayerDisconnectResp
-	5,  // 8: pb.Inner.csFetchPlayerQueue:output_type -> pb.InnerCsFetchPlayerQueueResp
-	7,  // 9: pb.Inner.csConnectPlayer:output_type -> pb.InnerCsConnectPlayerResp
-	9,  // 10: pb.Inner.updateUserStatus:output_type -> pb.UpdateUserStatusResp
-	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
+	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
 }
 
 func init() { file_pb_inner_proto_init() }
@@ -725,7 +623,7 @@ func file_pb_inner_proto_init() {
 			}
 		}
 		file_pb_inner_proto_msgTypes[2].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*InnerPlayerDisconnectReq); i {
+			switch v := v.(*InnerCsFetchPlayerQueueReq); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -737,7 +635,7 @@ func file_pb_inner_proto_init() {
 			}
 		}
 		file_pb_inner_proto_msgTypes[3].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*InnerPlayerDisconnectResp); i {
+			switch v := v.(*InnerCsFetchPlayerQueueResp); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -749,7 +647,7 @@ func file_pb_inner_proto_init() {
 			}
 		}
 		file_pb_inner_proto_msgTypes[4].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*InnerCsFetchPlayerQueueReq); i {
+			switch v := v.(*InnerCsConnectPlayerReq); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -761,7 +659,7 @@ func file_pb_inner_proto_init() {
 			}
 		}
 		file_pb_inner_proto_msgTypes[5].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*InnerCsFetchPlayerQueueResp); i {
+			switch v := v.(*InnerCsConnectPlayerResp); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -773,7 +671,7 @@ func file_pb_inner_proto_init() {
 			}
 		}
 		file_pb_inner_proto_msgTypes[6].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*InnerCsConnectPlayerReq); i {
+			switch v := v.(*NotifyUserStatusReq); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -785,31 +683,7 @@ 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.(*UpdateUserStatusReq); 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.(*UpdateUserStatusResp); i {
+			switch v := v.(*NotifyUserStatusResp); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -827,7 +701,7 @@ func file_pb_inner_proto_init() {
 			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
 			RawDescriptor: file_pb_inner_proto_rawDesc,
 			NumEnums:      0,
-			NumMessages:   10,
+			NumMessages:   8,
 			NumExtensions: 0,
 			NumServices:   1,
 		},

core/inner/rpc/pb/inner.proto

@@ -23,13 +23,6 @@ message InnerPlayerFetchCsInfoResp{
   int64 online_status = 5;
 }
 
-message InnerPlayerDisconnectReq{
-  string player_id = 1;
-  string game_id = 2;
-}
-
-message InnerPlayerDisconnectResp{}
-
 /**
 Cs Command Request Bean
  */
@@ -50,20 +43,22 @@ message InnerCsConnectPlayerReq{
 
 message InnerCsConnectPlayerResp{}
 
-message UpdateUserStatusReq{
+
+message NotifyUserStatusReq{
   int64 type = 1;
   string uid = 2;
   string game_id = 3;
 }
 
-message UpdateUserStatusResp{}
+message NotifyUserStatusResp{}
 
 service Inner {
   rpc playerFetchCsInfo (InnerPlayerFetchCsInfoReq) returns (InnerPlayerFetchCsInfoResp);
-  rpc playerDisconnect (InnerPlayerDisconnectReq) returns (InnerPlayerDisconnectResp);
 
   rpc csFetchPlayerQueue (InnerCsFetchPlayerQueueReq) returns (InnerCsFetchPlayerQueueResp);
   rpc csConnectPlayer (InnerCsConnectPlayerReq) returns (InnerCsConnectPlayerResp);
 
-  rpc updateUserStatus (UpdateUserStatusReq) returns (UpdateUserStatusResp);
+  rpc notifyUserOnline (NotifyUserStatusReq) returns (NotifyUserStatusResp);
+  rpc notifyUserOffline (NotifyUserStatusReq) returns (NotifyUserStatusResp);
+
 }

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

@@ -23,10 +23,10 @@ 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)
-	UpdateUserStatus(ctx context.Context, in *UpdateUserStatusReq, opts ...grpc.CallOption) (*UpdateUserStatusResp, error)
+	NotifyUserOnline(ctx context.Context, in *NotifyUserStatusReq, opts ...grpc.CallOption) (*NotifyUserStatusResp, error)
+	NotifyUserOffline(ctx context.Context, in *NotifyUserStatusReq, opts ...grpc.CallOption) (*NotifyUserStatusResp, error)
 }
 
 type innerClient struct {
@@ -46,15 +46,6 @@ 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...)
@@ -73,9 +64,18 @@ func (c *innerClient) CsConnectPlayer(ctx context.Context, in *InnerCsConnectPla
 	return out, nil
 }
 
-func (c *innerClient) UpdateUserStatus(ctx context.Context, in *UpdateUserStatusReq, opts ...grpc.CallOption) (*UpdateUserStatusResp, error) {
-	out := new(UpdateUserStatusResp)
-	err := c.cc.Invoke(ctx, "/pb.Inner/updateUserStatus", in, out, opts...)
+func (c *innerClient) NotifyUserOnline(ctx context.Context, in *NotifyUserStatusReq, opts ...grpc.CallOption) (*NotifyUserStatusResp, error) {
+	out := new(NotifyUserStatusResp)
+	err := c.cc.Invoke(ctx, "/pb.Inner/notifyUserOnline", in, out, opts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *innerClient) NotifyUserOffline(ctx context.Context, in *NotifyUserStatusReq, opts ...grpc.CallOption) (*NotifyUserStatusResp, error) {
+	out := new(NotifyUserStatusResp)
+	err := c.cc.Invoke(ctx, "/pb.Inner/notifyUserOffline", in, out, opts...)
 	if err != nil {
 		return nil, err
 	}
@@ -87,10 +87,10 @@ func (c *innerClient) UpdateUserStatus(ctx context.Context, in *UpdateUserStatus
 // 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)
-	UpdateUserStatus(context.Context, *UpdateUserStatusReq) (*UpdateUserStatusResp, error)
+	NotifyUserOnline(context.Context, *NotifyUserStatusReq) (*NotifyUserStatusResp, error)
+	NotifyUserOffline(context.Context, *NotifyUserStatusReq) (*NotifyUserStatusResp, error)
 	mustEmbedUnimplementedInnerServer()
 }
 
@@ -101,17 +101,17 @@ 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")
 }
 func (UnimplementedInnerServer) CsConnectPlayer(context.Context, *InnerCsConnectPlayerReq) (*InnerCsConnectPlayerResp, error) {
 	return nil, status.Errorf(codes.Unimplemented, "method CsConnectPlayer not implemented")
 }
-func (UnimplementedInnerServer) UpdateUserStatus(context.Context, *UpdateUserStatusReq) (*UpdateUserStatusResp, error) {
-	return nil, status.Errorf(codes.Unimplemented, "method UpdateUserStatus not implemented")
+func (UnimplementedInnerServer) NotifyUserOnline(context.Context, *NotifyUserStatusReq) (*NotifyUserStatusResp, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method NotifyUserOnline not implemented")
+}
+func (UnimplementedInnerServer) NotifyUserOffline(context.Context, *NotifyUserStatusReq) (*NotifyUserStatusResp, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method NotifyUserOffline not implemented")
 }
 func (UnimplementedInnerServer) mustEmbedUnimplementedInnerServer() {}
 
@@ -144,74 +144,74 @@ 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)
+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 {
 		return nil, err
 	}
 	if interceptor == nil {
-		return srv.(InnerServer).PlayerDisconnect(ctx, in)
+		return srv.(InnerServer).CsFetchPlayerQueue(ctx, in)
 	}
 	info := &grpc.UnaryServerInfo{
 		Server:     srv,
-		FullMethod: "/pb.Inner/playerDisconnect",
+		FullMethod: "/pb.Inner/csFetchPlayerQueue",
 	}
 	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
-		return srv.(InnerServer).PlayerDisconnect(ctx, req.(*InnerPlayerDisconnectReq))
+		return srv.(InnerServer).CsFetchPlayerQueue(ctx, req.(*InnerCsFetchPlayerQueueReq))
 	}
 	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)
+func _Inner_CsConnectPlayer_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(InnerCsConnectPlayerReq)
 	if err := dec(in); err != nil {
 		return nil, err
 	}
 	if interceptor == nil {
-		return srv.(InnerServer).CsFetchPlayerQueue(ctx, in)
+		return srv.(InnerServer).CsConnectPlayer(ctx, in)
 	}
 	info := &grpc.UnaryServerInfo{
 		Server:     srv,
-		FullMethod: "/pb.Inner/csFetchPlayerQueue",
+		FullMethod: "/pb.Inner/csConnectPlayer",
 	}
 	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
-		return srv.(InnerServer).CsFetchPlayerQueue(ctx, req.(*InnerCsFetchPlayerQueueReq))
+		return srv.(InnerServer).CsConnectPlayer(ctx, req.(*InnerCsConnectPlayerReq))
 	}
 	return interceptor(ctx, in, info, handler)
 }
 
-func _Inner_CsConnectPlayer_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
-	in := new(InnerCsConnectPlayerReq)
+func _Inner_NotifyUserOnline_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(NotifyUserStatusReq)
 	if err := dec(in); err != nil {
 		return nil, err
 	}
 	if interceptor == nil {
-		return srv.(InnerServer).CsConnectPlayer(ctx, in)
+		return srv.(InnerServer).NotifyUserOnline(ctx, in)
 	}
 	info := &grpc.UnaryServerInfo{
 		Server:     srv,
-		FullMethod: "/pb.Inner/csConnectPlayer",
+		FullMethod: "/pb.Inner/notifyUserOnline",
 	}
 	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
-		return srv.(InnerServer).CsConnectPlayer(ctx, req.(*InnerCsConnectPlayerReq))
+		return srv.(InnerServer).NotifyUserOnline(ctx, req.(*NotifyUserStatusReq))
 	}
 	return interceptor(ctx, in, info, handler)
 }
 
-func _Inner_UpdateUserStatus_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
-	in := new(UpdateUserStatusReq)
+func _Inner_NotifyUserOffline_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(NotifyUserStatusReq)
 	if err := dec(in); err != nil {
 		return nil, err
 	}
 	if interceptor == nil {
-		return srv.(InnerServer).UpdateUserStatus(ctx, in)
+		return srv.(InnerServer).NotifyUserOffline(ctx, in)
 	}
 	info := &grpc.UnaryServerInfo{
 		Server:     srv,
-		FullMethod: "/pb.Inner/updateUserStatus",
+		FullMethod: "/pb.Inner/notifyUserOffline",
 	}
 	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
-		return srv.(InnerServer).UpdateUserStatus(ctx, req.(*UpdateUserStatusReq))
+		return srv.(InnerServer).NotifyUserOffline(ctx, req.(*NotifyUserStatusReq))
 	}
 	return interceptor(ctx, in, info, handler)
 }
@@ -227,10 +227,6 @@ var Inner_ServiceDesc = grpc.ServiceDesc{
 			MethodName: "playerFetchCsInfo",
 			Handler:    _Inner_PlayerFetchCsInfo_Handler,
 		},
-		{
-			MethodName: "playerDisconnect",
-			Handler:    _Inner_PlayerDisconnect_Handler,
-		},
 		{
 			MethodName: "csFetchPlayerQueue",
 			Handler:    _Inner_CsFetchPlayerQueue_Handler,
@@ -240,8 +236,12 @@ var Inner_ServiceDesc = grpc.ServiceDesc{
 			Handler:    _Inner_CsConnectPlayer_Handler,
 		},
 		{
-			MethodName: "updateUserStatus",
-			Handler:    _Inner_UpdateUserStatus_Handler,
+			MethodName: "notifyUserOnline",
+			Handler:    _Inner_NotifyUserOnline_Handler,
+		},
+		{
+			MethodName: "notifyUserOffline",
+			Handler:    _Inner_NotifyUserOffline_Handler,
 		},
 	},
 	Streams:  []grpc.StreamDesc{},

db/pb/db.proto

@@ -1,8 +0,0 @@
-syntax = "proto3";
-
-option go_package = "./pb";
-
-package pb;
-
-service Db {
-}

flowsrv/rpc/internal/logic/connectlogic.go

@@ -39,7 +39,7 @@ func (l *ConnectLogic) Connect(in *pb.CommandReq, stream pb.Flowsrv_ConnectServe
 			Data: nil,
 		})
 	}
-	_, err = l.svcCtx.InnerRpc.UpdateUserStatus(l.ctx, &inner.UpdateUserStatusReq{
+	_, err = l.svcCtx.InnerRpc.NotifyUserOnline(l.ctx, &inner.NotifyUserStatusReq{
 		Type:   in.Type,
 		Uid:    uid,
 		GameId: gameId,

flowsrv/rpc/internal/logic/disconnectlogic.go

@@ -39,7 +39,7 @@ func (l *DisconnectLogic) Disconnect(in *pb.CommandReq) (*pb.CommandResp, error)
 			Data: nil,
 		}, err
 	}
-	_, err = l.svcCtx.InnerRpc.UpdateUserStatus(l.ctx, &inner.UpdateUserStatusReq{
+	_, err = l.svcCtx.InnerRpc.NotifyUserOffline(l.ctx, &inner.NotifyUserStatusReq{
 		Type:   in.Type,
 		Uid:    uid,
 		GameId: gameId,
@@ -53,7 +53,7 @@ func (l *DisconnectLogic) Disconnect(in *pb.CommandReq) (*pb.CommandResp, error)
 	}
 
 	mgr.GetFlowMgrInstance().RemoveFlow(uid)
-	
+
 	return &pb.CommandResp{
 		Code: result.Ok,
 		Msg:  "success",

go.mod

@@ -5,8 +5,11 @@ go 1.18
 require (
 	github.com/Shopify/sarama v1.32.0
 	github.com/golang-jwt/jwt/v4 v4.4.1
+	github.com/liyue201/gostl v1.0.1
 	github.com/pkg/errors v0.9.1
 	github.com/zeromicro/go-zero v1.3.3
+	go.opentelemetry.io/otel v1.3.0
+	go.opentelemetry.io/otel/trace v1.3.0
 	google.golang.org/grpc v1.46.0
 	google.golang.org/protobuf v1.28.0
 )
@@ -42,13 +45,11 @@ require (
 	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/justinas/alice v1.2.0 // indirect
 	github.com/klauspost/compress v1.14.4 // indirect
-	github.com/liyue201/gostl v1.0.1 // indirect
 	github.com/matttproud/golang_protobuf_extensions v1.0.2-0.20181231171920-c182affec369 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/openzipkin/zipkin-go v0.4.0 // indirect
 	github.com/pierrec/lz4 v2.6.1+incompatible // indirect
-	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/prometheus/client_golang v1.11.0 // indirect
 	github.com/prometheus/client_model v0.2.0 // indirect
 	github.com/prometheus/common v0.30.0 // indirect
@@ -58,11 +59,9 @@ require (
 	go.etcd.io/etcd/api/v3 v3.5.2 // indirect
 	go.etcd.io/etcd/client/pkg/v3 v3.5.2 // indirect
 	go.etcd.io/etcd/client/v3 v3.5.2 // indirect
-	go.opentelemetry.io/otel v1.3.0 // indirect
 	go.opentelemetry.io/otel/exporters/jaeger v1.3.0 // indirect
 	go.opentelemetry.io/otel/exporters/zipkin v1.3.0 // indirect
 	go.opentelemetry.io/otel/sdk v1.3.0 // indirect
-	go.opentelemetry.io/otel/trace v1.3.0 // indirect
 	go.uber.org/atomic v1.9.0 // indirect
 	go.uber.org/automaxprocs v1.4.0 // indirect
 	go.uber.org/multierr v1.8.0 // indirect
@@ -78,7 +77,6 @@ require (
 	google.golang.org/genproto v0.0.0-20220422154200-b37d22cd5731 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
-	gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b // indirect
 	k8s.io/api v0.20.12 // indirect
 	k8s.io/apimachinery v0.20.12 // indirect
 	k8s.io/client-go v0.20.12 // indirect

go.sum

@@ -71,8 +71,6 @@ github.com/beorn7/perks v1.0.0/go.mod h1:KWe93zE9D1o94FZ5RNwFwVgaQK1VOXiVxmqh+Ce
 github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
 github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
 github.com/bkaradzic/go-lz4 v1.0.0/go.mod h1:0YdlkowM3VswSROI7qDxhRvJ3sLhlFrRRwjwegp5jy4=
-github.com/bytedance/gopkg v0.0.0-20220118071334-3db87571198b h1:LTGVFpNmNHhj0vhOlfgWueFJ32eK9blaIlHR2ciXOT0=
-github.com/bytedance/gopkg v0.0.0-20220118071334-3db87571198b/go.mod h1:2ZlV9BaUH4+NXIBF0aMdKKAnHTzqH+iMU4KUjAbL23Q=
 github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
 github.com/cespare/xxhash/v2 v2.1.1/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/cespare/xxhash/v2 v2.1.2 h1:YRXhKfTDauu4ajMg1TPgFO5jnlC2HCbmLXMcTG5cbYE=
@@ -103,8 +101,6 @@ github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/r
 github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc=
 github.com/docker/spdystream v0.0.0-20160310174837-449fdfce4d96/go.mod h1:Qh8CwZgvJUkLughtfhJv5dyTYa91l1fOUCrgjqmcifM=
 github.com/docopt/docopt-go v0.0.0-20180111231733-ee0de3bc6815/go.mod h1:WwZ+bS3ebgob9U8Nd0kOddGdZWjyMGR8Wziv+TBNwSE=
-github.com/duke-git/lancet/v2 v2.0.6 h1:HZKwz3Lcslh1wKYscKy21MhF1JOFrF9bbE1mKM54P/s=
-github.com/duke-git/lancet/v2 v2.0.6/go.mod h1:5Nawyf/bK783rCiHyVkZLx+jj8028oVVjLOrC21ZONA=
 github.com/dustin/go-humanize v1.0.0/go.mod h1:HtrtbFcZ19U5GC7JDqmcUSB87Iq5E25KnS6fMYU6eOk=
 github.com/eapache/go-resiliency v1.2.0 h1:v7g92e/KSN71Rq7vSThKaWIq68fL4YHvWyiUKorFR1Q=
 github.com/eapache/go-resiliency v1.2.0/go.mod h1:kFI+JgMyC7bLPUVY133qvEBtVayf5mFgVsvEsIPBvNs=
@@ -626,7 +622,6 @@ golang.org/x/sys v0.0.0-20210510120138-977fb7262007/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20210603081109-ebe580a85c40/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20211216021012-1d35b9e2eb4e/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.0.0-20220110181412-a018aaa089fe/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220422013727-9388b58f7150 h1:xHms4gcpe1YE7A3yIllJXP16CMAGuqwO2lX1mTyyRRc=
 golang.org/x/sys v0.0.0-20220422013727-9388b58f7150/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201117132131-f5c789dd3221/go.mod h1:Nr5EML6q2oocZ2LXRh80K7BxOlk5/8JxuGnuhpl+muw=