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

EX MSG

Browse Source
test-errcode
wangchuxiao 3 years ago
parent 8d66128009
commit b75507396b
6 changed files with 183 additions and 60 deletions
Show Stats Download Patch File Download Diff File

39
internal/cron_task/clear_msg.go
Unescape View File

@ -82,7 +82,7 @@ func (d *delMsgRecursionStruct) getSetMinSeq() uint32 {
// index 0....19(del) 20...69 // index 0....19(del) 20...69
// seq 70 // seq 70
// set minSeq 21 // set minSeq 21
// recursion // recursion 删除list并且返回设置的最小seq
func deleteMongoMsg(operationID string, ID string, index int64, delStruct *delMsgRecursionStruct) (uint32, error) { func deleteMongoMsg(operationID string, ID string, index int64, delStruct *delMsgRecursionStruct) (uint32, error) {
// find from oldest list // find from oldest list
msgs, err := db.DB.GetUserMsgListByIndex(ID, index) msgs, err := db.DB.GetUserMsgListByIndex(ID, index)
@ -105,11 +105,13 @@ func deleteMongoMsg(operationID string, ID string, index int64, delStruct *delMs
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)
} }
var hasMsgDoNotNeedDel bool
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) {
log.NewDebug(operationID, ID, "find uid", msgs.UID) log.NewDebug(operationID, ID, "find uid", msgs.UID)
// 删除块失败 递归结束 返回0 // 删除块失败 递归结束 返回0
hasMsgDoNotNeedDel = true
if err := delMongoMsgsPhysical(delStruct.delUidList); err != nil { if err := delMongoMsgsPhysical(delStruct.delUidList); err != nil {
return 0, err return 0, err
} }
@ -120,7 +122,7 @@ func deleteMongoMsg(operationID string, ID string, index int64, delStruct *delMs
} }
// 如果不是块中第一个，就把前面比他早插入的全部设置空 seq字段除外。 // 如果不是块中第一个，就把前面比他早插入的全部设置空 seq字段除外。
if i > 0 { if i > 0 {
err = db.DB.ReplaceMsgToBlankByIndex(msgs.UID, i-1) delStruct.minSeq, 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 delStruct.getSetMinSeq(), utils.Wrap(err, "") return delStruct.getSetMinSeq(), utils.Wrap(err, "")
@ -128,6 +130,10 @@ func deleteMongoMsg(operationID string, ID string, index int64, delStruct *delMs
} }
// 递归结束 // 递归结束
return msgPb.Seq, nil return msgPb.Seq, nil
} else {
if !msgListIsFull(msgs) {
}
} }
} }
// 该列表中消息全部为老消息并且列表满了, 加入删除列表继续递归 // 该列表中消息全部为老消息并且列表满了, 加入删除列表继续递归
@ -141,15 +147,25 @@ func deleteMongoMsg(operationID string, ID string, index int64, delStruct *delMs
if msgListIsFull(msgs) { if msgListIsFull(msgs) {
log.NewDebug(operationID, "msg list is full", msgs.UID) log.NewDebug(operationID, "msg list is full", msgs.UID)
delStruct.delUidList = append(delStruct.delUidList, msgs.UID) delStruct.delUidList = append(delStruct.delUidList, msgs.UID)
} else {
// 列表没有满且没有不需要被删除的消息 代表他是最新的消息块
if !hasMsgDoNotNeedDel {
delStruct.minSeq, err = db.DB.ReplaceMsgToBlankByIndex(msgs.UID, len(msgs.Msg)-1)
if err != nil {
log.NewError(operationID, utils.GetSelfFuncName(), err.Error(), msgs.UID, "Index:", len(msgs.Msg)-1)
err = delMongoMsgsPhysical(delStruct.delUidList)
if err != nil {
return delStruct.getSetMinSeq(), err
} }
log.NewDebug(operationID, ID, "continue", delStruct) return delStruct.getSetMinSeq(), nil
}
}
}
log.NewDebug(operationID, ID, "continue to", delStruct)
// 继续递归 index+1 // 继续递归 index+1
seq, err := deleteMongoMsg(operationID, ID, index+1, delStruct) seq, err := deleteMongoMsg(operationID, ID, index+1, delStruct)
if err != nil {
return seq, utils.Wrap(err, "deleteMongoMsg failed") return seq, utils.Wrap(err, "deleteMongoMsg failed")
} }
return seq, nil
}
func msgListIsFull(chat *db.UserChat) bool { func msgListIsFull(chat *db.UserChat) bool {
index, _ := strconv.Atoi(strings.Split(chat.UID, ":")[1]) index, _ := strconv.Atoi(strings.Split(chat.UID, ":")[1])
@ -164,18 +180,11 @@ func msgListIsFull(chat *db.UserChat) bool {
return false return false
} }
func CheckGroupUserMinSeq(operationID, groupID, userID string, diffusionType int) error { func CheckGroupUserMinSeq(operationID, groupID, userID string) error {
return nil return nil
} }
func CheckUserMinSeqWithMongo(operationID, userID string, diffusionType int) error { func CheckUserMinSeqWithMongo(operationID, userID string) error {
//var seqRedis uint64
//var err error
//if diffusionType == constant.WriteDiffusion {
// seqRedis, err = db.DB.GetUserMinSeq(ID)
//} else {
// seqRedis, err = db.DB.GetGroupUserMinSeq(ID)
//}
return nil return nil
} }

4
internal/cron_task/cron_task.go
Unescape View File

@ -77,7 +77,7 @@ func StartClearMsg(operationID string, userIDList []string) {
if err := checkMaxSeqWithMongo(operationID, userID, constant.WriteDiffusion); err != nil { if err := checkMaxSeqWithMongo(operationID, userID, constant.WriteDiffusion); err != nil {
log.NewError(operationID, utils.GetSelfFuncName(), userID, err) log.NewError(operationID, utils.GetSelfFuncName(), userID, err)
} }
if err := CheckUserMinSeqWithMongo(operationID, userID, constant.WriteDiffusion); err != nil { if err := CheckUserMinSeqWithMongo(operationID, userID); err != nil {
log.NewError(operationID, utils.GetSelfFuncName(), userID, err) log.NewError(operationID, utils.GetSelfFuncName(), userID, err)
} }
} }
@ -99,7 +99,7 @@ func StartClearWorkingGroupMsg(operationID string, workingGroupIDList []string)
log.NewError(operationID, utils.GetSelfFuncName(), groupID, err) log.NewError(operationID, utils.GetSelfFuncName(), groupID, err)
} }
for _, userID := range userIDList { for _, userID := range userIDList {
if err := CheckGroupUserMinSeq(operationID, groupID, userID, constant.ReadDiffusion); err != nil { if err := CheckGroupUserMinSeq(operationID, groupID, userID); err != nil {
log.NewError(operationID, utils.GetSelfFuncName(), groupID, err) log.NewError(operationID, utils.GetSelfFuncName(), groupID, err)
} }
} }

