blackteay/open-im-server
1
0
Fork 0
mirror of https://github.com/openimsdk/open-im-server.git synced 2025-08-31 03:05:43 +08:00
Code Issues Packages Projects Releases Wiki Activity

send msg file modify

Browse Source
This commit is contained in:
Gordon 2021-12-23 17:19:57 +08:00
parent cfddc75ccb
commit 333453df79
1 changed files with 156 additions and 108 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

264
internal/rpc/chat/send_msg.go → internal/rpc/msg/send_msg.go
View File

@ -1,8 +1,6 @@
package chat package msg
import ( import (
"Open_IM/internal/api/group"
"Open_IM/internal/push/content_struct"
"Open_IM/pkg/common/config" "Open_IM/pkg/common/config"
"Open_IM/pkg/common/constant" "Open_IM/pkg/common/constant"
"Open_IM/pkg/common/db" "Open_IM/pkg/common/db"
@ -13,9 +11,11 @@ import (
pbChat "Open_IM/pkg/proto/chat" pbChat "Open_IM/pkg/proto/chat"
pbGroup "Open_IM/pkg/proto/group" pbGroup "Open_IM/pkg/proto/group"
open_im_sdk "Open_IM/pkg/proto/sdk_ws" open_im_sdk "Open_IM/pkg/proto/sdk_ws"
sdk_ws "Open_IM/pkg/proto/sdk_ws"
"Open_IM/pkg/utils" "Open_IM/pkg/utils"
"context" "context"
"encoding/json" "encoding/json"
"github.com/garyburd/redigo/redis"
"math/rand" "math/rand"
"net/http" "net/http"
"strconv" "strconv"
@ -45,46 +45,72 @@ type MsgCallBackResp struct {
} }
} }
func (rpc *rpcChat) UserSendMsg(_ context.Context, pb *pbChat.UserSendMsgReq) (*pbChat.UserSendMsgResp, error) { func (rpc *rpcChat) encapsulateMsgData(msg *sdk_ws.MsgData) {
replay := pbChat.UserSendMsgResp{} msg.ServerMsgID = GetMsgID(msg.SendID)
if msg.SendTime == 0 {
msg.SendTime = utils.GetCurrentTimestampByNano()
}
switch msg.ContentType {
case constant.Text:
fallthrough
case constant.Picture:
fallthrough
case constant.Voice:
fallthrough
case constant.Video:
fallthrough
case constant.File:
fallthrough
case constant.AtText:
fallthrough
case constant.Merger:
fallthrough
case constant.Card:
fallthrough
case constant.Location:
fallthrough
case constant.Custom:
fallthrough
case constant.Quote:
utils.SetSwitchFromOptions(msg.Options, constant.IsConversationUpdate, true)
utils.SetSwitchFromOptions(msg.Options, constant.IsUnreadCount, true)
utils.SetSwitchFromOptions(msg.Options, constant.IsSenderSync, true)
case constant.Revoke:
utils.SetSwitchFromOptions(msg.Options, constant.IsUnreadCount, false)
utils.SetSwitchFromOptions(msg.Options, constant.IsOfflinePush, false)
case constant.HasReadReceipt:
utils.SetSwitchFromOptions(msg.Options, constant.IsConversationUpdate, false)
utils.SetSwitchFromOptions(msg.Options, constant.IsUnreadCount, false)
utils.SetSwitchFromOptions(msg.Options, constant.IsOfflinePush, false)
case constant.Typing:
utils.SetSwitchFromOptions(msg.Options, constant.IsHistory, false)
utils.SetSwitchFromOptions(msg.Options, constant.IsPersistent, false)
utils.SetSwitchFromOptions(msg.Options, constant.IsSenderSync, false)
utils.SetSwitchFromOptions(msg.Options, constant.IsConversationUpdate, false)
utils.SetSwitchFromOptions(msg.Options, constant.IsUnreadCount, false)
utils.SetSwitchFromOptions(msg.Options, constant.IsOfflinePush, false)
}
}
func (rpc *rpcChat) SendMsg(_ context.Context, pb *pbChat.SendMsgReq) (*pbChat.SendMsgResp, error) {
replay := pbChat.SendMsgResp{}
log.NewDebug(pb.OperationID, "rpc sendMsg come here", pb.String()) log.NewDebug(pb.OperationID, "rpc sendMsg come here", pb.String())
//if !utils.VerifyToken(pb.Token, pb.SendID) { //if !utils.VerifyToken(pb.Token, pb.SendID) {
// return returnMsg(&replay, pb, http.StatusUnauthorized, "token validate err,not authorized", "", 0) // return returnMsg(&replay, pb, http.StatusUnauthorized, "token validate err,not authorized", "", 0)
serverMsgID := GetMsgID(pb.SendID) rpc.encapsulateMsgData(pb.MsgData)
pbData := pbChat.WSToMsgSvrChatMsg{} msgToMQ := pbChat.MsgDataToMQ{Token: pb.Token, OperationID: pb.OperationID}
pbData.MsgFrom = pb.MsgFrom //options := utils.JsonStringToMap(pbData.Options)
pbData.SessionType = pb.SessionType isHistory := utils.GetSwitchFromOptions(pb.MsgData.Options, constant.IsHistory)
pbData.ContentType = pb.ContentType
pbData.Content = pb.Content
pbData.RecvID = pb.RecvID
pbData.ForceList = pb.ForceList
pbData.OfflineInfo = pb.OffLineInfo
pbData.Options = pb.Options
pbData.PlatformID = pb.PlatformID
pbData.ClientMsgID = pb.ClientMsgID
pbData.SendID = pb.SendID
pbData.SenderNickName = pb.SenderNickName
pbData.SenderFaceURL = pb.SenderFaceURL
pbData.MsgID = serverMsgID
pbData.OperationID = pb.OperationID
pbData.Token = pb.Token
if pb.SendTime == 0 {
pbData.SendTime = utils.GetCurrentTimestampByNano()
} else {
pbData.SendTime = pb.SendTime
}
options := utils.JsonStringToMap(pbData.Options)
isHistory := utils.GetSwitchFromOptions(options, "history")
mReq := MsgCallBackReq{ mReq := MsgCallBackReq{
SendID: pb.SendID, SendID: pb.MsgData.SendID,
RecvID: pb.RecvID, RecvID: pb.MsgData.RecvID,
Content: pb.Content, Content: string(pb.MsgData.Content),
SendTime: pbData.SendTime, SendTime: pb.MsgData.SendTime,
MsgFrom: pbData.MsgFrom, MsgFrom: pb.MsgData.MsgFrom,
ContentType: pb.ContentType, ContentType: pb.MsgData.ContentType,
SessionType: pb.SessionType, SessionType: pb.MsgData.SessionType,
PlatformID: pb.PlatformID, PlatformID: pb.MsgData.SenderPlatformID,
MsgID: pb.ClientMsgID, MsgID: pb.MsgData.ClientMsgID,
} }
if !isHistory { if !isHistory {
mReq.IsOnlineOnly = true mReq.IsOnlineOnly = true
@ -102,97 +128,65 @@ func (rpc *rpcChat) UserSendMsg(_ context.Context, pb *pbChat.UserSendMsgReq) (*
if mResp.ErrCode != 0 { if mResp.ErrCode != 0 {
return returnMsg(&replay, pb, mResp.ResponseErrCode, mResp.ErrMsg, "", 0) return returnMsg(&replay, pb, mResp.ResponseErrCode, mResp.ErrMsg, "", 0)
} else { } else {
pbData.Content = mResp.ResponseResult.ModifiedMsg pb.MsgData.Content = []byte(mResp.ResponseResult.ModifiedMsg)
} }
} }
} }
switch pbData.SessionType { switch pb.MsgData.SessionType {
case constant.SingleChatType: case constant.SingleChatType:
isSend := modifyMessageByUserMessageReceiveOpt(pbData.RecvID, pbData.SendID, constant.SingleChatType, &pbData) isSend := modifyMessageByUserMessageReceiveOpt(pb.MsgData.RecvID, pb.MsgData.SendID, constant.SingleChatType, pb)
if isSend { if isSend {
err1 := rpc.sendMsgToKafka(&pbData, pbData.RecvID) msgToMQ.MsgData = pb.MsgData
err1 := rpc.sendMsgToKafka(&msgToMQ, msgToMQ.MsgData.RecvID)
if err1 != nil { if err1 != nil {
log.NewError(pbData.OperationID, "kafka send msg err:RecvID", pbData.RecvID, pbData.String()) log.NewError(msgToMQ.OperationID, "kafka send msg err:RecvID", msgToMQ.MsgData.RecvID, msgToMQ.String())
return returnMsg(&replay, pb, 201, "kafka send msg err", "", 0) return returnMsg(&replay, pb, 201, "kafka send msg err", "", 0)
} }
} }
err2 := rpc.sendMsgToKafka(&pbData, pbData.SendID) err2 := rpc.sendMsgToKafka(&msgToMQ, msgToMQ.MsgData.SendID)
if err2 != nil { if err2 != nil {
log.NewError(pbData.OperationID, "kafka send msg err:SendID", pbData.SendID, pbData.String()) log.NewError(msgToMQ.OperationID, "kafka send msg err:SendID", msgToMQ.MsgData.SendID, msgToMQ.String())
return returnMsg(&replay, pb, 201, "kafka send msg err", "", 0) return returnMsg(&replay, pb, 201, "kafka send msg err", "", 0)
} }
return returnMsg(&replay, pb, 0, "", serverMsgID, pbData.SendTime) return returnMsg(&replay, pb, 0, "", msgToMQ.MsgData.ServerMsgID, msgToMQ.MsgData.SendTime)
case constant.GroupChatType: case constant.GroupChatType:
etcdConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImGroupName) etcdConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImGroupName)
client := pbGroup.NewGroupClient(etcdConn) client := pbGroup.NewGroupClient(etcdConn)
req := &pbGroup.GetGroupAllMemberReq{ req := &pbGroup.GetGroupAllMemberReq{
GroupID: pbData.RecvID, GroupID: pb.MsgData.GroupID,
Token: pbData.Token, Token: pb.Token,
OperationID: pbData.OperationID, OperationID: pb.OperationID,
} }
reply, err := client.GetGroupAllMember(context.Background(), req) reply, err := client.GetGroupAllMember(context.Background(), req)
if err != nil { if err != nil {
log.Error(pbData.Token, pbData.OperationID, "rpc send_msg getGroupInfo failed, err = %s", err.Error()) log.Error(pb.Token, pb.OperationID, "rpc send_msg getGroupInfo failed, err = %s", err.Error())
return returnMsg(&replay, pb, 201, err.Error(), "", 0) return returnMsg(&replay, pb, 201, err.Error(), "", 0)
} }
if reply.ErrorCode != 0 { if reply.ErrCode != 0 {
log.Error(pbData.Token, pbData.OperationID, "rpc send_msg getGroupInfo failed, err = %s", reply.ErrorMsg) log.Error(pb.Token, pb.OperationID, "rpc send_msg getGroupInfo failed, err = %s", reply.ErrMsg)
return returnMsg(&replay, pb, reply.ErrorCode, reply.ErrorMsg, "", 0) return returnMsg(&replay, pb, reply.ErrCode, reply.ErrMsg, "", 0)
} }
var addUidList []string groupID := pb.MsgData.GroupID
switch pbData.ContentType { for _, v := range reply.MemberList {
case constant.KickGroupMemberTip: pb.MsgData.RecvID = v.UserId
var notification content_struct.NotificationContent isSend := modifyMessageByUserMessageReceiveOpt(v.UserId, groupID, constant.GroupChatType, pb)
var kickContent group.KickGroupMemberReq
err := utils.JsonStringToStruct(pbData.Content, &notification)
if err != nil {
log.ErrorByKv("json unmarshall err", pbData.OperationID, "err", err.Error())
return returnMsg(&replay, pb, 200, err.Error(), "", 0)
} else {
err := utils.JsonStringToStruct(notification.Detail, &kickContent)
if err != nil {
log.ErrorByKv("json unmarshall err", pbData.OperationID, "err", err.Error())
return returnMsg(&replay, pb, 200, err.Error(), "", 0)
}
for _, v := range kickContent.UidListInfo {
addUidList = append(addUidList, v.UserId)
}
}
case constant.QuitGroupTip:
addUidList = append(addUidList, pbData.SendID)
default:
}
groupID := pbData.RecvID
for i, v := range reply.MemberList {
pbData.RecvID = v.UserId + " " + groupID
isSend := modifyMessageByUserMessageReceiveOpt(v.UserId, groupID, constant.GroupChatType, &pbData)
if isSend { if isSend {
err := rpc.sendMsgToKafka(&pbData, utils.IntToString(i)) msgToMQ.MsgData = pb.MsgData
err := rpc.sendMsgToKafka(&msgToMQ, v.UserId)
if err != nil { if err != nil {
log.NewError(pbData.OperationID, "kafka send msg err:UserId", v.UserId, pbData.String()) log.NewError(msgToMQ.OperationID, "kafka send msg err:UserId", v.UserId, msgToMQ.String())
return returnMsg(&replay, pb, 201, "kafka send msg err", "", 0) return returnMsg(&replay, pb, 201, "kafka send msg err", "", 0)
} }
} }
} }
for i, v := range addUidList { return returnMsg(&replay, pb, 0, "", msgToMQ.MsgData.ServerMsgID, msgToMQ.MsgData.SendTime)
pbData.RecvID = v + " " + groupID
isSend := modifyMessageByUserMessageReceiveOpt(v, groupID, constant.GroupChatType, &pbData)
if isSend {
err := rpc.sendMsgToKafka(&pbData, utils.IntToString(i+1))
if err != nil {
log.NewError(pbData.OperationID, "kafka send msg err:UserId", v, pbData.String())
return returnMsg(&replay, pb, 201, "kafka send msg err", "", 0)
}
}
}
return returnMsg(&replay, pb, 0, "", serverMsgID, pbData.SendTime)
default: default:
return returnMsg(&replay, pb, 203, "unkonwn sessionType", "", 0) return returnMsg(&replay, pb, 203, "unkonwn sessionType", "", 0)
} }
} }
func (rpc *rpcChat) sendMsgToKafka(m *pbChat.WSToMsgSvrChatMsg, key string) error { func (rpc *rpcChat) sendMsgToKafka(m *pbChat.MsgDataToMQ, key string) error {
pid, offset, err := rpc.producer.SendMessage(m, key) pid, offset, err := rpc.producer.SendMessage(m, key)
if err != nil { if err != nil {
log.ErrorByKv("kafka send failed", m.OperationID, "send data", m.String(), "pid", pid, "offset", offset, "err", err.Error(), "key", key) log.ErrorByKv("kafka send failed", m.OperationID, "send data", m.String(), "pid", pid, "offset", offset, "err", err.Error(), "key", key)
@ -204,21 +198,20 @@ func GetMsgID(sendID string) string {
return t + "-" + sendID + "-" + strconv.Itoa(rand.Int()) return t + "-" + sendID + "-" + strconv.Itoa(rand.Int())
} }
func returnMsg(replay *pbChat.UserSendMsgResp, pb *pbChat.UserSendMsgReq, errCode int32, errMsg, serverMsgID string, sendTime int64) (*pbChat.UserSendMsgResp, error) { func returnMsg(replay *pbChat.SendMsgResp, pb *pbChat.SendMsgReq, errCode int32, errMsg, serverMsgID string, sendTime int64) (*pbChat.SendMsgResp, error) {
replay.ErrCode = errCode replay.ErrCode = errCode
replay.ErrMsg = errMsg replay.ErrMsg = errMsg
replay.ReqIdentifier = pb.ReqIdentifier
replay.ClientMsgID = pb.ClientMsgID
replay.ServerMsgID = serverMsgID replay.ServerMsgID = serverMsgID
replay.ClientMsgID = pb.MsgData.ClientMsgID
replay.SendTime = sendTime replay.SendTime = sendTime
return replay, nil return replay, nil
} }
func modifyMessageByUserMessageReceiveOpt(userID, sourceID string, sessionType int, msg *pbChat.WSToMsgSvrChatMsg) bool { func modifyMessageByUserMessageReceiveOpt(userID, sourceID string, sessionType int, pb *pbChat.SendMsgReq) bool {
conversationID := utils.GetConversationIDBySessionType(sourceID, sessionType) conversationID := utils.GetConversationIDBySessionType(sourceID, sessionType)
opt, err := db.DB.GetSingleConversationMsgOpt(userID, conversationID) opt, err := db.DB.GetSingleConversationMsgOpt(userID, conversationID)
if err != nil { if err != nil || err != redis.ErrNil {
log.NewError(msg.OperationID, "GetSingleConversationMsgOpt from redis err", msg.String()) log.NewError(pb.OperationID, "GetSingleConversationMsgOpt from redis err", pb.String())
return true return true
} }
switch opt { switch opt {
@ -227,12 +220,10 @@ func modifyMessageByUserMessageReceiveOpt(userID, sourceID string, sessionType i
case constant.NotReceiveMessage: case constant.NotReceiveMessage:
return false return false
case constant.ReceiveNotNotifyMessage: case constant.ReceiveNotNotifyMessage:
options := utils.JsonStringToMap(msg.Options) if pb.MsgData.Options == nil {
if options == nil { pb.MsgData.Options = make(map[string]bool, 10)
options = make(map[string]int32, 2)
} }
utils.SetSwitchFromOptions(options, "offlinePush", 0) utils.SetSwitchFromOptions(pb.MsgData.Options, constant.IsOfflinePush, false)
msg.Options = utils.MapIntToJsonString(options)
return true return true
} }
@ -250,7 +241,64 @@ type NotificationMsg struct {
} }
func Notification(n *NotificationMsg, onlineUserOnly bool) { func Notification(n *NotificationMsg, onlineUserOnly bool) {
var req pbChat.SendMsgReq
var msg sdk_ws.MsgData
var offlineInfo sdk_ws.OfflinePushInfo
var title, desc, ext string
var pushSwitch bool
req.OperationID = n.OperationID
msg.SendID = n.SendID
msg.RecvID = n.RecvID
msg.Content = n.Content
msg.MsgFrom = n.MsgFrom
msg.ContentType = n.ContentType
msg.SessionType = n.SessionType
msg.CreateTime = utils.GetCurrentTimestampByNano()
msg.ClientMsgID = utils.GetMsgID(n.SendID)
switch n.SessionType {
case constant.GroupChatType:
msg.RecvID = ""
msg.GroupID = n.RecvID
}
if onlineUserOnly {
msg.Options = make(map[string]bool, 10)
utils.SetSwitchFromOptions(msg.Options, constant.IsOfflinePush, false)
utils.SetSwitchFromOptions(msg.Options, constant.IsHistory, false)
utils.SetSwitchFromOptions(msg.Options, constant.IsPersistent, false)
}
offlineInfo.IOSBadgeCount = config.Config.IOSPush.BadgeCount
offlineInfo.IOSPushSound = config.Config.IOSPush.PushSound
switch msg.ContentType {
case constant.CreateGroupTip:
pushSwitch = config.Config.Notification.GroupCreated.OfflinePush.PushSwitch
title = config.Config.Notification.GroupCreated.OfflinePush.Title
desc = config.Config.Notification.GroupCreated.OfflinePush.Desc
ext = config.Config.Notification.GroupCreated.OfflinePush.Ext
case constant.ChangeGroupInfoTip:
pushSwitch = config.Config.Notification.GroupInfoChanged.OfflinePush.PushSwitch
title = config.Config.Notification.GroupInfoChanged.OfflinePush.Title
desc = config.Config.Notification.GroupInfoChanged.OfflinePush.Desc
ext = config.Config.Notification.GroupInfoChanged.OfflinePush.Ext
case constant.ApplyJoinGroupTip:
pushSwitch = config.Config.Notification.ApplyJoinGroup.OfflinePush.PushSwitch
title = config.Config.Notification.ApplyJoinGroup.OfflinePush.Title
desc = config.Config.Notification.ApplyJoinGroup.OfflinePush.Desc
ext = config.Config.Notification.ApplyJoinGroup.OfflinePush.Ext
}
utils.SetSwitchFromOptions(msg.Options, constant.IsOfflinePush, pushSwitch)
offlineInfo.Title = title
offlineInfo.Desc = desc
offlineInfo.Ext = ext
msg.OfflinePushInfo = &offlineInfo
req.MsgData = &msg
etcdConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImOfflineMessageName)
client := pbChat.NewChatClient(etcdConn)
reply, err := client.SendMsg(context.Background(), &req)
if err != nil {
log.NewError(req.OperationID, "SendMsg rpc failed, ", req.String(), err.Error())
} else if reply.ErrCode != 0 {
log.NewError(req.OperationID, "SendMsg rpc failed, ", req.String())
}
} }
//message GroupCreatedTips{ //message GroupCreatedTips{