msb-public
/
Open-IM-Server
mirror of https://github.com/OpenIMSDK/Open-IM-Server
2
0
Code Issues Projects Releases Wiki Activity

Merge branch 'errcode' of github.com:OpenIMSDK/Open-IM-Server into errcode

Browse Source
test-errcode
wangchuxiao 3 years ago
parent 2c7f7f27d3 198cae5ade
commit 62ff4213e2
2 changed files with 156 additions and 7 deletions
Show Stats Download Patch File Download Diff File

151
internal/rpc/group/group.go
Unescape View File

@ -15,6 +15,11 @@ import (
cp "Open_IM/pkg/common/utils"
open_im_sdk "Open_IM/pkg/proto/sdk_ws"
"github.com/OpenIMSDK/getcdv3"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/metadata"
"google.golang.org/grpc/status"
"path"
"runtime/debug"
pbCache "Open_IM/pkg/proto/cache"
pbConversation "Open_IM/pkg/proto/conversation"
@ -53,6 +58,42 @@ func NewGroupServer(port int) *groupServer {
}
}
func UnaryServerInterceptor(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp interface{}, err error) {
defer func() {
if r := recover(); r != nil {
log.NewError("", info.FullMethod, "panic", r, "stack", string(debug.Stack()))
}
}()
funcName := path.Base(info.FullMethod)
md, ok := metadata.FromIncomingContext(ctx)
if !ok {
return nil, errors.New("not metadata")
}
operationID := md.Get("operationID")[0]
opUserID := md.Get("opUserID")[0]
ctx = trace_log.NewRpcCtx(ctx, funcName, operationID)
defer trace_log.ShowLog(ctx)
_ = opUserID
trace_log.SetContextInfo(ctx, funcName, err, "rpc req", req.(interface{ String() string }).String())
resp, err = handler(ctx, req)
if err != nil {
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()
}
trace_log.SetContextInfo(ctx, funcName, err, "rpc resp", resp.(interface{ String() string }).String())
return
}
func (s *groupServer) Run() {
log.NewInfo("", "group rpc start ")
listenIP := ""
@ -75,6 +116,7 @@ func (s *groupServer) Run() {
var grpcOpts = []grpc.ServerOption{
grpc.MaxRecvMsgSize(recvSize),
grpc.MaxSendMsgSize(sendSize),
grpc.UnaryInterceptor(UnaryServerInterceptor),
}
if config.Config.Prometheus.Enable {
promePkg.NewGrpcRequestCounter()
@ -114,7 +156,114 @@ func (s *groupServer) Run() {
log.NewInfo("", "group rpc success")
}
func (s *groupServer) CreateGroup(ctx context.Context, req *pbGroup.CreateGroupReq) (resp *pbGroup.CreateGroupResp, _ error) {
func (s *groupServer) CreateGroup(ctx context.Context, req *pbGroup.CreateGroupReq) (*pbGroup.CreateGroupResp, error) {
resp := &pbGroup.CreateGroupResp{GroupInfo: &open_im_sdk.GroupInfo{}}
if err := token_verify.CheckAccessV2(ctx, req.OpUserID, req.OwnerUserID); err != nil {
return nil, err
}
var groupOwnerNum int
var userIDs []string
for _, info := range req.InitMemberList {
if info.RoleLevel == constant.GroupOwner {
groupOwnerNum++
}
userIDs = append(userIDs, info.UserID)
}
if req.OwnerUserID != "" {
groupOwnerNum++
userIDs = append(userIDs, req.OwnerUserID)
}
if groupOwnerNum != 1 {
return nil, utils.Wrap(constant.ErrArgs, "")
}
if utils.IsRepeatStringSlice(userIDs) {
return nil, utils.Wrap(constant.ErrArgs, "")
}
users, err := rocksCache.GetUserInfoFromCacheBatch(ctx, userIDs)
if err != nil {
return nil, err
}
if len(users) != len(userIDs) {
return nil, utils.Wrap(constant.ErrArgs, "")
}
userMap := make(map[string]*imdb.User)
for i, user := range users {
userMap[user.UserID] = users[i]
}
if err := s.DelGroupAndUserCache(ctx, "", userIDs); err != nil {
return nil, err
}
if err := callbackBeforeCreateGroup(ctx, req); err != nil {
return nil, err
}
groupId := req.GroupInfo.GroupID
if groupId == "" {
groupId = utils.Md5(req.OperationID + strconv.FormatInt(time.Now().UnixNano(), 10))
bi := big.NewInt(0)
bi.SetString(groupId[0:8], 16)
groupId = bi.String()
}
groupInfo := imdb.Group{}
utils.CopyStructFields(&groupInfo, req.GroupInfo)
groupInfo.CreatorUserID = req.OpUserID
groupInfo.GroupID = groupId
groupInfo.CreateTime = time.Now()
if groupInfo.NotificationUpdateTime.Unix() < 0 {
groupInfo.NotificationUpdateTime = utils.UnixSecondToTime(0)
}
if req.GroupInfo.GroupType != constant.SuperGroup {
var groupMembers []*imdb.GroupMember
joinGroup := func(userID string, roleLevel int32) error {
groupMember := &imdb.GroupMember{GroupID: groupId, RoleLevel: roleLevel, OperatorUserID: req.OpUserID, JoinSource: constant.JoinByInvitation, InviterUserID: req.OpUserID}
user := userMap[userID]
utils.CopyStructFields(&groupMember, user)
if err := CallbackBeforeMemberJoinGroup(ctx, req.OperationID, groupMember, groupInfo.Ex); err != nil {
return err
}
groupMembers = append(groupMembers, groupMember)
return nil
}
if req.OwnerUserID == "" {
if err := joinGroup(req.OwnerUserID, constant.GroupOwner); err != nil {
return nil, err
}
}
for _, info := range req.InitMemberList {
if err := joinGroup(info.UserID, info.RoleLevel); err != nil {
return nil, err
}
}
if err := (*imdb.GroupMember)(nil).Create(ctx, groupMembers); err != nil {
return nil, err
}
} else {
if err := db.DB.CreateSuperGroup(groupId, userIDs, len(userIDs)); err != nil {
return nil, err
}
}
if err := (*imdb.Group)(nil).Create(ctx, []*imdb.Group{&groupInfo}); err != nil {
return nil, err
}
utils.CopyStructFields(resp.GroupInfo, groupInfo)
resp.GroupInfo.MemberCount = uint32(len(userIDs))
if req.GroupInfo.GroupType != constant.SuperGroup {
chat.GroupCreatedNotification(req.OperationID, req.OpUserID, groupId, userIDs)
} else {
for _, userID := range userIDs {
if err := rocksCache.DelJoinedSuperGroupIDListFromCache(ctx, userID); err != nil {
trace_log.SetContextInfo(ctx, "DelJoinedSuperGroupIDListFromCache", err, "userID", userID)
}
}
go func() {
for _, v := range userIDs {
chat.SuperGroupNotification(req.OperationID, v, v)
}
}()
}
return resp, nil
}
func (s *groupServer) CreateGroup1(ctx context.Context, req *pbGroup.CreateGroupReq) (resp *pbGroup.CreateGroupResp, _ error) {
resp = &pbGroup.CreateGroupResp{CommonResp: &open_im_sdk.CommonResp{}, GroupInfo: &open_im_sdk.GroupInfo{}}
ctx = trace_log.NewRpcCtx(ctx, utils.GetSelfFuncName(), req.OperationID)
defer func() {

12
pkg/common/db/mysql_model/im_mysql_model/group_member_model_k.go
Unescape View File

@ -38,33 +38,33 @@ type GroupMember struct {
func (*GroupMember) Create(ctx context.Context, groupMemberList []*GroupMember) (err error) {
defer func() {
trace_log.SetContextInfo(ctx, utils.GetSelfFuncName(), err, "groupMemberList", groupMemberList)
trace_log.SetContextInfo(ctx, utils.GetFuncName(1), err, "groupMemberList", groupMemberList)
}()
return utils.Wrap(db.DB.MysqlDB.DefaultGormDB().Create(&groupMemberList).Error, "")
}
func (*GroupMember) Delete(ctx context.Context, groupMembers []*GroupMember) (err error) {
defer func() {
trace_log.SetContextInfo(ctx, utils.GetSelfFuncName(), err, "groupMembers", groupMembers)
trace_log.SetContextInfo(ctx, utils.GetFuncName(1), err, "groupMembers", groupMembers)
}()
return utils.Wrap(db.DB.MysqlDB.DefaultGormDB().Delete(groupMembers).Error, "")
}
func (*GroupMember) UpdateByMap(ctx context.Context, groupID string, userID string, args map[string]interface{}) (err error) {
defer func() {
trace_log.SetContextInfo(ctx, utils.GetSelfFuncName(), err, "groupID", groupID, "userID", userID, "args", args)
trace_log.SetContextInfo(ctx, utils.GetFuncName(1), err, "groupID", groupID, "userID", userID, "args", args)
}()
return utils.Wrap(db.DB.MysqlDB.DefaultGormDB().Model(&GroupMember{}).Where("group_id = ? and user_id = ?", groupID, userID).Updates(args).Error, "")
}
func (*GroupMember) Update(ctx context.Context, groupMembers []*GroupMember) (err error) {
defer func() { trace_log.SetContextInfo(ctx, utils.GetSelfFuncName(), err, "groupMembers", groupMembers) }()
defer func() { trace_log.SetContextInfo(ctx, utils.GetFuncName(1), err, "groupMembers", groupMembers) }()
return utils.Wrap(db.DB.MysqlDB.DefaultGormDB().Updates(&groupMembers).Error, "")
}
func (*GroupMember) Find(ctx context.Context, groupMembers []*GroupMember) (groupList []*GroupMember, err error) {
defer func() {
trace_log.SetContextInfo(ctx, utils.GetSelfFuncName(), err, "groupMembers", groupMembers, "groupList", groupList)
trace_log.SetContextInfo(ctx, utils.GetFuncName(1), err, "groupMembers", groupMembers, "groupList", groupList)
}()
var where [][]interface{}
for _, groupMember := range groupMembers {
@ -76,7 +76,7 @@ func (*GroupMember) Find(ctx context.Context, groupMembers []*GroupMember) (grou
func (*GroupMember) Take(ctx context.Context, groupID string, userID string) (groupMember *GroupMember, err error) {
defer func() {
trace_log.SetContextInfo(ctx, utils.GetSelfFuncName(), err, "groupID", groupID, "userID", userID, "groupMember", *groupMember)
trace_log.SetContextInfo(ctx, utils.GetFuncName(1), err, "groupID", groupID, "userID", userID, "groupMember", *groupMember)
}()
groupMember = &GroupMember{}
return groupMember, utils.Wrap(db.DB.MysqlDB.DefaultGormDB().Where("group_id = ? and user_id = ?", groupID, userID).Take(groupMember).Error, "")

Write Preview
Loading…
Cancel
Save