blackteay/open-im-server
1
0
Fork 0
mirror of https://github.com/openimsdk/open-im-server.git synced 2025-04-06 04:15:46 +08:00
Code Issues Packages Projects Releases Wiki Activity

rpc mw

Browse Source
This commit is contained in:
withchao 2023-03-07 16:57:49 +08:00
parent 555fc52acc
commit 102ab98276
17 changed files with 139 additions and 276 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
internal/api/auth.go
View File

@ -20,7 +20,7 @@ type Auth struct {
}
func (o *Auth) client() (auth.AuthClient, error) {
conn, err := o.zk.GetConn(config.Config.RpcRegisterName.OpenImGroupName, nil)
conn, err := o.zk.GetDefaultConn(config.Config.RpcRegisterName.OpenImAuthName)
if err != nil {
return nil, err
}

2
internal/api/conversation.go
View File

@ -20,7 +20,7 @@ type Conversation struct {
}
func (o *Conversation) client() (conversation.ConversationClient, error) {
conn, err := o.zk.GetConn(config.Config.RpcRegisterName.OpenImGroupName)
conn, err := o.zk.GetDefaultConn(config.Config.RpcRegisterName.OpenImConversationName)
if err != nil {
return nil, err
}

2
internal/api/friend.go
View File

@ -20,7 +20,7 @@ type Friend struct {
}
func (o *Friend) client() (friend.FriendClient, error) {
conn, err := o.zk.GetConn(config.Config.RpcRegisterName.OpenImGroupName)
conn, err := o.zk.GetDefaultConn(config.Config.RpcRegisterName.OpenImFriendName)
if err != nil {
return nil, err
}

2
internal/api/group.go
View File

@ -20,7 +20,7 @@ type Group struct {
}
func (o *Group) client() (group.GroupClient, error) {
conn, err := o.zk.GetConn(config.Config.RpcRegisterName.OpenImGroupName)
conn, err := o.zk.GetDefaultConn(config.Config.RpcRegisterName.OpenImGroupName)
if err != nil {
return nil, err
}

2
internal/api/msg.go
View File

@ -107,7 +107,7 @@ func newUserSendMsgReq(params *apistruct.ManagementSendMsgReq) *msg.SendMsgReq {
}
func (o *Msg) client() (msg.MsgClient, error) {
conn, err := o.zk.GetConn(config.Config.RpcRegisterName.OpenImMsgName)
conn, err := o.zk.GetDefaultConn(config.Config.RpcRegisterName.OpenImMsgName)
if err != nil {
return nil, err
}

29
internal/api/route.go
View File

@ -3,23 +3,24 @@ package api
import (
"OpenIM/pkg/common/config"
"OpenIM/pkg/common/log"
"OpenIM/pkg/common/middleware"
"OpenIM/pkg/common/mw"
"OpenIM/pkg/common/prome"
"github.com/OpenIMSDK/openKeeper"
"github.com/gin-gonic/gin"
"google.golang.org/grpc"
"io"
"os"
)
func NewGinRouter() *gin.Engine {
openKeeper.DefaultOptions = []grpc.DialOption{mw.GrpcClient()} // 默认RPC中间件
gin.SetMode(gin.ReleaseMode)
f, _ := os.Create("../logs/api.log")
gin.DefaultWriter = io.MultiWriter(f)
// gin.SetMode(gin.DebugMode)
r := gin.New()
r.Use(gin.Recovery())
r.Use(middleware.GinParseOperationID)
log.Info("load config: ", config.Config)
r.Use(gin.Recovery(), mw.CorsHandler(), mw.GinParseOperationID())
if config.Config.Prometheus.Enable {
prome.NewApiRequestCounter()
prome.NewApiRequestFailedCounter()
@ -148,25 +149,3 @@ func NewGinRouter() *gin.Engine {
}
return r
}
/*
{
GetSeq
SendMsg
PullMsgBySeqList
DelMsg
DelSuperGroupMsg
ClearMsg
SetMsgMinSeq
SetMessageReactionExtensions
GetMessageListReactionExtensions
AddMessageReactionExtensions
DeleteMessageReactionExtensions
ManagementSendMsg
ManagementBatchSendMsg
CheckMsgIsSendSuccess
}
*/

2
internal/api/third.go
View File

@ -20,7 +20,7 @@ type Third struct {
}
func (o *Third) client() (third.ThirdClient, error) {
conn, err := o.zk.GetConn(config.Config.RpcRegisterName.OpenImThirdName)
conn, err := o.zk.GetDefaultConn(config.Config.RpcRegisterName.OpenImThirdName)
if err != nil {
return nil, err
}

3
internal/api/user.go
View File

@ -20,7 +20,7 @@ type User struct {
}
func (o *User) client() (user.UserClient, error) {
conn, err := o.zk.GetConn(config.Config.RpcRegisterName.OpenImGroupName)
conn, err := o.zk.GetDefaultConn(config.Config.RpcRegisterName.OpenImUserName)
if err != nil {
return nil, err
}
@ -47,7 +47,6 @@ func (o *User) GetAllUsersID(c *gin.Context) {
a2r.Call(user.UserClient.GetDesignateUsers, o.client, c)
}
//
func (u *User) AccountCheck(c *gin.Context) {
a2r.Call(user.UserClient.AccountCheck, u.client, c)
}

4
internal/startrpc/start.go
View File

@ -5,7 +5,7 @@ import (
"OpenIM/pkg/common/config"
"OpenIM/pkg/common/constant"
"OpenIM/pkg/common/log"
"OpenIM/pkg/common/middleware"
"OpenIM/pkg/common/mw"
"OpenIM/pkg/common/prome"
"OpenIM/pkg/discoveryregistry"
"flag"
@ -36,7 +36,7 @@ func start(rpcPort int, rpcRegisterName string, prometheusPorts int, rpcFn func(
if err != nil {
return err
}
options = append(options, grpc.UnaryInterceptor(middleware.RpcServerInterceptor)) // ctx 中间件
options = append(options, mw.GrpcServer()) // ctx 中间件
if config.Config.Prometheus.Enable {
prome.NewGrpcRequestCounter()
prome.NewGrpcRequestFailedCounter()

3
pkg/callbackstruct/common.go
View File

@ -1,7 +1,6 @@
package callbackstruct
import (
"OpenIM/pkg/common/constant"
"OpenIM/pkg/errs"
"fmt"
)
@ -45,7 +44,7 @@ type CommonCallbackResp struct {
}
func (c CommonCallbackResp) Parse() error {
if c.ActionCode != constant.NoError || c.ErrCode != constant.NoError {
if c.ActionCode != errs.NoError || c.ErrCode != errs.NoError {
return errs.NewCodeError(int(c.ErrCode), "Callback").Wrap(fmt.Sprintf("callback response error actionCode is %d, errCode is %d, errMsg is %s", c.ActionCode, c.ErrCode, c.ErrMsg))
}
return nil

167
pkg/common/constant/errors.go
View File

@ -1,167 +0,0 @@
package constant
var (
ErrNone = &ErrInfo{0, "", ""}
ErrArgs = &ErrInfo{ArgsError, "ArgsError", ""}
ErrDatabase = &ErrInfo{DatabaseError, "DatabaseError", ""}
ErrInternalServer = &ErrInfo{ServerInternalError, "ServerInternalError", ""}
ErrNetwork = &ErrInfo{NetworkError, "NetworkError", ""}
ErrNoPermission = &ErrInfo{NoPermissionError, "NoPermissionError", ""}
ErrIdentity = &ErrInfo{IdentityError, "IdentityError", ""}
ErrCallback = &ErrInfo{CallbackError, "CallbackError", ""}
ErrCallbackContinue = &ErrInfo{ErrMsg: "CallbackContinueError"}
ErrUserIDNotFound = &ErrInfo{UserIDNotFoundError, "UserIDNotFoundError", ""}
ErrGroupIDNotFound = &ErrInfo{GroupIDNotFoundError, "GroupIDNotFoundError", ""}
ErrGroupIDExisted = &ErrInfo{GroupIDNotFoundError, "GroupIDExisted", ""} // todo group id 已存在
ErrRecordNotFound = &ErrInfo{RecordNotFoundError, "RecordNotFoundError", ""}
ErrRelationshipAlready = &ErrInfo{RelationshipAlreadyError, "RelationshipAlreadyError", ""}
ErrNotRelationshipYet = &ErrInfo{NotRelationshipYetError, "NotRelationshipYetError", ""}
ErrCanNotAddYourself = &ErrInfo{CanNotAddYourselfError, "CanNotAddYourselfError", ""}
ErrOnlyOneOwner = &ErrInfo{OnlyOneOwnerError, "OnlyOneOwnerError", ""}
ErrInGroupAlready = &ErrInfo{InGroupAlreadyError, "InGroupAlreadyError", ""}
ErrNotInGroupYet = &ErrInfo{NotInGroupYetError, "NotInGroupYetError", ""}
ErrDismissedAlready = &ErrInfo{DismissedAlreadyError, "DismissedAlreadyError", ""}
ErrOwnerNotAllowedQuit = &ErrInfo{OwnerNotAllowedQuitError, "OwnerNotAllowedQuitError", ""}
ErrRegisteredAlready = &ErrInfo{RegisteredAlreadyError, "RegisteredAlreadyError", ""}
ErrGroupTypeNotSupport = &ErrInfo{GroupTypeNotSupport, "", ""}
ErrGroupNoOwner = &ErrInfo{GroupNoOwner, "ErrGroupNoOwner", ""}
ErrDefaultOther = &ErrInfo{DefaultOtherError, "DefaultOtherError", ""}
ErrData = &ErrInfo{DataError, "DataError", ""}
ErrTokenExpired = &ErrInfo{TokenExpiredError, "TokenExpiredError", ""}
ErrTokenInvalid = &ErrInfo{TokenInvalidError, "TokenInvalidError", ""} //
ErrTokenMalformed = &ErrInfo{TokenMalformedError, "TokenMalformedError", ""} //格式错误
ErrTokenNotValidYet = &ErrInfo{TokenNotValidYetError, "TokenNotValidYetError", ""} //还未生效
ErrTokenUnknown = &ErrInfo{TokenUnknownError, "TokenUnknownError", ""} //未知错误
ErrTokenKicked = &ErrInfo{TokenKickedError, "TokenKickedError", ""}
ErrTokenNotExist = &ErrInfo{TokenNotExistError, "TokenNotExistError", ""} //在redis中不存在
ErrTokenDifferentPlatformID = &ErrInfo{TokenDifferentPlatformIDError, "TokenDifferentPlatformIDError", ""}
ErrTokenDifferentUserID = &ErrInfo{TokenDifferentUserIDError, "TokenDifferentUserIDError", ""}
ErrMessageHasReadDisable = &ErrInfo{MessageHasReadDisable, "MessageHasReadDisable", ""}
ErrDB = ErrDatabase
ErrSendLimit = ErrInternalServer
ErrBlockedByPeer = &ErrInfo{BlockedByPeer, "BlockedByPeer", ""}
//不是对方的好友
ErrNotPeersFriend = &ErrInfo{NotPeersFriend, "NotPeersFriend", ""}
//
ErrMutedInGroup = &ErrInfo{MutedInGroup, "MutedInGroup", ""}
ErrMutedGroup = &ErrInfo{MutedGroup, "MutedGroup", ""}
ErrConnOverMaxNumLimit = &ErrInfo{ConnOverMaxNumLimit, "ConnOverMaxNumLimit", ""}
ErrConnArgsErr = &ErrInfo{ConnArgsErr, "args err, need token, sendID, platformID", ""}
ErrConnUpdateErr = &ErrInfo{ConnArgsErr, "upgrade http conn err", ""}
ErrConfig = &ErrInfo{ConfigError, "ConfigError", ""}
)
const (
FormattingError = 10001
HasRegistered = 10002
NotRegistered = 10003
PasswordErr = 10004
GetIMTokenErr = 10005
RepeatSendCode = 10006
MailSendCodeErr = 10007
SmsSendCodeErr = 10008
CodeInvalidOrExpired = 10009
RegisterFailed = 10010
ResetPasswordFailed = 10011
RegisterLimit = 10012
LoginLimit = 10013
InvitationError = 10014
)
// 通用错误码
const (
NoError = 0 //无错误
ArgsError = 90001 //输入参数错误
DatabaseError = 90002 //redis/mysql等db错误
ServerInternalError = 90003 //服务器内部错误
NetworkError = 90004 //网络错误
NoPermissionError = 90005 //权限不足
GRPCConnIsNil = 90006 //grpc连接空
DefaultOtherError = 90006 //其他错误
DataError = 90007 //数据错误
IdentityError = 90008 // 身份错误 非管理员token，且token中userID与请求userID不一致
ConfigError = 90009
CallbackError = 80000
)
// 账号错误码
const (
UserIDNotFoundError = 91001 //UserID不存在 或未注册
GroupIDNotFoundError = 91002 //GroupID不存在
RecordNotFoundError = 91002 //记录不存在
)
// 关系链错误码
const (
RelationshipAlreadyError = 92001 //已经是好友关系（或者黑名单）
NotRelationshipYetError = 92002 //不是好友关系（或者黑名单）
CanNotAddYourselfError = 92003 //不能添加自己为好友
BlockedByPeer = 92004 //被对方拉黑
NotPeersFriend = 92005 //不是对方的好友
)
// 群组错误码
const (
OnlyOneOwnerError = 93001 //只能有一个群主
InGroupAlreadyError = 93003 //已在群组中
NotInGroupYetError = 93004 //不在群组中
DismissedAlreadyError = 93004 //群组已经解散
OwnerNotAllowedQuitError = 93004 //群主不能退群
GroupTypeNotSupport = 93005
GroupNoOwner = 93006
MutedInGroup = 93007 //群成员被禁言
MutedGroup = 93008 //群被禁言
)
// 用户错误码
const (
RegisteredAlreadyError = 94001 //用户已经注册过了
)
// token错误码
const (
TokenExpiredError = 95001
TokenInvalidError = 95002
TokenMalformedError = 95003
TokenNotValidYetError = 95004
TokenUnknownError = 95005
TokenKickedError = 95006
TokenDifferentPlatformIDError = 95007
TokenDifferentUserIDError = 95008
TokenNotExistError = 95009
)
// 消息错误码
const (
MessageHasReadDisable = 96001
)
// 长连接网关错误码
const (
ConnOverMaxNumLimit = 970001
ConnArgsErr = 970002
ConnUpdateErr = 970003
)
// temp
var (
ErrServer = &ErrInfo{500, "server error", ""}
ErrTencentCredential = &ErrInfo{400, "ErrTencentCredential", ""}
)

62
pkg/common/middleware/gin.go → pkg/common/mw/gin.go
View File

@ -1,11 +1,11 @@
package middleware
package mw
import (
"OpenIM/pkg/common/constant"
"bytes"
"encoding/json"
"github.com/gin-gonic/gin"
"io/ioutil"
"io"
"net/http"
)
@ -26,36 +26,38 @@ func CorsHandler() gin.HandlerFunc {
}
}
func GinParseOperationID(c *gin.Context) {
if c.Request.Method == http.MethodPost {
operationID := c.Request.Header.Get(constant.OperationID)
if operationID == "" {
body, err := ioutil.ReadAll(c.Request.Body)
if err != nil {
c.String(400, "read request body error: "+err.Error())
c.Abort()
return
func GinParseOperationID() gin.HandlerFunc {
return func(c *gin.Context) {
if c.Request.Method == http.MethodPost {
operationID := c.Request.Header.Get(constant.OperationID)
if operationID == "" {
body, err := io.ReadAll(c.Request.Body)
if err != nil {
c.String(400, "read request body error: "+err.Error())
c.Abort()
return
}
req := struct {
OperationID string `json:"operationID"`
}{}
if err := json.Unmarshal(body, &req); err != nil {
c.String(400, "get operationID error: "+err.Error())
c.Abort()
return
}
if req.OperationID == "" {
c.String(400, "operationID empty")
c.Abort()
return
}
c.Request.Body = io.NopCloser(bytes.NewReader(body))
operationID = req.OperationID
c.Request.Header.Set(constant.OperationID, operationID)
}
req := struct {
OperationID string `json:"operationID"`
}{}
if err := json.Unmarshal(body, &req); err != nil {
c.String(400, "get operationID error: "+err.Error())
c.Abort()
return
}
if req.OperationID == "" {
c.String(400, "operationID empty")
c.Abort()
return
}
c.Request.Body = ioutil.NopCloser(bytes.NewReader(body))
operationID = req.OperationID
c.Request.Header.Set(constant.OperationID, operationID)
c.Set(constant.OperationID, operationID)
c.Next()
return
}
c.Set(constant.OperationID, operationID)
c.Next()
return
}
c.Next()
}

72
pkg/common/middleware/rpc.go → pkg/common/mw/rpc.go
View File

@ -1,11 +1,10 @@
package middleware
package mw
import (
"OpenIM/pkg/common/constant"
"OpenIM/pkg/common/log"
"OpenIM/pkg/common/tracelog"
"OpenIM/pkg/errs"
"OpenIM/pkg/utils"
"context"
"fmt"
"google.golang.org/grpc"
@ -16,14 +15,13 @@ import (
"runtime/debug"
)
func RpcServerInterceptor(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp interface{}, err error) {
func rpcServerInterceptor(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp interface{}, err error) {
var operationID string
defer func() {
if r := recover(); r != nil {
log.NewError(operationID, info.FullMethod, "type:", fmt.Sprintf("%T", r), "panic:", r, "stack:", string(debug.Stack()))
}
}()
//funcName := path.Base(info.FullMethod)
funcName := info.FullMethod
md, ok := metadata.FromIncomingContext(ctx)
if !ok {
@ -35,7 +33,7 @@ func RpcServerInterceptor(ctx context.Context, req interface{}, info *grpc.Unary
operationID = opts[0]
}
var opUserID string
if opts := md.Get("opUserID"); len(opts) == 1 {
if opts := md.Get(constant.OpUserID); len(opts) == 1 {
opUserID = opts[0]
}
ctx = tracelog.NewRpcCtx(ctx, funcName, operationID)
@ -44,43 +42,51 @@ func RpcServerInterceptor(ctx context.Context, req interface{}, info *grpc.Unary
resp, err = handler(ctx, req)
if err != nil {
tracelog.SetCtxInfo(ctx, funcName, err)
errInfo := constant.ToAPIErrWithErr(err)
var code codes.Code
if errInfo.ErrCode == 0 {
code = codes.Unknown
} else {
code = codes.Code(errInfo.ErrCode)
}
sta, err := status.New(code, errInfo.ErrMsg).WithDetails(wrapperspb.String(errInfo.DetailErrMsg))
if err != nil {
return nil, err
}
return nil, sta.Err()
return nil, rpcErrorToCode(err).Err()
}
tracelog.SetCtxInfo(ctx, funcName, nil, "rpcResp", rpcString(resp))
return
}
func rpcString(v interface{}) string {
if s, ok := v.(interface{ String() string }); ok {
return s.String()
func rpcClientInterceptor(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) (err error) {
if ctx == nil {
return errs.ErrInternalServer.Wrap("call rpc request context is nil")
}
return fmt.Sprintf("%+v", v)
}
func RpcClientInterceptor(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) (err error) {
//if cc == nil {
// return utils.Wrap(errs.ErrRpcConn, "")
//}
operationID, ok := ctx.Value(constant.OperationID).(string)
if !ok {
return utils.Wrap(errs.ErrArgs, "ctx missing operationID")
return errs.ErrArgs.Wrap("ctx missing operationID")
}
opUserID, ok := ctx.Value("opUserID").(string)
md := metadata.Pairs(constant.OperationID, operationID)
opUserID, ok := ctx.Value(constant.OpUserID).(string)
if ok {
md.Append(constant.OpUserID, opUserID)
}
err = invoker(metadata.NewOutgoingContext(ctx, md), method, req, reply, cc, opts...)
if err == nil {
return nil
}
rpcErr, ok := err.(interface{ GRPCStatus() *status.Status })
if !ok {
return utils.Wrap(errs.ErrArgs, "ctx missing opUserID")
return errs.NewCodeError(errs.DefaultOtherError, err.Error()).Wrap()
}
md := metadata.Pairs(constant.OperationID, operationID, "opUserID", opUserID)
return invoker(metadata.NewOutgoingContext(ctx, md), method, req, reply, cc, opts...)
sta := rpcErr.GRPCStatus()
if sta.Code() == 0 {
return errs.NewCodeError(errs.DefaultOtherError, err.Error()).Wrap()
}
details := sta.Details()
if len(details) == 0 {
return errs.NewCodeError(int(sta.Code()), sta.Message()).Wrap()
}
if v, ok := details[0].(*wrapperspb.StringValue); ok {
return errs.NewCodeError(int(sta.Code()), sta.Message()).Wrap(v.String())
}
return errs.NewCodeError(int(sta.Code()), sta.Message()).Wrap()
}
func GrpcServer() grpc.ServerOption {
return grpc.UnaryInterceptor(rpcServerInterceptor)
}
func GrpcClient() grpc.DialOption {
return grpc.WithUnaryInterceptor(rpcClientInterceptor)
}

46
pkg/common/mw/util.go Normal file
View File

@ -0,0 +1,46 @@
package mw
import (
"OpenIM/pkg/errs"
"fmt"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
"google.golang.org/protobuf/types/known/wrapperspb"
"math"
)
func rpcString(v interface{}) string {
if s, ok := v.(interface{ String() string }); ok {
return s.String()
}
return fmt.Sprintf("%+v", v)
}
func rpcErrorToCode(err error) *status.Status {
unwrap := errs.Unwrap(err)
var (
code codes.Code
msg string
)
if unwrap.(errs.CodeError) != nil {
c := unwrap.(errs.CodeError).Code()
if c <= 0 || c > math.MaxUint32 {
code = codes.OutOfRange // 错误码超出范围
} else {
code = codes.Code(c)
}
msg = unwrap.(errs.CodeError).Msg()
} else {
code = codes.Unknown
msg = unwrap.Error()
}
sta := status.New(code, msg)
if unwrap == err {
return sta
}
details, err := sta.WithDetails(wrapperspb.String(fmt.Sprintf("%+v", err)))
if err != nil {
return sta
}
return details
}

16
pkg/errs/coderr.go
View File

@ -6,39 +6,39 @@ import (
"strings"
)
type Coderr interface {
type CodeError interface {
Code() int
Msg() string
Wrap(msg ...string) error
error
}
func NewCodeError(code int, msg string) Coderr {
return &errInfo{
func NewCodeError(code int, msg string) CodeError {
return &codeError{
code: code,
msg: msg,
}
}
type errInfo struct {
type codeError struct {
code int
msg string
detail string
}
func (e *errInfo) Code() int {
func (e *codeError) Code() int {
return e.code
}
func (e *errInfo) Msg() string {
func (e *codeError) Msg() string {
return e.msg
}
func (e *errInfo) Wrap(w ...string) error {
func (e *codeError) Wrap(w ...string) error {
return errors.Wrap(e, strings.Join(w, ", "))
}
func (e *errInfo) Error() string {
func (e *codeError) Error() string {
return fmt.Sprintf("[%d]%s", e.code, e.msg)
}

1
pkg/errs/define.go
View File

@ -1 +0,0 @@
package errs

0
pkg/errs/errors.go → pkg/errs/predefine.go
View File