blackteay/open-im-server
1
0
Fork 0
mirror of https://github.com/openimsdk/open-im-server.git synced 2025-04-06 04:15:46 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix delete

Browse Source
This commit is contained in:
wangchuxiao 2022-08-10 15:14:51 +08:00
parent 2ee0943115
commit 2892add766
2 changed files with 32 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

48
internal/cron_task/clear_msg.go
View File

@ -48,12 +48,29 @@ func DeleteMongoMsgAndResetRedisSeq(operationID, userID string) error {
return err return err
} }
func delMongoMsgs(operationID string, delMsgIDList *[][2]interface{}) error {
if len(*delMsgIDList) > 0 {
var IDList []string
for _, v := range *delMsgIDList {
IDList = append(IDList, v[0].(string))
}
err := db.DB.DelMongoMsgs(IDList)
if err != nil {
return utils.Wrap(err, "DelMongoMsgs failed")
}
}
return nil
}
// recursion // recursion
func deleteMongoMsg(operationID string, ID string, index int64, delMsgIDList *[][2]interface{}) (uint32, error) { func deleteMongoMsg(operationID string, ID string, index int64, delMsgIDList *[][2]interface{}) (uint32, error) {
// 从最旧的列表开始找 // 从最旧的列表开始找
msgs, err := db.DB.GetUserMsgListByIndex(ID, index) msgs, err := db.DB.GetUserMsgListByIndex(ID, index)
if err != nil { if err != nil || msgs.UID == "" {
return 0, utils.Wrap(err, "GetUserMsgListByIndex failed") if err != nil {
log.NewError(operationID, utils.GetSelfFuncName(), "GetUserMsgListByIndex failed", err.Error(), index, ID)
}
return getDelMaxSeqByIDList(*delMsgIDList), delMongoMsgs(operationID, delMsgIDList)
} }
if len(msgs.Msg) > db.GetSingleGocMsgNum() { if len(msgs.Msg) > db.GetSingleGocMsgNum() {
log.NewWarn(operationID, utils.GetSelfFuncName(), "msgs too large", len(msgs.Msg), msgs.UID) log.NewWarn(operationID, utils.GetSelfFuncName(), "msgs too large", len(msgs.Msg), msgs.UID)
@ -62,15 +79,8 @@ func deleteMongoMsg(operationID string, ID string, index int64, delMsgIDList *[]
for i, msg := range msgs.Msg { for i, msg := range msgs.Msg {
// 找到列表中不需要删除的消息了 // 找到列表中不需要删除的消息了
if utils.GetCurrentTimestampByMill() < msg.SendTime+int64(config.Config.Mongo.DBRetainChatRecords)*24*60*60*1000 { if utils.GetCurrentTimestampByMill() < msg.SendTime+int64(config.Config.Mongo.DBRetainChatRecords)*24*60*60*1000 {
if len(*delMsgIDList) > 0 { if err := delMongoMsgs(operationID, delMsgIDList); err != nil {
var IDList []string return 0, err
for _, v := range *delMsgIDList {
IDList = append(IDList, v[0].(string))
}
err := db.DB.DelMongoMsgs(IDList)
if err != nil {
return 0, utils.Wrap(err, "DelMongoMsgs failed")
}
} }
minSeq := getDelMaxSeqByIDList(*delMsgIDList) minSeq := getDelMaxSeqByIDList(*delMsgIDList)
if i > 0 { if i > 0 {
@ -84,19 +94,21 @@ func deleteMongoMsg(operationID string, ID string, index int64, delMsgIDList *[]
log.NewError(operationID, utils.GetSelfFuncName(), err.Error(), msgs.UID, i) log.NewError(operationID, utils.GetSelfFuncName(), err.Error(), msgs.UID, i)
return minSeq, nil return minSeq, nil
} }
minSeq = msgPb.Seq - 1 minSeq = msgPb.Seq
} }
} }
return minSeq, nil return minSeq, nil
} }
} }
msgPb := &server_api_params.MsgData{} if len(msgs.Msg) > 0 {
err = proto.Unmarshal(msgs.Msg[len(msgs.Msg)-1].Msg, msgPb) msgPb := &server_api_params.MsgData{}
if err != nil { err = proto.Unmarshal(msgs.Msg[len(msgs.Msg)-1].Msg, msgPb)
log.NewError(operationID, utils.GetSelfFuncName(), err.Error(), len(msgs.Msg)-1, msgs.UID) if err != nil {
return 0, utils.Wrap(err, "proto.Unmarshal failed") log.NewError(operationID, utils.GetSelfFuncName(), err.Error(), len(msgs.Msg)-1, msgs.UID)
return 0, utils.Wrap(err, "proto.Unmarshal failed")
}
*delMsgIDList = append(*delMsgIDList, [2]interface{}{msgs.UID, msgPb.Seq})
} }
*delMsgIDList = append(*delMsgIDList, [2]interface{}{msgs.UID, msgPb.Seq})
// 没有找到 代表需要全部删除掉 继续递归查找下一个比较旧的列表 // 没有找到 代表需要全部删除掉 继续递归查找下一个比较旧的列表
seq, err := deleteMongoMsg(operationID, utils.GetSelfFuncName(), index+1, delMsgIDList) seq, err := deleteMongoMsg(operationID, utils.GetSelfFuncName(), index+1, delMsgIDList)
if err != nil { if err != nil {

3
internal/cron_task/cron_task.go
View File

@ -1,6 +1,7 @@
package cronTask package cronTask
import ( import (
"Open_IM/pkg/common/config"
"Open_IM/pkg/common/constant" "Open_IM/pkg/common/constant"
"Open_IM/pkg/common/db/mysql_model/im_mysql_model" "Open_IM/pkg/common/db/mysql_model/im_mysql_model"
rocksCache "Open_IM/pkg/common/db/rocks_cache" rocksCache "Open_IM/pkg/common/db/rocks_cache"
@ -16,7 +17,7 @@ const cronTaskOperationID = "cronTaskOperationID-"
func StartCronTask() { func StartCronTask() {
log.NewInfo(utils.OperationIDGenerator(), "start cron task") log.NewInfo(utils.OperationIDGenerator(), "start cron task")
c := cron.New() c := cron.New()
_, err := c.AddFunc("30 3-6,20-23 * * *", func() { _, err := c.AddFunc(config.Config.Mongo.ChatRecordsClearTime, func() {
operationID := getCronTaskOperationID() operationID := getCronTaskOperationID()
userIDList, err := im_mysql_model.SelectAllUserID() userIDList, err := im_mysql_model.SelectAllUserID()
if err == nil { if err == nil {