package msg import ( "Open_IM/pkg/common/config" "Open_IM/pkg/common/constant" "Open_IM/pkg/common/db" http2 "Open_IM/pkg/common/http" "Open_IM/pkg/common/log" "Open_IM/pkg/grpc-etcdv3/getcdv3" pbChat "Open_IM/pkg/proto/chat" pbGroup "Open_IM/pkg/proto/group" sdk_ws "Open_IM/pkg/proto/sdk_ws" "Open_IM/pkg/utils" "context" "encoding/json" "github.com/garyburd/redigo/redis" "github.com/golang/protobuf/proto" "math/rand" "net/http" "strconv" "strings" "time" ) type MsgCallBackReq struct { SendID string `json:"sendID"` RecvID string `json:"recvID"` Content string `json:"content"` SendTime int64 `json:"sendTime"` MsgFrom int32 `json:"msgFrom"` ContentType int32 `json:"contentType"` SessionType int32 `json:"sessionType"` PlatformID int32 `json:"senderPlatformID"` MsgID string `json:"msgID"` IsOnlineOnly bool `json:"isOnlineOnly"` } type MsgCallBackResp struct { ErrCode int32 `json:"errCode"` ErrMsg string `json:"errMsg"` ResponseErrCode int32 `json:"responseErrCode"` ResponseResult struct { ModifiedMsg string `json:"modifiedMsg"` Ext string `json:"ext"` } } func (rpc *rpcChat) encapsulateMsgData(msg *sdk_ws.MsgData) { msg.ServerMsgID = GetMsgID(msg.SendID) if msg.SendTime == 0 { msg.SendTime = utils.GetCurrentTimestampByMill() } 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()) //if !utils.VerifyToken(pb.Token, pb.SendID) { // return returnMsg(&replay, pb, http.StatusUnauthorized, "token validate err,not authorized", "", 0) rpc.encapsulateMsgData(pb.MsgData) msgToMQ := pbChat.MsgDataToMQ{Token: pb.Token, OperationID: pb.OperationID} //options := utils.JsonStringToMap(pbData.Options) isHistory := utils.GetSwitchFromOptions(pb.MsgData.Options, constant.IsHistory) mReq := MsgCallBackReq{ SendID: pb.MsgData.SendID, RecvID: pb.MsgData.RecvID, Content: string(pb.MsgData.Content), SendTime: pb.MsgData.SendTime, MsgFrom: pb.MsgData.MsgFrom, ContentType: pb.MsgData.ContentType, SessionType: pb.MsgData.SessionType, PlatformID: pb.MsgData.SenderPlatformID, MsgID: pb.MsgData.ClientMsgID, } if !isHistory { mReq.IsOnlineOnly = true } mResp := MsgCallBackResp{} if config.Config.MessageCallBack.CallbackSwitch { bMsg, err := http2.Post(config.Config.MessageCallBack.CallbackUrl, mReq, config.Config.MessageCallBack.CallBackTimeOut) if err != nil { log.ErrorByKv("callback to Business server err", pb.OperationID, "args", pb.String(), "err", err.Error()) return returnMsg(&replay, pb, http.StatusInternalServerError, http.StatusText(http.StatusInternalServerError), "", 0) } else if err = json.Unmarshal(bMsg, &mResp); err != nil { log.ErrorByKv("ws json Unmarshal err", pb.OperationID, "args", pb.String(), "err", err.Error()) return returnMsg(&replay, pb, 200, err.Error(), "", 0) } else { if mResp.ErrCode != 0 { return returnMsg(&replay, pb, mResp.ResponseErrCode, mResp.ErrMsg, "", 0) } else { pb.MsgData.Content = []byte(mResp.ResponseResult.ModifiedMsg) } } } switch pb.MsgData.SessionType { case constant.SingleChatType: isSend := modifyMessageByUserMessageReceiveOpt(pb.MsgData.RecvID, pb.MsgData.SendID, constant.SingleChatType, pb) if isSend { msgToMQ.MsgData = pb.MsgData err1 := rpc.sendMsgToKafka(&msgToMQ, msgToMQ.MsgData.RecvID) if err1 != nil { log.NewError(msgToMQ.OperationID, "kafka send msg err:RecvID", msgToMQ.MsgData.RecvID, msgToMQ.String()) return returnMsg(&replay, pb, 201, "kafka send msg err", "", 0) } } if msgToMQ.MsgData.SendID != msgToMQ.MsgData.RecvID { //Filter messages sent to yourself err2 := rpc.sendMsgToKafka(&msgToMQ, msgToMQ.MsgData.SendID) if err2 != nil { 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, 0, "", msgToMQ.MsgData.ServerMsgID, msgToMQ.MsgData.SendTime) case constant.GroupChatType: etcdConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImGroupName) client := pbGroup.NewGroupClient(etcdConn) req := &pbGroup.GetGroupAllMemberReq{ GroupID: pb.MsgData.GroupID, OperationID: pb.OperationID, } reply, err := client.GetGroupAllMember(context.Background(), req) if err != nil { log.Error(pb.Token, pb.OperationID, "rpc send_msg getGroupInfo failed, err = %s", err.Error()) return returnMsg(&replay, pb, 201, err.Error(), "", 0) } if reply.ErrCode != 0 { log.Error(pb.Token, pb.OperationID, "rpc send_msg getGroupInfo failed, err = %s", reply.ErrMsg) return returnMsg(&replay, pb, reply.ErrCode, reply.ErrMsg, "", 0) } var addUidList []string switch pb.MsgData.ContentType { case constant.MemberKickedNotification: var tips sdk_ws.TipsComm var memberKickedTips sdk_ws.MemberKickedTips _ = proto.Unmarshal(pb.MsgData.Content, &tips) _ = proto.Unmarshal(tips.Detail, &memberKickedTips) for _, v := range memberKickedTips.KickedUserList { addUidList = append(addUidList, v.UserID) } case constant.MemberQuitNotification: addUidList = append(addUidList, pb.MsgData.SendID) default: } groupID := pb.MsgData.GroupID for _, v := range reply.MemberList { pb.MsgData.RecvID = v.UserID isSend := modifyMessageByUserMessageReceiveOpt(v.UserID, groupID, constant.GroupChatType, pb) if isSend { msgToMQ.MsgData = pb.MsgData err := rpc.sendMsgToKafka(&msgToMQ, v.UserID) if err != nil { log.NewError(msgToMQ.OperationID, "kafka send msg err:UserId", v.UserID, msgToMQ.String()) return returnMsg(&replay, pb, 201, "kafka send msg err", "", 0) } } } for _, v := range addUidList { 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) } } } return returnMsg(&replay, pb, 0, "", msgToMQ.MsgData.ServerMsgID, msgToMQ.MsgData.SendTime) default: return returnMsg(&replay, pb, 203, "unkonwn sessionType", "", 0) } } func (rpc *rpcChat) sendMsgToKafka(m *pbChat.MsgDataToMQ, key string) error { pid, offset, err := rpc.producer.SendMessage(m, key) if err != nil { log.ErrorByKv("kafka send failed", m.OperationID, "send data", m.String(), "pid", pid, "offset", offset, "err", err.Error(), "key", key) } return err } func GetMsgID(sendID string) string { t := time.Now().Format("2006-01-02 15:04:05") return t + "-" + sendID + "-" + strconv.Itoa(rand.Int()) } func returnMsg(replay *pbChat.SendMsgResp, pb *pbChat.SendMsgReq, errCode int32, errMsg, serverMsgID string, sendTime int64) (*pbChat.SendMsgResp, error) { replay.ErrCode = errCode replay.ErrMsg = errMsg replay.ServerMsgID = serverMsgID replay.ClientMsgID = pb.MsgData.ClientMsgID replay.SendTime = sendTime return replay, nil } func modifyMessageByUserMessageReceiveOpt(userID, sourceID string, sessionType int, pb *pbChat.SendMsgReq) bool { conversationID := utils.GetConversationIDBySessionType(sourceID, sessionType) opt, err := db.DB.GetSingleConversationMsgOpt(userID, conversationID) if err != nil || err != redis.ErrNil { log.NewError(pb.OperationID, "GetSingleConversationMsgOpt from redis err", conversationID, pb.String(), err.Error()) return true } switch opt { case constant.ReceiveMessage: return true case constant.NotReceiveMessage: return false case constant.ReceiveNotNotifyMessage: if pb.MsgData.Options == nil { pb.MsgData.Options = make(map[string]bool, 10) } utils.SetSwitchFromOptions(pb.MsgData.Options, constant.IsOfflinePush, false) return true } return true } type NotificationMsg struct { SendID string RecvID string Content []byte // open_im_sdk.TipsComm MsgFrom int32 ContentType int32 SessionType int32 OperationID string } func Notification(n *NotificationMsg) { var req pbChat.SendMsgReq var msg sdk_ws.MsgData var offlineInfo sdk_ws.OfflinePushInfo var title, desc, ex string var pushSwitch, unReadCount bool var reliabilityLevel int 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.GetCurrentTimestampByMill() msg.ClientMsgID = utils.GetMsgID(n.SendID) msg.Options = make(map[string]bool, 7) switch n.SessionType { case constant.GroupChatType: msg.RecvID = "" msg.GroupID = n.RecvID } offlineInfo.IOSBadgeCount = config.Config.IOSPush.BadgeCount offlineInfo.IOSPushSound = config.Config.IOSPush.PushSound switch msg.ContentType { case constant.GroupCreatedNotification: pushSwitch = config.Config.Notification.GroupCreated.OfflinePush.PushSwitch title = config.Config.Notification.GroupCreated.OfflinePush.Title desc = config.Config.Notification.GroupCreated.OfflinePush.Desc ex = config.Config.Notification.GroupCreated.OfflinePush.Ext reliabilityLevel = config.Config.Notification.GroupCreated.Conversation.ReliabilityLevel unReadCount = config.Config.Notification.GroupCreated.Conversation.UnreadCount case constant.GroupInfoSetNotification: pushSwitch = config.Config.Notification.GroupInfoSet.OfflinePush.PushSwitch title = config.Config.Notification.GroupInfoSet.OfflinePush.Title desc = config.Config.Notification.GroupInfoSet.OfflinePush.Desc ex = config.Config.Notification.GroupInfoSet.OfflinePush.Ext reliabilityLevel = config.Config.Notification.GroupInfoSet.Conversation.ReliabilityLevel unReadCount = config.Config.Notification.GroupInfoSet.Conversation.UnreadCount case constant.JoinGroupApplicationNotification: pushSwitch = config.Config.Notification.JoinGroupApplication.OfflinePush.PushSwitch title = config.Config.Notification.JoinGroupApplication.OfflinePush.Title desc = config.Config.Notification.JoinGroupApplication.OfflinePush.Desc ex = config.Config.Notification.JoinGroupApplication.OfflinePush.Ext reliabilityLevel = config.Config.Notification.JoinGroupApplication.Conversation.ReliabilityLevel unReadCount = config.Config.Notification.JoinGroupApplication.Conversation.UnreadCount case constant.MemberQuitNotification: pushSwitch = config.Config.Notification.MemberQuit.OfflinePush.PushSwitch title = config.Config.Notification.MemberQuit.OfflinePush.Title desc = config.Config.Notification.MemberQuit.OfflinePush.Desc ex = config.Config.Notification.MemberQuit.OfflinePush.Ext reliabilityLevel = config.Config.Notification.MemberQuit.Conversation.ReliabilityLevel unReadCount = config.Config.Notification.MemberQuit.Conversation.UnreadCount case constant.GroupApplicationAcceptedNotification: pushSwitch = config.Config.Notification.GroupApplicationAccepted.OfflinePush.PushSwitch title = config.Config.Notification.GroupApplicationAccepted.OfflinePush.Title desc = config.Config.Notification.GroupApplicationAccepted.OfflinePush.Desc ex = config.Config.Notification.GroupApplicationAccepted.OfflinePush.Ext reliabilityLevel = config.Config.Notification.GroupApplicationAccepted.Conversation.ReliabilityLevel unReadCount = config.Config.Notification.GroupApplicationAccepted.Conversation.UnreadCount case constant.GroupApplicationRejectedNotification: pushSwitch = config.Config.Notification.GroupApplicationRejected.OfflinePush.PushSwitch title = config.Config.Notification.GroupApplicationRejected.OfflinePush.Title desc = config.Config.Notification.GroupApplicationRejected.OfflinePush.Desc ex = config.Config.Notification.GroupApplicationRejected.OfflinePush.Ext reliabilityLevel = config.Config.Notification.GroupApplicationRejected.Conversation.ReliabilityLevel unReadCount = config.Config.Notification.GroupApplicationRejected.Conversation.UnreadCount case constant.GroupOwnerTransferredNotification: pushSwitch = config.Config.Notification.GroupOwnerTransferred.OfflinePush.PushSwitch title = config.Config.Notification.GroupOwnerTransferred.OfflinePush.Title desc = config.Config.Notification.GroupOwnerTransferred.OfflinePush.Desc ex = config.Config.Notification.GroupOwnerTransferred.OfflinePush.Ext reliabilityLevel = config.Config.Notification.GroupOwnerTransferred.Conversation.ReliabilityLevel unReadCount = config.Config.Notification.GroupOwnerTransferred.Conversation.UnreadCount case constant.MemberKickedNotification: pushSwitch = config.Config.Notification.MemberKicked.OfflinePush.PushSwitch title = config.Config.Notification.MemberKicked.OfflinePush.Title desc = config.Config.Notification.MemberKicked.OfflinePush.Desc ex = config.Config.Notification.MemberKicked.OfflinePush.Ext reliabilityLevel = config.Config.Notification.MemberKicked.Conversation.ReliabilityLevel unReadCount = config.Config.Notification.MemberKicked.Conversation.UnreadCount case constant.MemberInvitedNotification: pushSwitch = config.Config.Notification.MemberInvited.OfflinePush.PushSwitch title = config.Config.Notification.MemberInvited.OfflinePush.Title desc = config.Config.Notification.MemberInvited.OfflinePush.Desc ex = config.Config.Notification.MemberInvited.OfflinePush.Ext reliabilityLevel = config.Config.Notification.MemberInvited.Conversation.ReliabilityLevel unReadCount = config.Config.Notification.MemberInvited.Conversation.UnreadCount case constant.MemberEnterNotification: pushSwitch = config.Config.Notification.MemberEnter.OfflinePush.PushSwitch title = config.Config.Notification.MemberEnter.OfflinePush.Title desc = config.Config.Notification.MemberEnter.OfflinePush.Desc ex = config.Config.Notification.MemberEnter.OfflinePush.Ext reliabilityLevel = config.Config.Notification.MemberEnter.Conversation.ReliabilityLevel unReadCount = config.Config.Notification.MemberEnter.Conversation.UnreadCount case constant.UserInfoUpdatedNotification: pushSwitch = config.Config.Notification.UserInfoUpdated.OfflinePush.PushSwitch title = config.Config.Notification.UserInfoUpdated.OfflinePush.Title desc = config.Config.Notification.UserInfoUpdated.OfflinePush.Desc ex = config.Config.Notification.UserInfoUpdated.OfflinePush.Ext reliabilityLevel = config.Config.Notification.UserInfoUpdated.Conversation.ReliabilityLevel unReadCount = config.Config.Notification.UserInfoUpdated.Conversation.UnreadCount case constant.FriendApplicationNotification: pushSwitch = config.Config.Notification.FriendApplication.OfflinePush.PushSwitch title = config.Config.Notification.FriendApplication.OfflinePush.Title desc = config.Config.Notification.FriendApplication.OfflinePush.Desc ex = config.Config.Notification.FriendApplication.OfflinePush.Ext reliabilityLevel = config.Config.Notification.FriendApplication.Conversation.ReliabilityLevel unReadCount = config.Config.Notification.FriendApplication.Conversation.UnreadCount case constant.FriendApplicationApprovedNotification: pushSwitch = config.Config.Notification.FriendApplicationApproved.OfflinePush.PushSwitch title = config.Config.Notification.FriendApplicationApproved.OfflinePush.Title desc = config.Config.Notification.FriendApplicationApproved.OfflinePush.Desc ex = config.Config.Notification.FriendApplicationApproved.OfflinePush.Ext reliabilityLevel = config.Config.Notification.FriendApplicationApproved.Conversation.ReliabilityLevel unReadCount = config.Config.Notification.FriendApplicationApproved.Conversation.UnreadCount case constant.FriendApplicationRejectedNotification: pushSwitch = config.Config.Notification.FriendApplicationRejected.OfflinePush.PushSwitch title = config.Config.Notification.FriendApplicationRejected.OfflinePush.Title desc = config.Config.Notification.FriendApplicationRejected.OfflinePush.Desc ex = config.Config.Notification.FriendApplicationRejected.OfflinePush.Ext reliabilityLevel = config.Config.Notification.FriendApplicationRejected.Conversation.ReliabilityLevel unReadCount = config.Config.Notification.FriendApplicationRejected.Conversation.UnreadCount case constant.FriendAddedNotification: pushSwitch = config.Config.Notification.FriendAdded.OfflinePush.PushSwitch title = config.Config.Notification.FriendAdded.OfflinePush.Title desc = config.Config.Notification.FriendAdded.OfflinePush.Desc ex = config.Config.Notification.FriendAdded.OfflinePush.Ext reliabilityLevel = config.Config.Notification.FriendAdded.Conversation.ReliabilityLevel unReadCount = config.Config.Notification.FriendAdded.Conversation.UnreadCount case constant.FriendDeletedNotification: pushSwitch = config.Config.Notification.FriendDeleted.OfflinePush.PushSwitch title = config.Config.Notification.FriendDeleted.OfflinePush.Title desc = config.Config.Notification.FriendDeleted.OfflinePush.Desc ex = config.Config.Notification.FriendDeleted.OfflinePush.Ext reliabilityLevel = config.Config.Notification.FriendDeleted.Conversation.ReliabilityLevel unReadCount = config.Config.Notification.FriendDeleted.Conversation.UnreadCount case constant.FriendRemarkSetNotification: pushSwitch = config.Config.Notification.FriendRemarkSet.OfflinePush.PushSwitch title = config.Config.Notification.FriendRemarkSet.OfflinePush.Title desc = config.Config.Notification.FriendRemarkSet.OfflinePush.Desc ex = config.Config.Notification.FriendRemarkSet.OfflinePush.Ext reliabilityLevel = config.Config.Notification.FriendRemarkSet.Conversation.ReliabilityLevel unReadCount = config.Config.Notification.FriendRemarkSet.Conversation.UnreadCount case constant.BlackAddedNotification: pushSwitch = config.Config.Notification.BlackAdded.OfflinePush.PushSwitch title = config.Config.Notification.BlackAdded.OfflinePush.Title desc = config.Config.Notification.BlackAdded.OfflinePush.Desc ex = config.Config.Notification.BlackAdded.OfflinePush.Ext reliabilityLevel = config.Config.Notification.BlackAdded.Conversation.ReliabilityLevel unReadCount = config.Config.Notification.BlackAdded.Conversation.UnreadCount case constant.BlackDeletedNotification: pushSwitch = config.Config.Notification.BlackDeleted.OfflinePush.PushSwitch title = config.Config.Notification.BlackDeleted.OfflinePush.Title desc = config.Config.Notification.BlackDeleted.OfflinePush.Desc ex = config.Config.Notification.BlackDeleted.OfflinePush.Ext reliabilityLevel = config.Config.Notification.BlackDeleted.Conversation.ReliabilityLevel unReadCount = config.Config.Notification.BlackDeleted.Conversation.UnreadCount } switch reliabilityLevel { case constant.UnreliableNotification: utils.SetSwitchFromOptions(msg.Options, constant.IsHistory, false) utils.SetSwitchFromOptions(msg.Options, constant.IsPersistent, false) utils.SetSwitchFromOptions(msg.Options, constant.IsConversationUpdate, false) case constant.ReliableNotificationNoMsg: utils.SetSwitchFromOptions(msg.Options, constant.IsConversationUpdate, false) case constant.ReliableNotificationMsg: } utils.SetSwitchFromOptions(msg.Options, constant.IsUnreadCount, unReadCount) utils.SetSwitchFromOptions(msg.Options, constant.IsOfflinePush, pushSwitch) offlineInfo.Title = title offlineInfo.Desc = desc offlineInfo.Ex = ex 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()) } }