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

send message parallel

Browse Source
pull/455/head
Gordon 3 years ago committed by Xinwei Xiong(cubxxw-openim)
parent 012b6d09f7
commit 0b68ba5874
1 changed files with 103 additions and 62 deletions
Show Stats Download Patch File Download Diff File

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

@ -18,6 +18,7 @@ import (
"math/rand" "math/rand"
"strconv" "strconv"
"strings" "strings"
"sync"
"time" "time"
) )
@ -253,72 +254,55 @@ func (rpc *rpcChat) SendMsg(_ context.Context, pb *pbChat.SendMsgReq) (*pbChat.S
} }
case constant.MemberQuitNotification: case constant.MemberQuitNotification:
addUidList = append(addUidList, pb.MsgData.SendID) addUidList = append(addUidList, pb.MsgData.SendID)
case constant.AtText:
go func() { default:
var conversationReq pbConversation.ModifyConversationFieldReq }
var tag bool groupID := pb.MsgData.GroupID
var atUserID []string //split parallel send
conversation := pbConversation.Conversation{ var wg sync.WaitGroup
OwnerUserID: pb.MsgData.SendID, var sendTag bool
ConversationID: utils.GetConversationIDBySessionType(pb.MsgData.GroupID, constant.GroupChatType), var split = 50
ConversationType: constant.GroupChatType, remain := len(memberUserIDList) % split
GroupID: pb.MsgData.GroupID, for i := 0; i < len(memberUserIDList)/split; i++ {
} wg.Add(1)
conversationReq.Conversation = &conversation go func(list []string) {
conversationReq.OperationID = pb.OperationID for _, v := range list {
conversationReq.FieldType = constant.FieldGroupAtType pb.MsgData.RecvID = v
tagAll := utils.IsContain(constant.AtAllString, pb.MsgData.AtUserIDList) isSend := modifyMessageByUserMessageReceiveOpt(v, groupID, constant.GroupChatType, pb)
if tagAll { if isSend {
atUserID = utils.DifferenceString([]string{constant.AtAllString}, pb.MsgData.AtUserIDList) msgToMQ.MsgData = pb.MsgData
if len(atUserID) == 0 { //just @everyone err := rpc.sendMsgToKafka(&msgToMQ, v)
conversationReq.UserIDList = memberUserIDList if err != nil {
conversation.GroupAtType = constant.AtAll log.NewError(msgToMQ.OperationID, "kafka send msg err:UserId", v, msgToMQ.String())
} else { //@Everyone and @other people } else {
conversationReq.UserIDList = atUserID sendTag = true
conversation.GroupAtType = constant.AtAllAtMe }
tag = true
} }
} else {
conversationReq.UserIDList = pb.MsgData.AtUserIDList
conversation.GroupAtType = constant.AtMe
}
etcdConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImConversationName)
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 { wg.Done()
conversationReq.UserIDList = utils.DifferenceString(atUserID, memberUserIDList) }(memberUserIDList[i*split : (i+1)*split])
conversation.GroupAtType = constant.AtAll }
etcdConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImConversationName) if remain > 0 {
client := pbConversation.NewConversationClient(etcdConn) wg.Add(1)
conversationReply, err := client.ModifyConversationField(context.Background(), &conversationReq) go func(list []string) {
if err != nil { for _, v := range list {
log.NewError(conversationReq.OperationID, "ModifyConversationField rpc failed, ", conversationReq.String(), err.Error()) pb.MsgData.RecvID = v
} else if conversationReply.CommonResp.ErrCode != 0 { isSend := modifyMessageByUserMessageReceiveOpt(v, groupID, constant.GroupChatType, pb)
log.NewError(conversationReq.OperationID, "ModifyConversationField rpc failed, ", conversationReq.String(), conversationReply.String()) if isSend {
msgToMQ.MsgData = pb.MsgData
err := rpc.sendMsgToKafka(&msgToMQ, v)
if err != nil {
log.NewError(msgToMQ.OperationID, "kafka send msg err:UserId", v, msgToMQ.String())
} else {
sendTag = true
}
} }
} }
}() wg.Done()
default: }(memberUserIDList[split*(len(memberUserIDList)/split):])
} }
groupID := pb.MsgData.GroupID wg.Wait()
for _, v := range memberUserIDList {
pb.MsgData.RecvID = v
isSend := modifyMessageByUserMessageReceiveOpt(v, groupID, constant.GroupChatType, pb)
if isSend {
msgToMQ.MsgData = pb.MsgData
err := rpc.sendMsgToKafka(&msgToMQ, v)
if err != nil {
log.NewError(msgToMQ.OperationID, "kafka send msg err:UserId", v, msgToMQ.String())
return returnMsg(&replay, pb, 201, "kafka send msg err", "", 0)
}
}
}
log.Info(msgToMQ.OperationID, "addUidList", addUidList) log.Info(msgToMQ.OperationID, "addUidList", addUidList)
for _, v := range addUidList { for _, v := range addUidList {
pb.MsgData.RecvID = v pb.MsgData.RecvID = v
@ -329,7 +313,8 @@ func (rpc *rpcChat) SendMsg(_ context.Context, pb *pbChat.SendMsgReq) (*pbChat.S
err := rpc.sendMsgToKafka(&msgToMQ, v) err := rpc.sendMsgToKafka(&msgToMQ, v)
if err != nil { if err != nil {
log.NewError(msgToMQ.OperationID, "kafka send msg err:UserId", v, msgToMQ.String()) log.NewError(msgToMQ.OperationID, "kafka send msg err:UserId", v, msgToMQ.String())
return returnMsg(&replay, pb, 201, "kafka send msg err", "", 0) } else {
sendTag = true
} }
} }
} }
@ -337,7 +322,63 @@ func (rpc *rpcChat) SendMsg(_ context.Context, pb *pbChat.SendMsgReq) (*pbChat.S
if err := callbackAfterSendGroupMsg(pb); err != nil { if err := callbackAfterSendGroupMsg(pb); err != nil {
log.NewError(pb.OperationID, utils.GetSelfFuncName(), "callbackAfterSendGroupMsg failed", err.Error()) log.NewError(pb.OperationID, utils.GetSelfFuncName(), "callbackAfterSendGroupMsg failed", err.Error())
} }
return returnMsg(&replay, pb, 0, "", msgToMQ.MsgData.ServerMsgID, msgToMQ.MsgData.SendTime) if !sendTag {
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 := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImConversationName)
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 := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImConversationName)
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())
}
}
}()
}
return returnMsg(&replay, pb, 0, "", msgToMQ.MsgData.ServerMsgID, msgToMQ.MsgData.SendTime)
}
case constant.NotificationChatType: case constant.NotificationChatType:
msgToMQ.MsgData = pb.MsgData msgToMQ.MsgData = pb.MsgData
log.NewInfo(msgToMQ.OperationID, msgToMQ) log.NewInfo(msgToMQ.OperationID, msgToMQ)

Write Preview
Loading…
Cancel
Save