IM/open-im-server
1
0
Fork 0
mirror of https://github.com/openimsdk/open-im-server.git synced 2026-05-06 18:15:59 +08:00
Code Issues Packages Projects Releases Wiki Activity

callback

Browse Source
This commit is contained in:
wangchuxiao
2023-06-09 18:40:44 +08:00
parent 16f215a82b
commit 08f9bc705b
5 changed files with 23 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/rpc/msg/message_interceptor.go
-26
View File
@@ -5,7 +5,6 @@ import (
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/config"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/constant"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/log"
"github.com/OpenIMSDK/Open-IM-Server/pkg/errs"
"github.com/OpenIMSDK/Open-IM-Server/pkg/proto/msg"
"github.com/OpenIMSDK/Open-IM-Server/pkg/proto/sdkws"
@@ -30,28 +29,3 @@ func MessageHasReadEnabled(_ context.Context, req *msg.SendMsgReq) (*sdkws.MsgDa
}
return req.MsgData, nil
}
func MessageModifyCallback(ctx context.Context, req *msg.SendMsgReq) (*sdkws.MsgData, error) {
if err := callbackMsgModify(ctx, req); err != nil && err != errs.ErrCallbackContinue {
log.ZWarn(ctx, "CallbackMsgModify failed", err, "req", req.String())
return nil, err
}
return req.MsgData, nil
}
func MessageBeforeSendCallback(ctx context.Context, req *msg.SendMsgReq) (*sdkws.MsgData, error) {
switch req.MsgData.SessionType {
case constant.SingleChatType:
if err := callbackBeforeSendSingleMsg(ctx, req); err != nil && err != errs.ErrCallbackContinue {
log.ZWarn(ctx, "CallbackBeforeSendSingleMsg failed", err, "req", req.String())
return nil, err
}
case constant.NotificationChatType:
case constant.SuperGroupChatType:
if err := callbackBeforeSendGroupMsg(ctx, req); err != nil && err != errs.ErrCallbackContinue {
log.ZWarn(ctx, "CallbackBeforeSendGroupMsg failed", err, "req", req.String())
return nil, err
}
default:
return nil, errs.ErrArgs.Wrap("unknown sessionType")
}
return req.MsgData, nil
}
internal/rpc/msg/send.go
+15 -6
View File
@@ -18,9 +18,6 @@ func (m *msgServer) SendMsg(ctx context.Context, req *pbMsg.SendMsgReq) (resp *p
return nil, errs.ErrMessageHasReadDisable.Wrap()
}
m.encapsulateMsgData(req.MsgData)
if err := callbackMsgModify(ctx, req); err != nil && err != errs.ErrCallbackContinue {
return nil, err
}
switch req.MsgData.SessionType {
case constant.SingleChatType:
return m.sendMsgSingleChat(ctx, req)
@@ -40,12 +37,18 @@ func (m *msgServer) sendMsgSuperGroupChat(ctx context.Context, req *pbMsg.SendMs
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 err = callbackAfterSendGroupMsg(ctx, req); err != nil {
log.ZError(ctx, "CallbackAfterSendGroupMsg", err)
log.ZWarn(ctx, "CallbackAfterSendGroupMsg", err)
}
promePkg.Inc(promePkg.WorkSuperGroupChatMsgProcessSuccessCounter)
resp.SendTime = req.MsgData.SendTime
@@ -85,13 +88,19 @@ func (m *msgServer) sendMsgSingleChat(ctx context.Context, req *pbMsg.SendMsgReq
promePkg.Inc(promePkg.SingleChatMsgProcessFailedCounter)
return nil, errs.ErrUserNotRecvMsg
} 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 && err != errs.ErrCallbackContinue {
return nil, err
if err != nil {
log.ZWarn(ctx, "CallbackAfterSendSingleMsg", err, "req", req)
}
resp = &pbMsg.SendMsgResp{
ServerMsgID: req.MsgData.ServerMsgID,
internal/rpc/msg/server.go
+1 -1
View File
@@ -81,7 +81,7 @@ func Start(client discoveryregistry.SvcDiscoveryRegistry, server *grpc.Server) e
MessageLocker: NewLockerMessage(cacheModel),
}
s.notificationSender = rpcclient.NewNotificationSender(rpcclient.WithLocalSendMsg(s.SendMsg))
s.addInterceptorHandler(MessageHasReadEnabled, MessageModifyCallback)
s.addInterceptorHandler(MessageHasReadEnabled)
s.initPrometheus()
msg.RegisterMsgServer(server, s)
return nil