connectlogic.go 2.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113
  1. package logic
  2. import (
  3. "context"
  4. "github.com/golang-jwt/jwt/v4"
  5. "github.com/pkg/errors"
  6. "ylink/comm/globalkey"
  7. "ylink/comm/jwtkey"
  8. "ylink/comm/result"
  9. "ylink/core/inner/rpc/inner"
  10. "ylink/flowsrv/rpc/internal/mgr"
  11. "ylink/flowsrv/rpc/internal/model"
  12. "ylink/flowsrv/rpc/internal/svc"
  13. "ylink/flowsrv/rpc/pb"
  14. "github.com/zeromicro/go-zero/core/logx"
  15. )
  16. type ConnectLogic struct {
  17. ctx context.Context
  18. svcCtx *svc.ServiceContext
  19. logx.Logger
  20. }
  21. func NewConnectLogic(ctx context.Context, svcCtx *svc.ServiceContext) *ConnectLogic {
  22. return &ConnectLogic{
  23. ctx: ctx,
  24. svcCtx: svcCtx,
  25. Logger: logx.WithContext(ctx),
  26. }
  27. }
  28. func (l *ConnectLogic) Connect(in *pb.CommandReq, stream pb.Flowsrv_ConnectServer) error {
  29. uid, gameId, err := l.checkAuth(in)
  30. if err != nil {
  31. return stream.Send(&pb.CommandResp{
  32. Code: result.TokenParseError,
  33. Msg: err.Error(),
  34. Data: nil,
  35. })
  36. }
  37. _, err = l.svcCtx.InnerRpc.NotifyUserOnline(l.ctx, &inner.NotifyUserStatusReq{
  38. Type: in.Type,
  39. Uid: uid,
  40. GameId: gameId,
  41. })
  42. if err != nil {
  43. return stream.Send(&pb.CommandResp{
  44. Code: result.ServerCommonError,
  45. Msg: err.Error(),
  46. Data: nil,
  47. })
  48. }
  49. var flowId string
  50. if in.Type == globalkey.ConnectTypePlayer {
  51. flowId = gameId + "_" + uid
  52. } else {
  53. flowId = uid
  54. }
  55. flow := &model.Flow{
  56. EndFlow: make(chan int),
  57. Message: make(chan string),
  58. Stream: stream,
  59. SvcCtx: l.svcCtx,
  60. Logger: l.Logger,
  61. Type: in.Type,
  62. Uid: uid,
  63. GameId: gameId,
  64. FlowId: flowId,
  65. }
  66. defer func() {
  67. close(flow.EndFlow)
  68. flow = nil
  69. }()
  70. mgr.GetFlowMgrInstance().Register(flow)
  71. <-flow.EndFlow
  72. return nil
  73. }
  74. func (l *ConnectLogic) checkAuth(in *pb.CommandReq) (string, string, error) {
  75. token, err := jwt.Parse(in.AccessToken, func(token *jwt.Token) (i interface{}, err error) {
  76. return []byte(l.svcCtx.Config.JwtAuth.AccessSecret), nil
  77. })
  78. uid := ""
  79. gameId := ""
  80. if token.Valid {
  81. //将获取的token中的Claims强转为MapClaims
  82. claims, _ := token.Claims.(jwt.MapClaims)
  83. if in.Type == globalkey.ConnectTypePlayer {
  84. uid = claims[jwtkey.PlayerId].(string)
  85. gameId = claims[jwtkey.GameId].(string)
  86. } else {
  87. uid = claims[jwtkey.CsId].(string)
  88. }
  89. return uid, gameId, nil
  90. } else if ve, ok := err.(*jwt.ValidationError); ok {
  91. if ve.Errors&jwt.ValidationErrorMalformed != 0 {
  92. return uid, gameId, errors.Wrap(result.NewErrCode(result.TokenParseError), "")
  93. } else if ve.Errors&(jwt.ValidationErrorExpired|jwt.ValidationErrorNotValidYet) != 0 {
  94. // Token is either expired or not active yet
  95. return uid, gameId, errors.Wrap(result.NewErrCode(result.TokenExpireError), "")
  96. } else {
  97. return uid, gameId, errors.Wrap(result.NewErrCode(result.TokenParseError), "")
  98. }
  99. } else {
  100. return uid, gameId, errors.Wrap(result.NewErrCode(result.TokenParseError), "")
  101. }
  102. }