You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
Open-IM-Server/internal/rpc/msg/send.go

199 lines
6.8 KiB

// Copyright © 2023 OpenIM. All rights reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package msg
import (
"context"
"github.com/OpenIMSDK/Open-IM-Server/pkg/msgprocessor"
"github.com/OpenIMSDK/protocol/constant"
pbconversation "github.com/OpenIMSDK/protocol/conversation"
pbmsg "github.com/OpenIMSDK/protocol/msg"
"github.com/OpenIMSDK/protocol/sdkws"
"github.com/OpenIMSDK/protocol/wrapperspb"
"github.com/OpenIMSDK/tools/errs"
"github.com/OpenIMSDK/tools/log"
"github.com/OpenIMSDK/tools/mcontext"
"github.com/OpenIMSDK/tools/utils"
promepkg "github.com/OpenIMSDK/Open-IM-Server/pkg/common/prome"
)
func (m *msgServer) SendMsg(ctx context.Context, req *pbmsg.SendMsgReq) (resp *pbmsg.SendMsgResp, error error) {
resp = &pbmsg.SendMsgResp{}
if req.MsgData != nil {
flag := isMessageHasReadEnabled(req.MsgData)
if !flag {
return nil, errs.ErrMessageHasReadDisable.Wrap()
}
m.encapsulateMsgData(req.MsgData)
switch req.MsgData.SessionType {
case constant.SingleChatType:
return m.sendMsgSingleChat(ctx, req)
case constant.NotificationChatType:
return m.sendMsgNotification(ctx, req)
case constant.SuperGroupChatType:
return m.sendMsgSuperGroupChat(ctx, req)
default:
return nil, errs.ErrArgs.Wrap("unknown sessionType")
}
} else {
return nil, errs.ErrArgs.Wrap("msgData is nil")
}
}
func (m *msgServer) sendMsgSuperGroupChat(
ctx context.Context,
req *pbmsg.SendMsgReq,
) (resp *pbmsg.SendMsgResp, err error) {
promepkg.Inc(promepkg.WorkSuperGroupChatMsgRecvSuccessCounter)
if err = m.messageVerification(ctx, req); err != nil {
promepkg.Inc(promepkg.WorkSuperGroupChatMsgProcessFailedCounter)
return nil, err
}
if err = callbackBeforeSendGroupMsg(ctx, req); err != nil {
return nil, err
}
if err := callbackMsgModify(ctx, req); err != nil {
return nil, err
}
err = m.MsgDatabase.MsgToMQ(ctx, utils.GenConversationUniqueKeyForGroup(req.MsgData.GroupID), req.MsgData)
if err != nil {
return nil, err
}
if req.MsgData.ContentType == constant.AtText {
go m.setConversationAtInfo(ctx, req.MsgData)
}
if err = callbackAfterSendGroupMsg(ctx, req); err != nil {
log.ZWarn(ctx, "CallbackAfterSendGroupMsg", err)
}
promepkg.Inc(promepkg.WorkSuperGroupChatMsgProcessSuccessCounter)
resp = &pbmsg.SendMsgResp{}
resp.SendTime = req.MsgData.SendTime
resp.ServerMsgID = req.MsgData.ServerMsgID
resp.ClientMsgID = req.MsgData.ClientMsgID
return resp, nil
}
func (m *msgServer) setConversationAtInfo(nctx context.Context, msg *sdkws.MsgData) {
log.ZDebug(nctx, "setConversationAtInfo", "msg", msg)
ctx := mcontext.NewCtx("@@@" + mcontext.GetOperationID(nctx))
var atUserID []string
conversation := &pbconversation.ConversationReq{
ConversationID: msgprocessor.GetConversationIDByMsg(msg),
ConversationType: msg.SessionType,
GroupID: msg.GroupID,
}
tagAll := utils.IsContain(constant.AtAllString, msg.AtUserIDList)
if tagAll {
memberUserIDList, err := m.Group.GetGroupMemberIDs(ctx, msg.GroupID)
if err != nil {
log.ZWarn(ctx, "GetGroupMemberIDs", err)
return
}
atUserID = utils.DifferenceString([]string{constant.AtAllString}, msg.AtUserIDList)
if len(atUserID) == 0 { // just @everyone
conversation.GroupAtType = &wrapperspb.Int32Value{Value: constant.AtAll}
} else { //@Everyone and @other people
conversation.GroupAtType = &wrapperspb.Int32Value{Value: constant.AtAllAtMe}
err := m.Conversation.SetConversations(ctx, atUserID, conversation)
if err != nil {
log.ZWarn(ctx, "SetConversations", err, "userID", atUserID, "conversation", conversation)
}
memberUserIDList = utils.DifferenceString(atUserID, memberUserIDList)
}
conversation.GroupAtType = &wrapperspb.Int32Value{Value: constant.AtAll}
err = m.Conversation.SetConversations(ctx, memberUserIDList, conversation)
if err != nil {
log.ZWarn(ctx, "SetConversations", err, "userID", memberUserIDList, "conversation", conversation)
}
} else {
conversation.GroupAtType = &wrapperspb.Int32Value{Value: constant.AtMe}
err := m.Conversation.SetConversations(ctx, msg.AtUserIDList, conversation)
if err != nil {
log.ZWarn(ctx, "SetConversations", err, msg.AtUserIDList, conversation)
}
}
}
func (m *msgServer) sendMsgNotification(
ctx context.Context,
req *pbmsg.SendMsgReq,
) (resp *pbmsg.SendMsgResp, err error) {
promepkg.Inc(promepkg.SingleChatMsgRecvSuccessCounter)
if err := m.MsgDatabase.MsgToMQ(ctx, utils.GenConversationUniqueKeyForSingle(req.MsgData.SendID, req.MsgData.RecvID), req.MsgData); err != nil {
promepkg.Inc(promepkg.SingleChatMsgProcessFailedCounter)
return nil, err
}
resp = &pbmsg.SendMsgResp{
ServerMsgID: req.MsgData.ServerMsgID,
ClientMsgID: req.MsgData.ClientMsgID,
SendTime: req.MsgData.SendTime,
}
return resp, nil
}
func (m *msgServer) sendMsgSingleChat(ctx context.Context, req *pbmsg.SendMsgReq) (resp *pbmsg.SendMsgResp, err error) {
promepkg.Inc(promepkg.SingleChatMsgRecvSuccessCounter)
if err := m.messageVerification(ctx, req); err != nil {
return nil, err
}
isSend := true
isNotification := msgprocessor.IsNotificationByMsg(req.MsgData)
if !isNotification {
isSend, err = m.modifyMessageByUserMessageReceiveOpt(
ctx,
req.MsgData.RecvID,
utils.GenConversationIDForSingle(req.MsgData.SendID, req.MsgData.RecvID),
constant.SingleChatType,
req,
)
if err != nil {
return nil, err
}
}
if !isSend {
promepkg.Inc(promepkg.SingleChatMsgProcessFailedCounter)
return nil, nil
} else {
if err = callbackBeforeSendSingleMsg(ctx, req); err != nil {
return nil, err
}
if err := callbackMsgModify(ctx, req); err != nil {
return nil, err
}
if err := m.MsgDatabase.MsgToMQ(ctx, utils.GenConversationUniqueKeyForSingle(req.MsgData.SendID, req.MsgData.RecvID), req.MsgData); err != nil {
promepkg.Inc(promepkg.SingleChatMsgProcessFailedCounter)
return nil, err
}
err = callbackAfterSendSingleMsg(ctx, req)
if err != nil {
log.ZWarn(ctx, "CallbackAfterSendSingleMsg", err, "req", req)
}
resp = &pbmsg.SendMsgResp{
ServerMsgID: req.MsgData.ServerMsgID,
ClientMsgID: req.MsgData.ClientMsgID,
SendTime: req.MsgData.SendTime,
}
promepkg.Inc(promepkg.SingleChatMsgProcessSuccessCounter)
return resp, nil
}
}
func (m *msgServer) BatchSendMsg(ctx context.Context, in *pbmsg.BatchSendMessageReq) (*pbmsg.BatchSendMessageResp, error) {
return nil, nil
}