msb-public
/
Open-IM-Server
mirror of https://github.com/OpenIMSDK/Open-IM-Server
2
0
Code Issues Projects Releases Wiki Activity

test cron

Browse Source
pull/351/head
wangchuxiao 3 years ago
parent 2cbecf439e
commit 1b3fbf56d0
2 changed files with 46 additions and 38 deletions
Show Stats Download Patch File Download Diff File

73
internal/cron_task/clear_msg.go
Unescape View File

@ -18,15 +18,15 @@ const oldestList = 0
const newestList = -1 const newestList = -1
func ResetUserGroupMinSeq(operationID, groupID string, userIDList []string) error { func ResetUserGroupMinSeq(operationID, groupID string, userIDList []string) error {
var delMsgIDList [][2]interface{} var delStruct delMsgRecursionStruct
minSeq, err := deleteMongoMsg(operationID, groupID, oldestList, &delMsgIDList) minSeq, err := deleteMongoMsg(operationID, groupID, oldestList, &delStruct)
if err != nil { if err != nil {
log.NewError(operationID, utils.GetSelfFuncName(), groupID, "deleteMongoMsg failed") log.NewError(operationID, utils.GetSelfFuncName(), groupID, "deleteMongoMsg failed")
} }
if minSeq == 0 { if minSeq == 0 {
return nil return nil
} }
log.NewDebug(operationID, utils.GetSelfFuncName(), "delMsgIDList:", delMsgIDList, "minSeq", minSeq) log.NewDebug(operationID, utils.GetSelfFuncName(), "delMsgIDList:", delStruct, "minSeq", minSeq)
for _, userID := range userIDList { for _, userID := range userIDList {
userMinSeq, err := db.DB.GetGroupUserMinSeq(groupID, userID) userMinSeq, err := db.DB.GetGroupUserMinSeq(groupID, userID)
if err != nil && err != goRedis.Nil { if err != nil && err != goRedis.Nil {
@ -46,27 +46,23 @@ func ResetUserGroupMinSeq(operationID, groupID string, userIDList []string) erro
} }
func DeleteMongoMsgAndResetRedisSeq(operationID, userID string) error { func DeleteMongoMsgAndResetRedisSeq(operationID, userID string) error {
var delMsgIDList [][2]interface{} var delStruct delMsgRecursionStruct
minSeq, err := deleteMongoMsg(operationID, userID, oldestList, &delMsgIDList) minSeq, err := deleteMongoMsg(operationID, userID, oldestList, &delStruct)
if err != nil { if err != nil {
return utils.Wrap(err, "") return utils.Wrap(err, "")
} }
if minSeq == 0 { if minSeq == 0 {
return nil return nil
} }
log.NewDebug(operationID, utils.GetSelfFuncName(), "delMsgIDMap: ", delMsgIDList, "minSeq", minSeq) log.NewDebug(operationID, utils.GetSelfFuncName(), "delMsgIDMap: ", delStruct, "minSeq", minSeq)
err = db.DB.SetUserMinSeq(userID, minSeq) err = db.DB.SetUserMinSeq(userID, minSeq)
return utils.Wrap(err, "") return utils.Wrap(err, "")
} }
// del list // del list
func delMongoMsgsPhysical(delMsgIDList *[][2]interface{}) error { func delMongoMsgsPhysical(uidList []string) error {
if len(*delMsgIDList) > 0 { if len(uidList) > 0 {
var IDList []string err := db.DB.DelMongoMsgs(uidList)
for _, v := range *delMsgIDList {
IDList = append(IDList, v[0].(string))
}
err := db.DB.DelMongoMsgs(IDList)
if err != nil { if err != nil {
return utils.Wrap(err, "DelMongoMsgs failed") return utils.Wrap(err, "DelMongoMsgs failed")
} }
@ -74,28 +70,37 @@ func delMongoMsgsPhysical(delMsgIDList *[][2]interface{}) error {
return nil return nil
} }
type delMsgRecursionStruct struct {
minSeq uint32
delUidList []string
}
func (d *delMsgRecursionStruct) getSetMinSeq() uint32 {
return d.minSeq
}
// index 0....19(del) 20...69 // index 0....19(del) 20...69
// seq 70 // seq 70
// set minSeq 21 // set minSeq 21
// recursion // recursion
func deleteMongoMsg(operationID string, ID string, index int64, delMsgIDList *[][2]interface{}) (uint32, error) { func deleteMongoMsg(operationID string, ID string, index int64, delStruct *delMsgRecursionStruct) (uint32, error) {
// delMsgIDList [[uid:0, minSeq], [uid:1, minSeq]] // delMsgIDList [[uid:0, minSeq], [uid:1, minSeq]]
// find from oldest list // find from oldest list
msgs, err := db.DB.GetUserMsgListByIndex(ID, index) msgs, err := db.DB.GetUserMsgListByIndex(ID, index)
if err != nil || msgs.UID == "" { if err != nil || msgs.UID == "" {
if err != nil { if err != nil {
if err == db.ErrMsgListNotExist { if err == db.ErrMsgListNotExist {
log.NewDebug(operationID, utils.GetSelfFuncName(), "ID:", ID, "index:", index, err.Error()) log.NewInfo(operationID, utils.GetSelfFuncName(), "ID:", ID, "index:", index, err.Error())
} else { } else {
log.NewError(operationID, utils.GetSelfFuncName(), "GetUserMsgListByIndex failed", err.Error(), index, ID) log.NewError(operationID, utils.GetSelfFuncName(), "GetUserMsgListByIndex failed", err.Error(), index, ID)
} }
} }
// 获取报错，或者获取不到了，物理删除并且返回seq // 获取报错，或者获取不到了，物理删除并且返回seq
err = delMongoMsgsPhysical(delMsgIDList) err = delMongoMsgsPhysical(delStruct.delUidList)
if err != nil { if err != nil {
return 0, err return 0, err
} }
return getDelMaxSeqByIDList(*delMsgIDList) + 1, nil return delStruct.getSetMinSeq(), nil
} }
log.NewDebug(operationID, "ID:", ID, "index:", index, "uid:", msgs.UID, "len:", len(msgs.Msg)) log.NewDebug(operationID, "ID:", ID, "index:", index, "uid:", msgs.UID, "len:", len(msgs.Msg))
if len(msgs.Msg) > db.GetSingleGocMsgNum() { if len(msgs.Msg) > db.GetSingleGocMsgNum() {
@ -106,20 +111,20 @@ func deleteMongoMsg(operationID string, ID string, index int64, delMsgIDList *[]
// 找到列表中不需要删除的消息了, 表示为递归到最后一个块 // 找到列表中不需要删除的消息了, 表示为递归到最后一个块
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) {
// 删除块失败 递归结束 返回0 // 删除块失败 递归结束 返回0
if err := delMongoMsgsPhysical(delMsgIDList); err != nil { if err := delMongoMsgsPhysical(delStruct.delUidList); err != nil {
return 0, err return 0, err
} }
// unMarshall失败 块删除成功 返回块的最大seq+1 设置为最小seq // unMarshall失败 块删除成功 设置为最小seq
msgPb := &server_api_params.MsgData{} msgPb := &server_api_params.MsgData{}
if err = proto.Unmarshal(msg.Msg, msgPb); err != nil { if err = proto.Unmarshal(msg.Msg, msgPb); err != nil {
return getDelMaxSeqByIDList(*delMsgIDList) + 1, utils.Wrap(err, "") return delStruct.getSetMinSeq(), utils.Wrap(err, "")
} }
// 如果不是块中第一个，就把前面比他早插入的全部设置空。 // 如果不是块中第一个，就把前面比他早插入的全部设置空 seq字段除外。
if i > 0 { if i > 0 {
err = db.DB.ReplaceMsgToBlankByIndex(msgs.UID, i-1) err = db.DB.ReplaceMsgToBlankByIndex(msgs.UID, i-1)
if err != nil { if err != nil {
log.NewError(operationID, utils.GetSelfFuncName(), err.Error(), msgs.UID, i) log.NewError(operationID, utils.GetSelfFuncName(), err.Error(), msgs.UID, i)
return getDelMaxSeqByIDList(*delMsgIDList) + 1, utils.Wrap(err, "") return delStruct.getSetMinSeq(), utils.Wrap(err, "")
} }
} }
// 递归结束 // 递归结束
@ -127,17 +132,18 @@ func deleteMongoMsg(operationID string, ID string, index int64, delMsgIDList *[]
} }
} }
// 该列表中消息全部为老消息并且列表满了, 加入删除列表继续递归 // 该列表中消息全部为老消息并且列表满了, 加入删除列表继续递归
lastMsgPb := &server_api_params.MsgData{}
err = proto.Unmarshal(msgs.Msg[len(msgs.Msg)-1].Msg, lastMsgPb)
if err != nil {
log.NewError(operationID, utils.GetSelfFuncName(), err.Error(), len(msgs.Msg)-1, msgs.UID)
return 0, utils.Wrap(err, "proto.Unmarshal failed")
}
delStruct.minSeq = lastMsgPb.Seq
if msgListIsFull(msgs) { if msgListIsFull(msgs) {
msgPb := &server_api_params.MsgData{} delStruct.delUidList = append(delStruct.delUidList, msgs.UID)
err = proto.Unmarshal(msgs.Msg[len(msgs.Msg)-1].Msg, msgPb)
if err != nil {
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})
} }
// 继续递归 index+1 // 继续递归 index+1
seq, err := deleteMongoMsg(operationID, ID, index+1, delMsgIDList) seq, err := deleteMongoMsg(operationID, ID, index+1, delStruct)
if err != nil { if err != nil {
return seq, utils.Wrap(err, "deleteMongoMsg failed") return seq, utils.Wrap(err, "deleteMongoMsg failed")
} }
@ -157,13 +163,6 @@ func msgListIsFull(chat *db.UserChat) bool {
return false return false
} }
func getDelMaxSeqByIDList(delMsgIDList [][2]interface{}) uint32 {
if len(delMsgIDList) == 0 {
return 0
}
return delMsgIDList[len(delMsgIDList)-1][1].(uint32)
}
func checkMaxSeqWithMongo(operationID, ID string, diffusionType int) error { func checkMaxSeqWithMongo(operationID, ID string, diffusionType int) error {
var seqRedis uint64 var seqRedis uint64
var err error var err error

11
pkg/common/db/mongoModel.go
Unescape View File

@ -301,7 +301,16 @@ func (d *DataBases) ReplaceMsgToBlankByIndex(suffixID string, index int) error {
} }
for i, msg := range userChat.Msg { for i, msg := range userChat.Msg {
if i <= index { if i <= index {
msg.Msg = nil msgPb := &open_im_sdk.MsgData{}
if err = proto.Unmarshal(msg.Msg, msgPb); err != nil {
continue
}
newMsgPb := &open_im_sdk.MsgData{Seq: msgPb.Seq}
bytes, err := proto.Marshal(newMsgPb)
if err != nil {
continue
}
msg.Msg = bytes
msg.SendTime = 0 msg.SendTime = 0
} }
} }

Write Preview
Loading…
Cancel
Save