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

push

Browse Source
This commit is contained in:
wangchuxiao
2023-02-23 17:28:57 +08:00
parent 2359d7ab37
commit 4fc00109ef
21 changed files with 87 additions and 139 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/crontask/clear_msg.go
+6 -2
View File
@@ -12,7 +12,7 @@ import (
)
type ClearMsgTool struct {
msgInterface controller.MsgInterface
msgInterface controller.MsgDatabase
userInterface controller.UserDatabase
groupInterface controller.GroupDatabase
}
@@ -73,7 +73,11 @@ func (c *ClearMsgTool) ClearSuperGroupMsg(ctx context.Context, workingGroupIDLis
log.NewError(tracelog.GetOperationID(ctx), utils.GetSelfFuncName(), err.Error(), "GetUserMinMaxSeqInMongoAndCache failed", groupID)
continue
}
c.FixGroupUserSeq(ctx, userIDs, groupID)
//for _, userID := range userIDs {
// c.msgInterface.getgroup
// c.FixGroupUserSeq(ctx, userID, groupID, )
//
//}
c.CheckMaxSeqWithMongo(ctx, groupID, maxSeqCache, maxSeqMongo, constant.WriteDiffusion)
}
}
internal/msgtransfer/online_history_msg_handler.go
+3 -3
View File
@@ -56,8 +56,8 @@ type OnlineHistoryRedisConsumerHandler struct {
producerToModify *kafka.Producer
producerToMongo *kafka.Producer
msgInterface controller.MsgInterface
cache cache.Cache
msgDatabase controller.MsgDatabase
cache cache.Cache
}
func (och *OnlineHistoryRedisConsumerHandler) Init() {
@@ -113,7 +113,7 @@ func (och *OnlineHistoryRedisConsumerHandler) Run(channelID int) {
}
log.Debug(triggerID, "msg storage length", len(storageMsgList), "push length", len(notStoragePushMsgList))
if len(storageMsgList) > 0 {
lastSeq, err := och.msgInterface.BatchInsertChat2Cache(ctx, msgChannelValue.aggregationID, storageMsgList)
lastSeq, err := och.msgDatabase.BatchInsertChat2Cache(ctx, msgChannelValue.aggregationID, storageMsgList)
if err != nil {
och.singleMsgFailedCountMutex.Lock()
och.singleMsgFailedCount += uint64(len(storageMsgList))
internal/msgtransfer/online_msg_to_mongo_handler.go
+5 -5
View File
@@ -18,7 +18,7 @@ import (
type OnlineHistoryMongoConsumerHandler struct {
historyConsumerGroup *kfk.MConsumerGroup
msgInterface controller.MsgInterface
msgDatabase controller.MsgDatabase
cache cache.Cache
}
@@ -39,12 +39,12 @@ func (mc *OnlineHistoryMongoConsumerHandler) handleChatWs2Mongo(cMsg *sarama.Con
ctx := context.Background()
tracelog.SetOperationID(ctx, msgFromMQ.TriggerID)
//err = db.DB.BatchInsertChat2DB(msgFromMQ.AggregationID, msgFromMQ.MessageList, msgFromMQ.TriggerID, msgFromMQ.LastSeq)
err = mc.msgInterface.BatchInsertChat2DB(ctx, msgFromMQ.AggregationID, msgFromMQ.MessageList, msgFromMQ.LastSeq)
err = mc.msgDatabase.BatchInsertChat2DB(ctx, msgFromMQ.AggregationID, msgFromMQ.MessageList, msgFromMQ.LastSeq)
if err != nil {
log.NewError(msgFromMQ.TriggerID, "single data insert to mongo err", err.Error(), msgFromMQ.MessageList, msgFromMQ.AggregationID, msgFromMQ.TriggerID)
}
//err = db.DB.DeleteMessageFromCache(msgFromMQ.MessageList, msgFromMQ.AggregationID, msgFromMQ.GetTriggerID())
err = mc.msgInterface.DeleteMessageFromCache(ctx, msgFromMQ.AggregationID, msgFromMQ.MessageList)
err = mc.msgDatabase.DeleteMessageFromCache(ctx, msgFromMQ.AggregationID, msgFromMQ.MessageList)
if err != nil {
log.NewError(msgFromMQ.TriggerID, "remove cache msg from redis err", err.Error(), msgFromMQ.MessageList, msgFromMQ.AggregationID, msgFromMQ.TriggerID)
}
@@ -62,8 +62,8 @@ func (mc *OnlineHistoryMongoConsumerHandler) handleChatWs2Mongo(cMsg *sarama.Con
log.NewError(msgFromMQ.TriggerID, "deleteMessageTips unmarshal err:", err.Error(), v.String())
continue
}
if totalUnExistSeqs, err := mc.msgInterface.DelMsgBySeqs(ctx, DeleteMessageTips.UserID, DeleteMessageTips.SeqList); err != nil {
log.NewError(v.OperationID, utils.GetSelfFuncName(), "DelMsgBySeqs args: ", DeleteMessageTips.UserID, DeleteMessageTips.SeqList, "error:", err.Error(), "totalUnExistSeqs: ", totalUnExistSeqs)
if totalUnExistSeqs, err := mc.msgDatabase.DelMsgBySeqs(ctx, DeleteMessageTips.UserID, DeleteMessageTips.Seqs); err != nil {
log.NewError(v.OperationID, utils.GetSelfFuncName(), "DelMsgBySeqs args: ", DeleteMessageTips.UserID, DeleteMessageTips.Seqs, "error:", err.Error(), "totalUnExistSeqs: ", totalUnExistSeqs)
}
}
internal/push/getui/push.go
+5
View File
@@ -9,6 +9,7 @@ import (
"Open_IM/pkg/common/tracelog"
"Open_IM/pkg/utils/splitter"
"github.com/go-redis/redis/v8"
"sync"
"Open_IM/pkg/utils"
"context"
@@ -64,13 +65,17 @@ func (g *Client) Push(ctx context.Context, userIDs []string, title, content stri
maxNum := 999
if len(userIDs) > maxNum {
s := splitter.NewSplitter(maxNum, userIDs)
wg := sync.WaitGroup{}
wg.Add(len(s.GetSplitResult()))
for i, v := range s.GetSplitResult() {
go func(index int, userIDs []string) {
defer wg.Done()
if err = g.batchPush(ctx, token, userIDs, pushReq); err != nil {
log.NewError(tracelog.GetOperationID(ctx), "batchPush failed", i, token, pushReq)
}
}(i, v.Item)
}
wg.Wait()
} else {
err = g.batchPush(ctx, token, userIDs, pushReq)
}
internal/push/init.go
+7 -3
View File
@@ -25,9 +25,12 @@ type Push struct {
successCount uint64
}
func (p *Push) Init(rpcPort int) {
var cacheInterface cache.Cache
func (p *Push) Init(rpcPort int) error {
redisClient, err := cache.NewRedis()
if err != nil {
return err
}
var cacheInterface cache.Cache = redisClient
p.rpcServer.Init(rpcPort, cacheInterface)
p.pushCh.Init()
statistics.NewStatistics(&p.successCount, config.Config.ModuleName.PushName, fmt.Sprintf("%d second push to msg_gateway count", constant.StatisticsTimeInterval), constant.StatisticsTimeInterval)
@@ -40,6 +43,7 @@ func (p *Push) Init(rpcPort int) {
if config.Config.Push.Fcm.Enable {
p.offlinePusher = fcm.NewClient(cacheInterface)
}
return nil
}
func (p *Push) initPrometheus() {
internal/push/push_handler.go
+12 -2
View File
@@ -11,15 +11,18 @@ import (
"Open_IM/pkg/common/constant"
kfk "Open_IM/pkg/common/kafka"
"Open_IM/pkg/common/log"
"Open_IM/pkg/common/tracelog"
pbChat "Open_IM/pkg/proto/msg"
pbPush "Open_IM/pkg/proto/push"
"Open_IM/pkg/utils"
"context"
"github.com/Shopify/sarama"
"github.com/golang/protobuf/proto"
)
type ConsumerHandler struct {
pushConsumerGroup *kfk.MConsumerGroup
pusher Pusher
}
func (c *ConsumerHandler) Init() {
@@ -27,6 +30,7 @@ func (c *ConsumerHandler) Init() {
OffsetsInitial: sarama.OffsetNewest, IsReturnErr: false}, []string{config.Config.Kafka.Ms2pschat.Topic}, config.Config.Kafka.Ms2pschat.Addr,
config.Config.Kafka.ConsumerGroupID.MsgToPush)
}
func (c *ConsumerHandler) handleMs2PsChat(msg []byte) {
log.NewDebug("", "msg come from kafka And push!!!", "msg", string(msg))
msgFromMQ := pbChat.PushMsgDataToMQ{}
@@ -43,11 +47,17 @@ func (c *ConsumerHandler) handleMs2PsChat(msg []byte) {
if nowSec-sec > 10 {
return
}
ctx := context.Background()
tracelog.SetOperationID(ctx, "")
var err error
switch msgFromMQ.MsgData.SessionType {
case constant.SuperGroupChatType:
MsgToSuperGroupUser(pbData)
err = c.pusher.MsgToSuperGroupUser(ctx, pbData.SourceID, pbData.MsgData)
default:
MsgToUser(pbData)
err = c.pusher.MsgToUser(ctx, pbData.SourceID, pbData.MsgData)
}
if err != nil {
log.NewError("", "push failed", *pbData)
}
}
func (ConsumerHandler) Setup(_ sarama.ConsumerGroupSession) error { return nil }
internal/push/push_rpc_server.go
+6 -9
View File
@@ -21,16 +21,13 @@ import (
type RPCServer struct {
rpcPort int
rpcRegisterName string
etcdSchema string
etcdAddr []string
push controller.PushInterface
pushInterface controller.PushInterface
pusher Pusher
}
func (r *RPCServer) Init(rpcPort int, cache cache.Cache) {
r.rpcPort = rpcPort
r.rpcRegisterName = config.Config.RpcRegisterName.OpenImPushName
r.etcdSchema = config.Config.Etcd.EtcdSchema
r.etcdAddr = config.Config.Etcd.EtcdAddr
}
func (r *RPCServer) run() {
@@ -84,13 +81,13 @@ func (r *RPCServer) run() {
func (r *RPCServer) PushMsg(ctx context.Context, pbData *pbPush.PushMsgReq) (resp *pbPush.PushMsgResp, err error) {
switch pbData.MsgData.SessionType {
case constant.SuperGroupChatType:
MsgToSuperGroupUser(pbData)
err = r.pusher.MsgToSuperGroupUser(ctx, pbData.SourceID, pbData.MsgData)
default:
MsgToUser(pbData)
err = r.pusher.MsgToUser(ctx, pbData.SourceID, pbData.MsgData)
}
return &pbPush.PushMsgResp{}, nil
return &pbPush.PushMsgResp{}, err
}
func (r *RPCServer) DelUserPushToken(ctx context.Context, req *pbPush.DelUserPushTokenReq) (resp *pbPush.DelUserPushTokenResp, err error) {
return &pbPush.DelUserPushTokenResp{}, r.push.DelFcmToken(ctx, req.UserID, int(req.PlatformID))
return &pbPush.DelUserPushTokenResp{}, r.pushInterface.DelFcmToken(ctx, req.UserID, int(req.PlatformID))
}
internal/push/push_to_client.go
-1
View File
@@ -79,7 +79,6 @@ func (p *Pusher) MsgToUser(ctx context.Context, userID string, msg *sdkws.MsgDat
}
err = p.OfflinePushMsg(ctx, userID, msg, userIDs)
if err != nil {
log.NewError(operationID, "OfflinePushMsg failed", userID)
return err
}
}
internal/push/tpns.go
-31
View File
@@ -1,31 +0,0 @@
package push
import (
"Open_IM/internal/push/sdk/tpns-server-sdk-go/go/auth"
"Open_IM/pkg/common/config"
)
var badgeType = -2
var iosAcceptId = auth.Auther{AccessID: config.Config.Push.Tpns.Ios.AccessID, SecretKey: config.Config.Push.Tpns.Ios.SecretKey}
func IOSAccountListPush(accounts []string, title, content, jsonCustomContent string) {
var iosMessage = tpns.Message{
Title: title,
Content: content,
IOS: &tpns.IOSParams{
Aps: &tpns.Aps{
BadgeType: &badgeType,
Sound: "default",
Category: "INVITE_CATEGORY",
},
CustomContent: jsonCustomContent,
//CustomContent: `"{"key\":\"value\"}"`,
},
}
pushReq, reqBody, err := req.NewListAccountPush(accounts, iosMessage)
if err != nil {
return
}
iosAcceptId.Auth(pushReq, auth.UseSignAuthored, iosAcceptId, reqBody)
common.PushAndGetResult(pushReq)
}
internal/rpc/friend/friend.go
+1 -1
View File
@@ -51,7 +51,7 @@ func (s *friendServer) ApplyToAddFriend(ctx context.Context, req *pbfriend.Apply
if err := tokenverify.CheckAccessV3(ctx, req.FromUserID); err != nil {
return nil, err
}
if err := CallbackBeforeAddFriend(ctx, req); err != nil {
if err := CallbackBeforeAddFriend(ctx, req); err != nil && err != constant.ErrCallbackContinue {
return nil, err
}
if req.ToUserID == req.FromUserID {
internal/rpc/group/group.go
+5 -5
View File
@@ -142,7 +142,7 @@ func (s *groupServer) CreateGroup(ctx context.Context, req *pbGroup.CreateGroupR
if err != nil {
return nil, err
}
if err := CallbackBeforeCreateGroup(ctx, req); err != nil {
if err := CallbackBeforeCreateGroup(ctx, req); err != nil && err != constant.ErrCallbackContinue {
return nil, err
}
var groupMembers []*relationTb.GroupMemberModel
@@ -158,7 +158,7 @@ func (s *groupServer) CreateGroup(ctx context.Context, req *pbGroup.CreateGroupR
groupMember.OperatorUserID = tracelog.GetOpUserID(ctx)
groupMember.JoinSource = constant.JoinByInvitation
groupMember.InviterUserID = tracelog.GetOpUserID(ctx)
if err := CallbackBeforeMemberJoinGroup(ctx, groupMember, group.Ex); err != nil {
if err := CallbackBeforeMemberJoinGroup(ctx, groupMember, group.Ex); err != nil && err != constant.ErrCallbackContinue {
return err
}
groupMembers = append(groupMembers, groupMember)
@@ -318,7 +318,7 @@ func (s *groupServer) InviteUserToGroup(ctx context.Context, req *pbGroup.Invite
member.OperatorUserID = opUserID
member.InviterUserID = opUserID
member.JoinSource = constant.JoinByInvitation
if err := CallbackBeforeMemberJoinGroup(ctx, member, group.Ex); err != nil {
if err := CallbackBeforeMemberJoinGroup(ctx, member, group.Ex); err != nil && err != constant.ErrCallbackContinue {
return nil, err
}
groupMembers = append(groupMembers, member)
@@ -601,7 +601,7 @@ func (s *groupServer) GroupApplicationResponse(ctx context.Context, req *pbGroup
OperatorUserID: tracelog.GetOpUserID(ctx),
Ex: groupRequest.Ex,
}
if err = CallbackBeforeMemberJoinGroup(ctx, member, group.Ex); err != nil {
if err = CallbackBeforeMemberJoinGroup(ctx, member, group.Ex); err != nil && err != constant.ErrCallbackContinue {
return nil, err
}
}
@@ -645,7 +645,7 @@ func (s *groupServer) JoinGroup(ctx context.Context, req *pbGroup.JoinGroupReq)
groupMember.OperatorUserID = tracelog.GetOpUserID(ctx)
groupMember.JoinSource = constant.JoinByInvitation
groupMember.InviterUserID = tracelog.GetOpUserID(ctx)
if err := CallbackBeforeMemberJoinGroup(ctx, groupMember, group.Ex); err != nil {
if err := CallbackBeforeMemberJoinGroup(ctx, groupMember, group.Ex); err != nil && err != constant.ErrCallbackContinue {
return nil, err
}
if err := s.GroupDatabase.CreateGroup(ctx, nil, []*relationTb.GroupMemberModel{groupMember}); err != nil {
internal/rpc/msg/send_pull.go
+5 -5
View File
@@ -16,7 +16,7 @@ func (m *msgServer) sendMsgSuperGroupChat(ctx context.Context, req *msg.SendMsgR
resp = &msg.SendMsgResp{}
promePkg.PromeInc(promePkg.WorkSuperGroupChatMsgRecvSuccessCounter)
// callback
if err = CallbackBeforeSendGroupMsg(ctx, req); err != nil {
if err = CallbackBeforeSendGroupMsg(ctx, req); err != nil && err != constant.ErrCallbackContinue {
return nil, err
}
@@ -61,7 +61,7 @@ func (m *msgServer) sendMsgNotification(ctx context.Context, req *msg.SendMsgReq
func (m *msgServer) sendMsgSingleChat(ctx context.Context, req *msg.SendMsgReq) (resp *msg.SendMsgResp, err error) {
promePkg.PromeInc(promePkg.SingleChatMsgRecvSuccessCounter)
if err = CallbackBeforeSendSingleMsg(ctx, req); err != nil {
if err = CallbackBeforeSendSingleMsg(ctx, req); err != nil && err != constant.ErrCallbackContinue {
return nil, err
}
_, err = m.messageVerification(ctx, req)
@@ -86,7 +86,7 @@ func (m *msgServer) sendMsgSingleChat(ctx context.Context, req *msg.SendMsgReq)
}
}
err = CallbackAfterSendSingleMsg(ctx, req)
if err != nil {
if err != nil && err != constant.ErrCallbackContinue {
return nil, err
}
promePkg.PromeInc(promePkg.SingleChatMsgProcessSuccessCounter)
@@ -100,7 +100,7 @@ func (m *msgServer) sendMsgGroupChat(ctx context.Context, req *msg.SendMsgReq) (
// callback
promePkg.PromeInc(promePkg.GroupChatMsgRecvSuccessCounter)
err = CallbackBeforeSendGroupMsg(ctx, req)
if err != nil {
if err != nil && err != constant.ErrCallbackContinue {
return nil, err
}
@@ -235,7 +235,7 @@ func (m *msgServer) SendMsg(ctx context.Context, req *msg.SendMsgReq) (resp *msg
return nil, constant.ErrMessageHasReadDisable.Wrap()
}
m.encapsulateMsgData(req.MsgData)
if err := CallbackMsgModify(ctx, req); err != nil {
if err := CallbackMsgModify(ctx, req); err != nil && err != constant.ErrCallbackContinue {
return nil, err
}
switch req.MsgData.SessionType {