4
internal/rpc/group/group.go
Unescape View File

@ -178,13 +178,13 @@ func (s *groupServer) CreateGroup(ctx context.Context, req *pbGroup.CreateGroupR
utils.CopyStructFields(&groupMember, us) utils.CopyStructFields(&groupMember, us)
callbackResp := CallbackBeforeMemberJoinGroup(req.OperationID, &groupMember, groupInfo.Ex) callbackResp := CallbackBeforeMemberJoinGroup(req.OperationID, &groupMember, groupInfo.Ex)
if callbackResp.ErrCode != 0 { if callbackResp.ErrCode != 0 {
log.NewError(req.OperationID, utils.GetSelfFuncName(), "callbackBeforeSendSingleMsg resp: ", callbackResp) log.NewError(req.OperationID, utils.GetSelfFuncName(), "CallbackBeforeMemberJoinGroup resp: ", callbackResp)
} }
if callbackResp.ActionCode != constant.ActionAllow { if callbackResp.ActionCode != constant.ActionAllow {
if callbackResp.ErrCode == 0 { if callbackResp.ErrCode == 0 {
callbackResp.ErrCode = 201 callbackResp.ErrCode = 201
} }
log.NewDebug(req.OperationID, utils.GetSelfFuncName(), "callbackBeforeSendSingleMsg result", "end rpc and return", callbackResp) log.NewDebug(req.OperationID, utils.GetSelfFuncName(), "CallbackBeforeMemberJoinGroup result", "end rpc and return", callbackResp)
return &pbGroup.CreateGroupResp{ return &pbGroup.CreateGroupResp{
ErrCode: int32(callbackResp.ErrCode), ErrCode: int32(callbackResp.ErrCode),
ErrMsg: callbackResp.ErrMsg, ErrMsg: callbackResp.ErrMsg,

109
pkg/common/db/extend_msg_mongo_model.go
Unescape View File

@ -1,8 +1,17 @@
package db package db
import (
"Open_IM/pkg/common/config"
"context"
"strconv"
"time"
)
const cExtendMsgSet = "extend_msg_set"
type ExtendMsgSet struct { type ExtendMsgSet struct {
ID string `bson:"id" json:"ID"` ID string `bson:"id" json:"ID"`
ExtendMsg []*ExtendMsg `bson:"extend_msg" json:"extendMsg"` ExtendMsgs []*ExtendMsg `bson:"extend_msg" json:"extendMsg"`
LatestUpdateTime int32 `bson:"latest_update_time" json:"latestUpdateTime"` LatestUpdateTime int32 `bson:"latest_update_time" json:"latestUpdateTime"`
AttachedInfo string `bson:"attached_info" json:"attachedInfo"` AttachedInfo string `bson:"attached_info" json:"attachedInfo"`
Ex string `bson:"ex" json:"ex"` Ex string `bson:"ex" json:"ex"`
@ -10,40 +19,82 @@ type ExtendMsgSet struct {
CreateTime int32 `bson:"create_time" json:"createTime"` CreateTime int32 `bson:"create_time" json:"createTime"`
} }
type ReactionExtendMsgSet struct {
TypeKey string `bson:"type_key" json:"typeKey"`
Value string `bson:"value" json:"value"`
}
type ExtendMsg struct { type ExtendMsg struct {
SendID string `bson:"send_id" json:"sendID"` Content []*ReactionExtendMsgSet `bson:"content" json:"content"`
ServerMsgID string `bson:"server_msg_id" json:"serverMsgID"` ClientMsgID string `bson:"client_msg_id" json:"clientMsgID"`
Ex string `bson:"ex" json:"ex"`
AttachedInfo string `bson:"attached_info" json:"attachedInfo"`
LikeUserIDList []string `bson:"like_user_id_list" json:"likeUserIDList"`
Content string `bson:"content" json:"content"`
ExtendMsgComments []*ExtendMsgComment `bson:"extend_msg_comments" json:"extendMsgComment"`
Vote *Vote `bson:"vote" json:"vote"`
Urls []string `bson:"urls" json:"urls"`
CreateTime int32 `bson:"create_time" json:"createTime"` CreateTime int32 `bson:"create_time" json:"createTime"`
} }
type Vote struct { //type Vote struct {
Content string `bson:"content" json:"content"` // Content string `bson:"content" json:"content"`
AttachedInfo string `bson:"attached_info" json:"attachedInfo"` // AttachedInfo string `bson:"attached_info" json:"attachedInfo"`
Ex string `bson:"ex" json:"ex"` // Ex string `bson:"ex" json:"ex"`
Options []*Options `bson:"options" json:"options"` // Options []*Options `bson:"options" json:"options"`
//}
//
//type Options struct {
// Content string `bson:"content" json:"content"`
// AttachedInfo string `bson:"attached_info" json:"attachedInfo"`
// Ex string `bson:"ex" json:"ex"`
// VoteUserIDList []string `bson:"vote_user_id_list" json:"voteUserIDList"`
//}
//
//type ExtendMsgComment struct {
// UserID string `bson:"user_id" json:"userID"`
// ReplyUserID string `bson:"reply_user_id" json:"replyUserID"`
// ReplyContentID string `bson:"reply_content_id" json:"replyContentID"`
// ContentID string `bson:"content_id" json:"contentID"`
// Content string `bson:"content" json:"content"`
// CreateTime int32 `bson:"create_time" json:"createTime"`
// AttachedInfo string `bson:"attached_info" json:"attachedInfo"`
// Ex string `bson:"ex" json:"ex"`
//}
func GetExtendMsgSetID(ID string, index int32) string {
return ID + ":" + strconv.Itoa(int(index))
} }
type Options struct { func (d *DataBases) CreateExtendMsgSet(set *ExtendMsgSet) error {
Content string `bson:"content" json:"content"` ctx, _ := context.WithTimeout(context.Background(), time.Duration(config.Config.Mongo.DBTimeout)*time.Second)
AttachedInfo string `bson:"attached_info" json:"attachedInfo"` c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cExtendMsgSet)
Ex string `bson:"ex" json:"ex"` _, err := c.InsertOne(ctx, set)
VoteUserIDList []string `bson:"vote_user_id_list" json:"voteUserIDList"` return err
} }
type ExtendMsgComment struct { func (d *DataBases) GetAllExtendMsgSet(ID string) ([]*ExtendMsgSet, error) {
UserID string `bson:"user_id" json:"userID"` //ctx, _ := context.WithTimeout(context.Background(), time.Duration(config.Config.Mongo.DBTimeout)*time.Second)
ReplyUserID string `bson:"reply_user_id" json:"replyUserID"` //c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cExtendMsgSet)
ReplyContentID string `bson:"reply_content_id" json:"replyContentID"`
ContentID string `bson:"content_id" json:"contentID"` }
Content string `bson:"content" json:"content"`
CreateTime int32 `bson:"create_time" json:"createTime"` type GetExtendMsgSetOpts struct {
AttachedInfo string `bson:"attached_info" json:"attachedInfo"` IncludeExtendMsgs bool
Ex string `bson:"ex" json:"ex"` }
func (d *DataBases) GetExtendMsgSet(ID string, index int32, opts *GetExtendMsgSetOpts) (*ExtendMsgSet, error) {
//ctx, _ := context.WithTimeout(context.Background(), time.Duration(config.Config.Mongo.DBTimeout)*time.Second)
//c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cExtendMsgSet)
}
func (d *DataBases) InsertExtendMsg(ID string, msg *ExtendMsg) error {
//ctx, _ := context.WithTimeout(context.Background(), time.Duration(config.Config.Mongo.DBTimeout)*time.Second)
//c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cExtendMsgSet)
return nil
}
func (d *DataBases) UpdateOneExtendMsgSet(ID string, index, MsgIndex int32, msg *ExtendMsg, msgSet *ExtendMsgSet) error {
//ctx, _ := context.WithTimeout(context.Background(), time.Duration(config.Config.Mongo.DBTimeout)*time.Second)
//c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cExtendMsgSet)
return nil
}
func (d *DataBases) GetExtendMsgList(ID string, index, msgStartIndex, msgEndIndex int32) ([]*ExtendMsgSet, error) {
//ctx, _ := context.WithTimeout(context.Background(), time.Duration(config.Config.Mongo.DBTimeout)*time.Second)
//c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cExtendMsgSet)
return nil, nil
} }

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

@ -291,13 +291,13 @@ func (d *DataBases) DelMongoMsgs(IDList []string) error {
return err return err
} }
func (d *DataBases) ReplaceMsgToBlankByIndex(suffixID string, index int) error { func (d *DataBases) ReplaceMsgToBlankByIndex(suffixID string, index int) (replaceMaxSeq uint32, err error) {
ctx, _ := context.WithTimeout(context.Background(), time.Duration(config.Config.Mongo.DBTimeout)*time.Second) ctx, _ := context.WithTimeout(context.Background(), time.Duration(config.Config.Mongo.DBTimeout)*time.Second)
c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cChat) c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cChat)
userChat := &UserChat{} userChat := &UserChat{}
err := c.FindOne(ctx, bson.M{"uid": suffixID}).Decode(&userChat) err = c.FindOne(ctx, bson.M{"uid": suffixID}).Decode(&userChat)
if err != nil { if err != nil {
return err return 0, err
} }
for i, msg := range userChat.Msg { for i, msg := range userChat.Msg {
if i <= index { if i <= index {
@ -312,10 +312,11 @@ func (d *DataBases) ReplaceMsgToBlankByIndex(suffixID string, index int) error {
} }
msg.Msg = bytes msg.Msg = bytes
msg.SendTime = 0 msg.SendTime = 0
replaceMaxSeq = msgPb.Seq
} }
} }
_, err = c.UpdateOne(ctx, bson.M{"uid": suffixID}, bson.M{"$set": bson.M{"msg": userChat.Msg}}) _, err = c.UpdateOne(ctx, bson.M{"uid": suffixID}, bson.M{"$set": bson.M{"msg": userChat.Msg}})
return err return replaceMaxSeq, err
} }
func (d *DataBases) GetNewestMsg(ID string) (msg *open_im_sdk.MsgData, err error) { func (d *DataBases) GetNewestMsg(ID string) (msg *open_im_sdk.MsgData, err error) {

62
pkg/common/db/rocks_cache/rocks_cache.go
Unescape View File

@ -33,6 +33,8 @@ const (
groupMemberNumCache = "GROUP_MEMBER_NUM_CACHE:" groupMemberNumCache = "GROUP_MEMBER_NUM_CACHE:"
conversationCache = "CONVERSATION_CACHE:" conversationCache = "CONVERSATION_CACHE:"
conversationIDListCache = "CONVERSATION_ID_LIST_CACHE:" conversationIDListCache = "CONVERSATION_ID_LIST_CACHE:"
extendMsgSetCache = "EXTEND_MSG_SET_CACHE:"
extendMsgCache = "EXTEND_MSG_CACHE:"
) )
func DelKeys() { func DelKeys() {
@ -564,3 +566,63 @@ func GetUserAllConversationList(ownerUserID string) ([]db.Conversation, error) {
func DelConversationFromCache(ownerUserID, conversationID string) error { func DelConversationFromCache(ownerUserID, conversationID string) error {
return utils.Wrap(db.DB.Rc.TagAsDeleted(conversationCache+ownerUserID+":"+conversationID), "DelConversationFromCache err") return utils.Wrap(db.DB.Rc.TagAsDeleted(conversationCache+ownerUserID+":"+conversationID), "DelConversationFromCache err")
} }
func GetExtendMsgSetFromCache(ID string, index int32) (*db.ExtendMsgSet, error) {
getExtendMsgSet := func() (string, error) {
extendMsgSet, err := db.DB.GetExtendMsgSet(ID, index, &db.GetExtendMsgSetOpts{IncludeExtendMsgs: false})
if err != nil {
return "", utils.Wrap(err, "GetExtendMsgSet failed")
}
bytes, err := json.Marshal(extendMsgSet)
if err != nil {
return "", utils.Wrap(err, "Marshal failed")
}
return string(bytes), nil
}
extendMsgSetStr, err := db.DB.Rc.Fetch(extendMsgSetCache+db.GetExtendMsgSetID(ID, index), time.Second*30*60, getExtendMsgSet)
if err != nil {
return nil, utils.Wrap(err, "Fetch failed")
}
extendMsgSet := &db.ExtendMsgSet{}
err = json.Unmarshal([]byte(extendMsgSetStr), extendMsgSet)
if err != nil {
return nil, utils.Wrap(err, "Unmarshal failed")
}
return extendMsgSet, nil
}
func DelExtendMsgSetFromCache(ID string, index int32) error {
return utils.Wrap(db.DB.Rc.TagAsDeleted(extendMsgSetCache+db.GetExtendMsgSetID(ID, index)), "DelExtendMsgSetFromCache err")
}
func GetExtendMsg(ID string, index, extendMsgIndex int32) (*db.ExtendMsg, error) {
getExtendMsg := func() (string, error) {
extendMsg, err := db.DB.GetExtendMsgList(ID, index, extendMsgIndex, extendMsgIndex+1)
if err != nil {
return "", utils.Wrap(err, "GetExtendMsgList failed")
}
if len(extendMsg) == 0 {
return "", nil
}
bytes, err := json.Marshal(extendMsg[0])
if err != nil {
return "", utils.Wrap(err, "Marshal failed")
}
return string(bytes), nil
}
extendMsgStr, err := db.DB.Rc.Fetch(extendMsgCache+db.GetExtendMsgSetID(ID, index)+":"+strconv.Itoa(int(extendMsgIndex)), time.Second*30*60, getExtendMsg)
if err != nil {
return nil, utils.Wrap(err, "Fetch failed")
}
extendMsg := &db.ExtendMsg{}
err = json.Unmarshal([]byte(extendMsgStr), extendMsg)
if err != nil {
return nil, utils.Wrap(err, "Unmarshal failed")
}
return extendMsg, nil
}
func DelExtendMsg(ID string, index, extendMsgIndex int32) error {
return utils.Wrap(db.DB.Rc.TagAsDeleted(extendMsgCache+db.GetExtendMsgSetID(ID, index)+":"+strconv.Itoa(int(extendMsgIndex))), "DelExtendMsg err")
}

Write Preview
Loading…
Cancel
Save