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

Error code standardization

Browse Source
test-errcode
skiffer-git 3 years ago
parent 03a7048d31
commit 6b47c471f7
4 changed files with 252 additions and 308 deletions
Show Stats Download Patch File Download Diff File

10
internal/rpc/msg/callback.go
Unescape View File

@ -33,7 +33,7 @@ func copyCallbackCommonReqStruct(msg *pbChat.SendMsgReq) cbapi.CommonCallbackReq
return req return req
} }
func callbackBeforeSendSingleMsg(msg *pbChat.SendMsgReq) cbapi.CommonCallbackResp { func callbackBeforeSendSingleMsg(msg *pbChat.SendMsgReq) error {
callbackResp := cbapi.CommonCallbackResp{OperationID: msg.OperationID} callbackResp := cbapi.CommonCallbackResp{OperationID: msg.OperationID}
if !config.Config.Callback.CallbackBeforeSendSingleMsg.Enable { if !config.Config.Callback.CallbackBeforeSendSingleMsg.Enable {
return callbackResp return callbackResp
@ -65,7 +65,7 @@ func callbackBeforeSendSingleMsg(msg *pbChat.SendMsgReq) cbapi.CommonCallbackRes
return callbackResp return callbackResp
} }
func callbackAfterSendSingleMsg(msg *pbChat.SendMsgReq) cbapi.CommonCallbackResp { func callbackAfterSendSingleMsg(msg *pbChat.SendMsgReq) error {
callbackResp := cbapi.CommonCallbackResp{OperationID: msg.OperationID} callbackResp := cbapi.CommonCallbackResp{OperationID: msg.OperationID}
if !config.Config.Callback.CallbackAfterSendSingleMsg.Enable { if !config.Config.Callback.CallbackAfterSendSingleMsg.Enable {
return callbackResp return callbackResp
@ -88,7 +88,7 @@ func callbackAfterSendSingleMsg(msg *pbChat.SendMsgReq) cbapi.CommonCallbackResp
return callbackResp return callbackResp
} }
func callbackBeforeSendGroupMsg(msg *pbChat.SendMsgReq) cbapi.CommonCallbackResp { func callbackBeforeSendGroupMsg(msg *pbChat.SendMsgReq) error {
callbackResp := cbapi.CommonCallbackResp{OperationID: msg.OperationID} callbackResp := cbapi.CommonCallbackResp{OperationID: msg.OperationID}
if !config.Config.Callback.CallbackBeforeSendGroupMsg.Enable { if !config.Config.Callback.CallbackBeforeSendGroupMsg.Enable {
return callbackResp return callbackResp
@ -117,7 +117,7 @@ func callbackBeforeSendGroupMsg(msg *pbChat.SendMsgReq) cbapi.CommonCallbackResp
return callbackResp return callbackResp
} }
func callbackAfterSendGroupMsg(msg *pbChat.SendMsgReq) cbapi.CommonCallbackResp { func callbackAfterSendGroupMsg(msg *pbChat.SendMsgReq) error {
callbackResp := cbapi.CommonCallbackResp{OperationID: msg.OperationID} callbackResp := cbapi.CommonCallbackResp{OperationID: msg.OperationID}
if !config.Config.Callback.CallbackAfterSendGroupMsg.Enable { if !config.Config.Callback.CallbackAfterSendGroupMsg.Enable {
return callbackResp return callbackResp
@ -140,7 +140,7 @@ func callbackAfterSendGroupMsg(msg *pbChat.SendMsgReq) cbapi.CommonCallbackResp
return callbackResp return callbackResp
} }
func callbackMsgModify(msg *pbChat.SendMsgReq) cbapi.CommonCallbackResp { func callbackMsgModify(msg *pbChat.SendMsgReq) (err error) {
log.NewDebug(msg.OperationID, utils.GetSelfFuncName(), msg) log.NewDebug(msg.OperationID, utils.GetSelfFuncName(), msg)
callbackResp := cbapi.CommonCallbackResp{OperationID: msg.OperationID} callbackResp := cbapi.CommonCallbackResp{OperationID: msg.OperationID}
if !config.Config.Callback.CallbackMsgModify.Enable || msg.MsgData.ContentType != constant.Text { if !config.Config.Callback.CallbackMsgModify.Enable || msg.MsgData.ContentType != constant.Text {

27
internal/rpc/msg/send_msg.go
Unescape View File

@ -66,24 +66,6 @@ type MsgCallBackResp struct {
} }
} }
func isMessageHasReadEnabled(pb *msg.SendMsgReq) (bool, int32, string) {
switch pb.MsgData.ContentType {
case constant.HasReadReceipt:
if config.Config.SingleMessageHasReadReceiptEnable {
return true, 0, ""
} else {
return false, constant.ErrMessageHasReadDisable.ErrCode, constant.ErrMessageHasReadDisable.ErrMsg
}
case constant.GroupHasReadReceipt:
if config.Config.GroupMessageHasReadReceiptEnable {
return true, 0, ""
} else {
return false, constant.ErrMessageHasReadDisable.ErrCode, constant.ErrMessageHasReadDisable.ErrMsg
}
}
return true, 0, ""
}
func userIsMuteAndIsAdminInGroup(ctx context.Context, groupID, userID string) (isMute bool, isAdmin bool, err error) { func userIsMuteAndIsAdminInGroup(ctx context.Context, groupID, userID string) (isMute bool, isAdmin bool, err error) {
groupMemberInfo, err := rocksCache.GetGroupMemberInfoFromCache(ctx, groupID, userID) groupMemberInfo, err := rocksCache.GetGroupMemberInfoFromCache(ctx, groupID, userID)
if err != nil { if err != nil {
@ -107,7 +89,7 @@ func groupIsMuted(ctx context.Context, groupID string) (bool, error) {
return false, nil return false, nil
} }
func (rpc *msgServer) messageVerification(ctx context.Context, data *pbChat.SendMsgReq) (bool, int32, string, []string) { func (rpc *msgServer) messageVerification(ctx context.Context, data *msg.SendMsgReq) ([]string, error) {
switch data.MsgData.SessionType { switch data.MsgData.SessionType {
case constant.SingleChatType: case constant.SingleChatType:
if utils.IsContain(data.MsgData.SendID, config.Config.Manager.AppManagerUid) { if utils.IsContain(data.MsgData.SendID, config.Config.Manager.AppManagerUid) {
@ -314,12 +296,10 @@ func (rpc *msgServer) encapsulateMsgData(msg *sdkws.MsgData) {
utils.SetSwitchFromOptions(msg.Options, constant.IsUnreadCount, false) utils.SetSwitchFromOptions(msg.Options, constant.IsUnreadCount, false)
utils.SetSwitchFromOptions(msg.Options, constant.IsOfflinePush, false) utils.SetSwitchFromOptions(msg.Options, constant.IsOfflinePush, false)
case constant.HasReadReceipt: case constant.HasReadReceipt:
log.Info("", "this is a test start", msg, msg.Options)
utils.SetSwitchFromOptions(msg.Options, constant.IsConversationUpdate, false) utils.SetSwitchFromOptions(msg.Options, constant.IsConversationUpdate, false)
utils.SetSwitchFromOptions(msg.Options, constant.IsSenderConversationUpdate, false) utils.SetSwitchFromOptions(msg.Options, constant.IsSenderConversationUpdate, false)
utils.SetSwitchFromOptions(msg.Options, constant.IsUnreadCount, false) utils.SetSwitchFromOptions(msg.Options, constant.IsUnreadCount, false)
utils.SetSwitchFromOptions(msg.Options, constant.IsOfflinePush, false) utils.SetSwitchFromOptions(msg.Options, constant.IsOfflinePush, false)
log.Info("", "this is a test end", msg, msg.Options)
case constant.Typing: case constant.Typing:
utils.SetSwitchFromOptions(msg.Options, constant.IsHistory, false) utils.SetSwitchFromOptions(msg.Options, constant.IsHistory, false)
utils.SetSwitchFromOptions(msg.Options, constant.IsPersistent, false) utils.SetSwitchFromOptions(msg.Options, constant.IsPersistent, false)
@ -379,11 +359,10 @@ func returnMsg(replay *pbChat.SendMsgResp, pb *pbChat.SendMsgReq, errCode int32,
return replay, nil return replay, nil
} }
func modifyMessageByUserMessageReceiveOpt(userID, sourceID string, sessionType int, pb *pbChat.SendMsgReq) bool { func modifyMessageByUserMessageReceiveOpt(userID, sourceID string, sessionType int, pb *pbChat.SendMsgReq) (bool, error) {
opt, err := db.DB.GetUserGlobalMsgRecvOpt(userID) opt, err := db.DB.GetUserGlobalMsgRecvOpt(userID)
if err != nil { if err != nil {
log.NewError(pb.OperationID, "GetUserGlobalMsgRecvOpt from redis err", userID, pb.String(), err.Error()) log.NewError(pb.OperationID, "GetUserGlobalMsgRecvOpt from redis err", userID, pb.String(), err.Error())
} }
switch opt { switch opt {
case constant.ReceiveMessage: case constant.ReceiveMessage:
@ -538,7 +517,7 @@ func (rpc *msgServer) sendMsgToGroup(ctx context.Context, list []string, pb pbCh
wg.Done() wg.Done()
} }
func (rpc *msgServer) sendMsgToGroupOptimization(ctx context.Context, list []string, groupPB *pbChat.SendMsgReq, status string, sendTag *bool, wg *sync.WaitGroup) { func (rpc *msgServer) sendMsgToGroupOptimization(ctx context.Context, list []string, groupPB *msg.SendMsgReq, sendTag *bool, wg *sync.WaitGroup) {
msgToMQGroup := pbChat.MsgDataToMQ{Token: groupPB.Token, OperationID: groupPB.OperationID, MsgData: groupPB.MsgData} msgToMQGroup := pbChat.MsgDataToMQ{Token: groupPB.Token, OperationID: groupPB.OperationID, MsgData: groupPB.MsgData}
tempOptions := make(map[string]bool, 1) tempOptions := make(map[string]bool, 1)
for k, v := range groupPB.MsgData.Options { for k, v := range groupPB.MsgData.Options {

496
internal/rpc/msg/send_pull.go
Unescape View File

@ -4,309 +4,265 @@ import (
"Open_IM/pkg/common/constant" "Open_IM/pkg/common/constant"
promePkg "Open_IM/pkg/common/prometheus" promePkg "Open_IM/pkg/common/prometheus"
pbConversation "Open_IM/pkg/proto/conversation" pbConversation "Open_IM/pkg/proto/conversation"
pbChat "Open_IM/pkg/proto/msg" "Open_IM/pkg/proto/msg"
"Open_IM/pkg/proto/sdkws" "Open_IM/pkg/proto/sdkws"
"Open_IM/pkg/utils"
"context" "context"
go_redis "github.com/go-redis/redis/v8" go_redis "github.com/go-redis/redis/v8"
"github.com/golang/protobuf/proto"
"strings" "strings"
"sync" "sync"
"time"
) )
func (m *msgServer) SendMsg(ctx context.Context, pb *pbChat.SendMsgReq) (*pbChat.SendMsgResp, error) { func (m *msgServer) sendMsgSuperGroupChat(ctx context.Context, req *msg.SendMsgReq) (resp *msg.SendMsgResp, err error) {
replay := pbChat.SendMsgResp{} promePkg.PromeInc(promePkg.WorkSuperGroupChatMsgRecvSuccessCounter)
// callback
if err = callbackBeforeSendGroupMsg(req); err != nil {
return nil, err
}
flag, errCode, errMsg := isMessageHasReadEnabled(pb) if _, err = m.messageVerification(ctx, req); err != nil {
if !flag { promePkg.PromeInc(promePkg.WorkSuperGroupChatMsgProcessFailedCounter)
return returnMsg(&replay, pb, errCode, errMsg, "", 0) return nil, err
}
msgToMQSingle := msg.MsgDataToMQ{MsgData: req.MsgData}
err = m.MsgInterface.MsgToMQ(ctx, msgToMQSingle.MsgData.GroupID, &msgToMQSingle)
if err != nil {
return nil, err
} }
t1 := time.Now()
m.encapsulateMsgData(pb.MsgData)
log.Debug(pb.OperationID, "encapsulateMsgData ", " cost time: ", time.Since(t1))
msgToMQSingle := pbChat.MsgDataToMQ{Token: pb.Token, OperationID: pb.OperationID, MsgData: pb.MsgData}
// callback // callback
t1 = time.Now() if err = callbackAfterSendGroupMsg(req); err != nil {
callbackResp := callbackMsgModify(pb) return nil, err
log.Debug(pb.OperationID, "callbackMsgModify ", callbackResp, "cost time: ", time.Since(t1))
if callbackResp.ErrCode != 0 {
log.Error(pb.OperationID, utils.GetSelfFuncName(), "callbackMsgModify resp: ", callbackResp)
} }
log.NewDebug(pb.OperationID, utils.GetSelfFuncName(), "callbackResp: ", callbackResp)
if callbackResp.ActionCode != constant.ActionAllow { promePkg.PromeInc(promePkg.WorkSuperGroupChatMsgProcessSuccessCounter)
if callbackResp.ErrCode == 0 { resp.SendTime = msgToMQSingle.MsgData.SendTime
callbackResp.ErrCode = 201 resp.ServerMsgID = msgToMQSingle.MsgData.ServerMsgID
} resp.ClientMsgID = msgToMQSingle.MsgData.ClientMsgID
log.NewDebug(pb.OperationID, utils.GetSelfFuncName(), "callbackMsgModify result", "end rpc and return", pb.MsgData) return resp, nil
return returnMsg(&replay, pb, int32(callbackResp.ErrCode), callbackResp.ErrMsg, "", 0) }
func (m *msgServer) sendMsgNotification(ctx context.Context, req *msg.SendMsgReq) (resp *msg.SendMsgResp, err error) {
msgToMQSingle := msg.MsgDataToMQ{MsgData: req.MsgData}
err = m.MsgInterface.MsgToMQ(ctx, msgToMQSingle.MsgData.RecvID, &msgToMQSingle)
if err != nil {
return nil, err
} }
switch pb.MsgData.SessionType { if msgToMQSingle.MsgData.SendID != msgToMQSingle.MsgData.RecvID { //Filter messages sent to yourself
case constant.SingleChatType: err = m.MsgInterface.MsgToMQ(ctx, msgToMQSingle.MsgData.SendID, &msgToMQSingle)
promePkg.PromeInc(promePkg.SingleChatMsgRecvSuccessCounter) if err != nil {
// callback return nil, err
t1 = time.Now()
callbackResp := callbackBeforeSendSingleMsg(pb)
log.Debug(pb.OperationID, "callbackBeforeSendSingleMsg ", " cost time: ", time.Since(t1))
if callbackResp.ErrCode != 0 {
log.NewError(pb.OperationID, utils.GetSelfFuncName(), "callbackBeforeSendSingleMsg resp: ", callbackResp)
} }
if callbackResp.ActionCode != constant.ActionAllow { }
if callbackResp.ErrCode == 0 {
callbackResp.ErrCode = 201 resp.SendTime = msgToMQSingle.MsgData.SendTime
} resp.ServerMsgID = msgToMQSingle.MsgData.ServerMsgID
log.NewDebug(pb.OperationID, utils.GetSelfFuncName(), "callbackBeforeSendSingleMsg result", "end rpc and return", callbackResp) resp.ClientMsgID = msgToMQSingle.MsgData.ClientMsgID
promePkg.PromeInc(promePkg.SingleChatMsgProcessFailedCounter) return resp, nil
return returnMsg(&replay, pb, int32(callbackResp.ErrCode), callbackResp.ErrMsg, "", 0) }
func (m *msgServer) sendMsgSingleChat(ctx context.Context, req *msg.SendMsgReq) (resp *msg.SendMsgResp, err error) {
promePkg.PromeInc(promePkg.SingleChatMsgRecvSuccessCounter)
if err = callbackBeforeSendSingleMsg(req); err != nil {
return nil, err
}
_, err = m.messageVerification(ctx, req)
if err != nil {
return nil, err
}
isSend, err := modifyMessageByUserMessageReceiveOpt(req.MsgData.RecvID, req.MsgData.SendID, constant.SingleChatType, req)
if err != nil {
return nil, err
}
msgToMQSingle := msg.MsgDataToMQ{MsgData: req.MsgData}
if isSend {
err = m.MsgInterface.MsgToMQ(ctx, req.MsgData.RecvID, &msgToMQSingle)
if err != nil {
return nil, constant.ErrInternalServer.Wrap("insert to mq")
} }
t1 = time.Now() }
flag, errCode, errMsg, _ = rpc.messageVerification(ctx, pb) if msgToMQSingle.MsgData.SendID != msgToMQSingle.MsgData.RecvID { //Filter messages sent to yourself
log.Debug(pb.OperationID, "messageVerification ", flag, " cost time: ", time.Since(t1)) err = m.MsgInterface.MsgToMQ(ctx, req.MsgData.SendID, &msgToMQSingle)
if !flag { if err != nil {
return returnMsg(&replay, pb, errCode, errMsg, "", 0) return nil, constant.ErrInternalServer.Wrap("insert to mq")
} }
t1 = time.Now() }
isSend := modifyMessageByUserMessageReceiveOpt(pb.MsgData.RecvID, pb.MsgData.SendID, constant.SingleChatType, pb) err = callbackAfterSendSingleMsg(req)
log.Info(pb.OperationID, "modifyMessageByUserMessageReceiveOpt ", " cost time: ", time.Since(t1)) if err != nil {
if isSend { return nil, err
msgToMQSingle.MsgData = pb.MsgData }
log.NewInfo(msgToMQSingle.OperationID, msgToMQSingle) promePkg.PromeInc(promePkg.SingleChatMsgProcessSuccessCounter)
t1 = time.Now() resp.SendTime = msgToMQSingle.MsgData.SendTime
err1 := rpc.sendMsgToWriter(ctx, &msgToMQSingle, msgToMQSingle.MsgData.RecvID, constant.OnlineStatus) resp.ServerMsgID = msgToMQSingle.MsgData.ServerMsgID
log.Info(pb.OperationID, "sendMsgToWriter ", " cost time: ", time.Since(t1)) resp.ClientMsgID = msgToMQSingle.MsgData.ClientMsgID
if err1 != nil { return resp, nil
log.NewError(msgToMQSingle.OperationID, "kafka send msg err :RecvID", msgToMQSingle.MsgData.RecvID, msgToMQSingle.String(), err1.Error()) }
promePkg.PromeInc(promePkg.SingleChatMsgProcessFailedCounter)
return returnMsg(&replay, pb, 201, "kafka send msg err", "", 0) func (m *msgServer) sendMsgGroupChat(ctx context.Context, req *msg.SendMsgReq) (resp *msg.SendMsgResp, err error) {
} // callback
promePkg.PromeInc(promePkg.GroupChatMsgRecvSuccessCounter)
err = callbackBeforeSendGroupMsg(req)
if err != nil {
return nil, err
}
var memberUserIDList []string
if memberUserIDList, err = m.messageVerification(ctx, req); err != nil {
promePkg.PromeInc(promePkg.GroupChatMsgProcessFailedCounter)
return nil, err
}
var addUidList []string
switch req.MsgData.ContentType {
case constant.MemberKickedNotification:
var tips sdkws.TipsComm
var memberKickedTips sdkws.MemberKickedTips
err := proto.Unmarshal(req.MsgData.Content, &tips)
if err != nil {
} }
if msgToMQSingle.MsgData.SendID != msgToMQSingle.MsgData.RecvID { //Filter messages sent to yourself err = proto.Unmarshal(tips.Detail, &memberKickedTips)
t1 = time.Now() if err != nil {
err2 := rpc.sendMsgToWriter(ctx, &msgToMQSingle, msgToMQSingle.MsgData.SendID, constant.OnlineStatus)
log.Info(pb.OperationID, "sendMsgToWriter ", " cost time: ", time.Since(t1))
if err2 != nil {
log.NewError(msgToMQSingle.OperationID, "kafka send msg err:SendID", msgToMQSingle.MsgData.SendID, msgToMQSingle.String())
promePkg.PromeInc(promePkg.SingleChatMsgProcessFailedCounter)
return returnMsg(&replay, pb, 201, "kafka send msg err", "", 0)
}
} }
// callback
t1 = time.Now() for _, v := range memberKickedTips.KickedUserList {
callbackResp = callbackAfterSendSingleMsg(pb) addUidList = append(addUidList, v.UserID)
log.Info(pb.OperationID, "callbackAfterSendSingleMsg ", " cost time: ", time.Since(t1))
if callbackResp.ErrCode != 0 {
log.NewError(pb.OperationID, utils.GetSelfFuncName(), "callbackAfterSendSingleMsg resp: ", callbackResp)
} }
promePkg.PromeInc(promePkg.SingleChatMsgProcessSuccessCounter) case constant.MemberQuitNotification:
return returnMsg(&replay, pb, 0, "", msgToMQSingle.MsgData.ServerMsgID, msgToMQSingle.MsgData.SendTime) addUidList = append(addUidList, req.MsgData.SendID)
case constant.GroupChatType:
// callback default:
promePkg.PromeInc(promePkg.GroupChatMsgRecvSuccessCounter) }
callbackResp := callbackBeforeSendGroupMsg(pb) if len(addUidList) > 0 {
if callbackResp.ErrCode != 0 { memberUserIDList = append(memberUserIDList, addUidList...)
log.NewError(pb.OperationID, utils.GetSelfFuncName(), "callbackBeforeSendGroupMsg resp:", callbackResp) }
//split parallel send
var wg sync.WaitGroup
var sendTag bool
var split = 20
msgToMQSingle := msg.MsgDataToMQ{MsgData: req.MsgData}
mErr := make(map[string]error, 0)
remain := len(memberUserIDList) % split
for i := 0; i < len(memberUserIDList)/split; i++ {
wg.Add(1)
tmp := valueCopy(req)
go m.sendMsgToGroupOptimization(ctx, memberUserIDList[i*split:(i+1)*split], tmp, &sendTag, &wg)
}
if remain > 0 {
wg.Add(1)
tmp := valueCopy(req)
go m.sendMsgToGroupOptimization(ctx, memberUserIDList[split*(len(memberUserIDList)/split):], tmp, &sendTag, &wg)
}
wg.Wait()
// callback
err = callbackAfterSendGroupMsg(req)
if err != nil {
return nil, err
}
for _, v := range mErr {
if v != nil {
return nil, v
} }
if callbackResp.ActionCode != constant.ActionAllow { }
if callbackResp.ErrCode == 0 {
callbackResp.ErrCode = 201 if req.MsgData.ContentType == constant.AtText {
go func() {
var conversationReq pbConversation.ModifyConversationFieldReq
var tag bool
var atUserID []string
conversation := pbConversation.Conversation{
OwnerUserID: req.MsgData.SendID,
ConversationID: utils.GetConversationIDBySessionType(pb.MsgData.GroupID, constant.GroupChatType),
ConversationType: constant.GroupChatType,
GroupID: req.MsgData.GroupID,
} }
log.NewDebug(pb.OperationID, utils.GetSelfFuncName(), "callbackBeforeSendSingleMsg result", "end rpc and return", callbackResp) conversationReq.Conversation = &conversation
promePkg.PromeInc(promePkg.GroupChatMsgProcessFailedCounter) conversationReq.OperationID = pb.OperationID
return returnMsg(&replay, pb, int32(callbackResp.ErrCode), callbackResp.ErrMsg, "", 0) conversationReq.FieldType = constant.FieldGroupAtType
} tagAll := utils.IsContain(constant.AtAllString, req.MsgData.AtUserIDList)
var memberUserIDList []string if tagAll {
if flag, errCode, errMsg, memberUserIDList = rpc.messageVerification(ctx, pb); !flag { atUserID = utils.DifferenceString([]string{constant.AtAllString}, pb.MsgData.AtUserIDList)
promePkg.PromeInc(promePkg.GroupChatMsgProcessFailedCounter) if len(atUserID) == 0 { //just @everyone
return returnMsg(&replay, pb, errCode, errMsg, "", 0) conversationReq.UserIDList = memberUserIDList
} conversation.GroupAtType = constant.AtAll
log.Debug(pb.OperationID, "GetGroupAllMember userID list", memberUserIDList, "len: ", len(memberUserIDList)) } else { //@Everyone and @other people
var addUidList []string conversationReq.UserIDList = atUserID
switch pb.MsgData.ContentType { conversation.GroupAtType = constant.AtAllAtMe
case constant.MemberKickedNotification: tag = true
var tips sdkws.TipsComm }
var memberKickedTips sdkws.MemberKickedTips } else {
err := proto.Unmarshal(pb.MsgData.Content, &tips) conversationReq.UserIDList = req.MsgData.AtUserIDList
conversation.GroupAtType = constant.AtMe
}
etcdConn, err := rpc.GetConn(ctx, config.Config.RpcRegisterName.OpenImConversationName)
if err != nil { if err != nil {
log.Error(pb.OperationID, "Unmarshal err", err.Error()) errMsg := pb.OperationID + "getcdv3.GetDefaultConn == nil"
log.NewError(pb.OperationID, errMsg)
return
} }
err = proto.Unmarshal(tips.Detail, &memberKickedTips) client := pbConversation.NewConversationClient(etcdConn)
conversationReply, err := client.ModifyConversationField(context.Background(), &conversationReq)
if err != nil { if err != nil {
log.Error(pb.OperationID, "Unmarshal err", err.Error()) log.NewError(conversationReq.OperationID, "ModifyConversationField rpc failed, ", conversationReq.String(), err.Error())
} else if conversationReply.CommonResp.ErrCode != 0 {
log.NewError(conversationReq.OperationID, "ModifyConversationField rpc failed, ", conversationReq.String(), conversationReply.String())
} }
log.Info(pb.OperationID, "data is ", memberKickedTips) if tag {
for _, v := range memberKickedTips.KickedUserList { conversationReq.UserIDList = utils.DifferenceString(atUserID, memberUserIDList)
addUidList = append(addUidList, v.UserID) conversation.GroupAtType = constant.AtAll
etcdConn := rpc.GetDefaultConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImConversationName, pb.OperationID)
if etcdConn == nil {
errMsg := pb.OperationID + "getcdv3.GetDefaultConn == nil"
log.NewError(pb.OperationID, errMsg)
return
}
client := pbConversation.NewConversationClient(etcdConn)
conversationReply, err := client.ModifyConversationField(context.Background(), &conversationReq)
if err != nil {
log.NewError(conversationReq.OperationID, "ModifyConversationField rpc failed, ", conversationReq.String(), err.Error())
} else if conversationReply.CommonResp.ErrCode != 0 {
log.NewError(conversationReq.OperationID, "ModifyConversationField rpc failed, ", conversationReq.String(), conversationReply.String())
}
} }
case constant.MemberQuitNotification: }()
addUidList = append(addUidList, pb.MsgData.SendID) }
default: promePkg.PromeInc(promePkg.GroupChatMsgProcessSuccessCounter)
} resp.SendTime = msgToMQSingle.MsgData.SendTime
if len(addUidList) > 0 { resp.ServerMsgID = msgToMQSingle.MsgData.ServerMsgID
memberUserIDList = append(memberUserIDList, addUidList...) resp.ClientMsgID = msgToMQSingle.MsgData.ClientMsgID
} return resp, nil
m := make(map[string][]string, 2)
m[constant.OnlineStatus] = memberUserIDList
t1 = time.Now()
//split parallel send }
var wg sync.WaitGroup func (m *msgServer) SendMsg(ctx context.Context, req *msg.SendMsgReq) (resp *msg.SendMsgResp, error error) {
var sendTag bool resp = &msg.SendMsgResp{}
var split = 20 flag := isMessageHasReadEnabled(req.MsgData)
for k, v := range m { if !flag {
remain := len(v) % split return nil, constant.ErrMessageHasReadDisable.Wrap()
for i := 0; i < len(v)/split; i++ { }
wg.Add(1) m.encapsulateMsgData(req.MsgData)
tmp := valueCopy(pb) if err := callbackMsgModify(req); err != nil {
// go rpc.sendMsgToGroup(v[i*split:(i+1)*split], *pb, k, &sendTag, &wg) return nil, err
go rpc.sendMsgToGroupOptimization(ctx, v[i*split:(i+1)*split], tmp, k, &sendTag, &wg) }
} switch req.MsgData.SessionType {
if remain > 0 { case constant.SingleChatType:
wg.Add(1) return m.sendMsgSingleChat(ctx, req)
tmp := valueCopy(pb) case constant.GroupChatType:
// go rpc.sendMsgToGroup(v[split*(len(v)/split):], *pb, k, &sendTag, &wg) return m.sendMsgGroupChat(ctx, req)
go rpc.sendMsgToGroupOptimization(ctx, v[split*(len(v)/split):], tmp, k, &sendTag, &wg)
}
}
log.Debug(pb.OperationID, "send msg cost time22 ", time.Since(t1), pb.MsgData.ClientMsgID, "uidList : ", len(addUidList))
//wg.Add(1)
//go rpc.sendMsgToGroup(addUidList, *pb, constant.OnlineStatus, &sendTag, &wg)
wg.Wait()
t1 = time.Now()
// callback
callbackResp = callbackAfterSendGroupMsg(pb)
if callbackResp.ErrCode != 0 {
log.NewError(pb.OperationID, utils.GetSelfFuncName(), "callbackAfterSendGroupMsg resp: ", callbackResp)
}
if !sendTag {
log.NewWarn(pb.OperationID, "send tag is ", sendTag)
promePkg.PromeInc(promePkg.GroupChatMsgProcessFailedCounter)
return returnMsg(&replay, pb, 201, "kafka send msg err", "", 0)
} else {
if pb.MsgData.ContentType == constant.AtText {
go func() {
var conversationReq pbConversation.ModifyConversationFieldReq
var tag bool
var atUserID []string
conversation := pbConversation.Conversation{
OwnerUserID: pb.MsgData.SendID,
ConversationID: utils.GetConversationIDBySessionType(pb.MsgData.GroupID, constant.GroupChatType),
ConversationType: constant.GroupChatType,
GroupID: pb.MsgData.GroupID,
}
conversationReq.Conversation = &conversation
conversationReq.OperationID = pb.OperationID
conversationReq.FieldType = constant.FieldGroupAtType
tagAll := utils.IsContain(constant.AtAllString, pb.MsgData.AtUserIDList)
if tagAll {
atUserID = utils.DifferenceString([]string{constant.AtAllString}, pb.MsgData.AtUserIDList)
if len(atUserID) == 0 { //just @everyone
conversationReq.UserIDList = memberUserIDList
conversation.GroupAtType = constant.AtAll
} else { //@Everyone and @other people
conversationReq.UserIDList = atUserID
conversation.GroupAtType = constant.AtAllAtMe
tag = true
}
} else {
conversationReq.UserIDList = pb.MsgData.AtUserIDList
conversation.GroupAtType = constant.AtMe
}
etcdConn, err := rpc.GetConn(ctx, config.Config.RpcRegisterName.OpenImConversationName)
if err != nil {
errMsg := pb.OperationID + "getcdv3.GetDefaultConn == nil"
log.NewError(pb.OperationID, errMsg)
return
}
client := pbConversation.NewConversationClient(etcdConn)
conversationReply, err := client.ModifyConversationField(context.Background(), &conversationReq)
if err != nil {
log.NewError(conversationReq.OperationID, "ModifyConversationField rpc failed, ", conversationReq.String(), err.Error())
} else if conversationReply.CommonResp.ErrCode != 0 {
log.NewError(conversationReq.OperationID, "ModifyConversationField rpc failed, ", conversationReq.String(), conversationReply.String())
}
if tag {
conversationReq.UserIDList = utils.DifferenceString(atUserID, memberUserIDList)
conversation.GroupAtType = constant.AtAll
etcdConn := rpc.GetDefaultConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImConversationName, pb.OperationID)
if etcdConn == nil {
errMsg := pb.OperationID + "getcdv3.GetDefaultConn == nil"
log.NewError(pb.OperationID, errMsg)
return
}
client := pbConversation.NewConversationClient(etcdConn)
conversationReply, err := client.ModifyConversationField(context.Background(), &conversationReq)
if err != nil {
log.NewError(conversationReq.OperationID, "ModifyConversationField rpc failed, ", conversationReq.String(), err.Error())
} else if conversationReply.CommonResp.ErrCode != 0 {
log.NewError(conversationReq.OperationID, "ModifyConversationField rpc failed, ", conversationReq.String(), conversationReply.String())
}
}
}()
}
log.Debug(pb.OperationID, "send msg cost time3 ", time.Since(t1), pb.MsgData.ClientMsgID)
promePkg.PromeInc(promePkg.GroupChatMsgProcessSuccessCounter)
return returnMsg(&replay, pb, 0, "", msgToMQSingle.MsgData.ServerMsgID, msgToMQSingle.MsgData.SendTime)
}
case constant.NotificationChatType: case constant.NotificationChatType:
t1 = time.Now() return m.sendMsgNotification(ctx, req)
msgToMQSingle.MsgData = pb.MsgData
log.NewInfo(msgToMQSingle.OperationID, msgToMQSingle)
err1 := rpc.sendMsgToWriter(ctx, &msgToMQSingle, msgToMQSingle.MsgData.RecvID, constant.OnlineStatus)
if err1 != nil {
log.NewError(msgToMQSingle.OperationID, "kafka send msg err:RecvID", msgToMQSingle.MsgData.RecvID, msgToMQSingle.String())
return returnMsg(&replay, pb, 201, "kafka send msg err", "", 0)
}
if msgToMQSingle.MsgData.SendID != msgToMQSingle.MsgData.RecvID { //Filter messages sent to yourself
err2 := rpc.sendMsgToWriter(ctx, &msgToMQSingle, msgToMQSingle.MsgData.SendID, constant.OnlineStatus)
if err2 != nil {
log.NewError(msgToMQSingle.OperationID, "kafka send msg err:SendID", msgToMQSingle.MsgData.SendID, msgToMQSingle.String())
return returnMsg(&replay, pb, 201, "kafka send msg err", "", 0)
}
}
log.Debug(pb.OperationID, "send msg cost time ", time.Since(t1), pb.MsgData.ClientMsgID)
return returnMsg(&replay, pb, 0, "", msgToMQSingle.MsgData.ServerMsgID, msgToMQSingle.MsgData.SendTime)
case constant.SuperGroupChatType: case constant.SuperGroupChatType:
promePkg.PromeInc(promePkg.WorkSuperGroupChatMsgRecvSuccessCounter) return m.sendMsgSuperGroupChat(ctx, req)
// callback
callbackResp := callbackBeforeSendGroupMsg(pb)
if callbackResp.ErrCode != 0 {
log.NewError(pb.OperationID, utils.GetSelfFuncName(), "callbackBeforeSendSuperGroupMsg resp: ", callbackResp)
}
if callbackResp.ActionCode != constant.ActionAllow {
if callbackResp.ErrCode == 0 {
callbackResp.ErrCode = 201
}
promePkg.PromeInc(promePkg.WorkSuperGroupChatMsgProcessFailedCounter)
log.NewDebug(pb.OperationID, utils.GetSelfFuncName(), "callbackBeforeSendSuperGroupMsg result", "end rpc and return", callbackResp)
return returnMsg(&replay, pb, int32(callbackResp.ErrCode), callbackResp.ErrMsg, "", 0)
}
if flag, errCode, errMsg, _ = rpc.messageVerification(ctx, pb); !flag {
promePkg.PromeInc(promePkg.WorkSuperGroupChatMsgProcessFailedCounter)
return returnMsg(&replay, pb, errCode, errMsg, "", 0)
}
msgToMQSingle.MsgData = pb.MsgData
log.NewInfo(msgToMQSingle.OperationID, msgToMQSingle)
err1 := rpc.sendMsgToWriter(ctx, &msgToMQSingle, msgToMQSingle.MsgData.GroupID, constant.OnlineStatus)
if err1 != nil {
log.NewError(msgToMQSingle.OperationID, "kafka send msg err:RecvID", msgToMQSingle.MsgData.RecvID, msgToMQSingle.String())
promePkg.PromeInc(promePkg.WorkSuperGroupChatMsgProcessFailedCounter)
return returnMsg(&replay, pb, 201, "kafka send msg err", "", 0)
}
// callback
callbackResp = callbackAfterSendGroupMsg(pb)
if callbackResp.ErrCode != 0 {
log.NewError(pb.OperationID, utils.GetSelfFuncName(), "callbackAfterSendSuperGroupMsg resp: ", callbackResp)
}
promePkg.PromeInc(promePkg.WorkSuperGroupChatMsgProcessSuccessCounter)
return returnMsg(&replay, pb, 0, "", msgToMQSingle.MsgData.ServerMsgID, msgToMQSingle.MsgData.SendTime)
default: default:
return returnMsg(&replay, pb, 203, "unknown sessionType", "", 0) return nil, constant.ErrArgs.Wrap("unknown sessionType")
} }
} }
func (rpc *rpcChat) GetMaxAndMinSeq(_ context.Context, in *sdkws.GetMaxAndMinSeqReq) (*sdkws.GetMaxAndMinSeqResp, error) { func (m *msgServer) GetMaxAndMinSeq(_ context.Context, in *sdkws.GetMaxAndMinSeqReq) (*sdkws.GetMaxAndMinSeqResp, error) {
log.NewInfo(in.OperationID, "rpc getMaxAndMinSeq is arriving", in.String()) log.NewInfo(in.OperationID, "rpc getMaxAndMinSeq is arriving", in.String())
resp := new(sdkws.GetMaxAndMinSeqResp) resp := new(sdkws.GetMaxAndMinSeqResp)
m := make(map[string]*sdkws.MaxAndMinSeq) m := make(map[string]*sdkws.MaxAndMinSeq)

27
internal/rpc/msg/utils.go
Unescape View File

@ -1,16 +1,25 @@
package msg package msg
import ( import (
"Open_IM/pkg/utils" "Open_IM/pkg/common/config"
"github.com/go-redis/redis/v8" "Open_IM/pkg/common/constant"
"gorm.io/gorm" "Open_IM/pkg/proto/sdkws"
) )
func IsNotFound(err error) bool { func isMessageHasReadEnabled(msgData *sdkws.MsgData) bool {
switch utils.Unwrap(err) { switch msgData.ContentType {
case gorm.ErrRecordNotFound, redis.Nil: case constant.HasReadReceipt:
return true if config.Config.SingleMessageHasReadReceiptEnable {
default: return true
return false } else {
return false
}
case constant.GroupHasReadReceipt:
if config.Config.GroupMessageHasReadReceiptEnable {
return true
} else {
return false
}
} }
return true
} }

Write Preview
Loading…
Cancel
Save