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

superGroupMaxSeq

Browse Source
This commit is contained in:
wangchuxiao 2022-07-22 17:51:17 +08:00
parent a5ef78fd1e
commit a3f62e2246
3 changed files with 39 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
internal/rpc/msg/pull_message.go
View File

@ -1,6 +1,7 @@
package msg package msg
import ( import (
"Open_IM/pkg/utils"
"context" "context"
go_redis "github.com/go-redis/redis/v8" go_redis "github.com/go-redis/redis/v8"
@ -14,11 +15,17 @@ func (rpc *rpcChat) GetMaxAndMinSeq(_ context.Context, in *open_im_sdk.GetMaxAnd
resp := new(open_im_sdk.GetMaxAndMinSeqResp) resp := new(open_im_sdk.GetMaxAndMinSeqResp)
m := make(map[string]*open_im_sdk.MaxAndMinSeq) m := make(map[string]*open_im_sdk.MaxAndMinSeq)
var maxSeq, minSeq uint64 var maxSeq, minSeq uint64
var err error var err1, err2 error
maxSeq, err = commonDB.DB.GetUserMaxSeq(in.UserID) maxSeq, err1 = commonDB.DB.GetUserMaxSeq(in.UserID)
minSeq, err = commonDB.DB.GetUserMinSeq(in.UserID) minSeq, err2 = commonDB.DB.GetUserMinSeq(in.UserID)
if err != nil && err != go_redis.Nil { if (err1 != nil && err1 != go_redis.Nil) || (err2 != nil && err2 != go_redis.Nil) {
log.NewError(in.OperationID, "getMaxSeq from redis error", in.String(), err.Error()) log.NewError(in.OperationID, "getMaxSeq from redis error", in.String())
if err1 != nil {
log.NewError(in.OperationID, utils.GetSelfFuncName(), err1.Error())
}
if err2 != nil {
log.NewError(in.OperationID, utils.GetSelfFuncName(), err2.Error())
}
resp.ErrCode = 200 resp.ErrCode = 200
resp.ErrMsg = "redis get err" resp.ErrMsg = "redis get err"
return resp, nil return resp, nil

2
pkg/common/db/RedisModel.go
View File

@ -108,7 +108,7 @@ func (d *DataBases) IncrGroupMaxSeq(groupID string) (uint64, error) {
return uint64(seq), err return uint64(seq), err
} }
func (d *DataBases) SetGroupMaxSeq(groupID string, maxSeq uint32) error { func (d *DataBases) SetGroupMaxSeq(groupID string, maxSeq uint64) error {
key := groupMaxSeq + groupID key := groupMaxSeq + groupID
return d.RDB.Set(context.Background(), key, maxSeq, 0).Err() return d.RDB.Set(context.Background(), key, maxSeq, 0).Err()
} }

39
pkg/common/db/batch_insert_chat.go
View File

@ -2,6 +2,7 @@ package db
import ( import (
"Open_IM/pkg/common/config" "Open_IM/pkg/common/config"
"Open_IM/pkg/common/constant"
"Open_IM/pkg/common/log" "Open_IM/pkg/common/log"
pbMsg "Open_IM/pkg/proto/msg" pbMsg "Open_IM/pkg/proto/msg"
"Open_IM/pkg/utils" "Open_IM/pkg/utils"
@ -104,21 +105,28 @@ func (d *DataBases) BatchInsertChat2DB(userID string, msgList []*pbMsg.MsgDataTo
return nil return nil
} }
func (d *DataBases) BatchInsertChat2Cache(userID string, msgList []*pbMsg.MsgDataToMQ, operationID string) (error, uint64) { func (d *DataBases) BatchInsertChat2Cache(insertID string, msgList []*pbMsg.MsgDataToMQ, operationID string) (error, uint64) {
newTime := getCurrentTimestampByMill() newTime := getCurrentTimestampByMill()
if len(msgList) > GetSingleGocMsgNum() { lenList := len(msgList)
if lenList > GetSingleGocMsgNum() {
return errors.New("too large"), 0 return errors.New("too large"), 0
} }
currentMaxSeq, err := d.GetUserMaxSeq(userID) if lenList < 1 {
if err == nil { return errors.New("too short as 0"), 0
}
} else if err == go_redis.Nil { // judge sessionType to get seq
currentMaxSeq = 0 var currentMaxSeq uint64
var err error
if msgList[0].MsgData.SessionType == constant.SuperGroupChatType {
currentMaxSeq, err = d.GetGroupMaxSeq(insertID)
} else { } else {
currentMaxSeq, err = d.GetUserMaxSeq(insertID)
}
if err != nil && err != go_redis.Nil {
return utils.Wrap(err, ""), 0 return utils.Wrap(err, ""), 0
} }
lastMaxSeq := currentMaxSeq
lastMaxSeq := currentMaxSeq
for _, m := range msgList { for _, m := range msgList {
log.Debug(operationID, "msg node ", m.String(), m.MsgData.ClientMsgID) log.Debug(operationID, "msg node ", m.String(), m.MsgData.ClientMsgID)
currentMaxSeq++ currentMaxSeq++
@ -126,13 +134,18 @@ func (d *DataBases) BatchInsertChat2Cache(userID string, msgList []*pbMsg.MsgDat
sMsg.SendTime = m.MsgData.SendTime sMsg.SendTime = m.MsgData.SendTime
m.MsgData.Seq = uint32(currentMaxSeq) m.MsgData.Seq = uint32(currentMaxSeq)
} }
log.Debug(operationID, "SetMessageToCache ", userID, len(msgList)) log.Debug(operationID, "SetMessageToCache ", insertID, len(msgList))
err = d.SetMessageToCache(msgList, userID, operationID) err = d.SetMessageToCache(msgList, insertID, operationID)
if err != nil { if err != nil {
log.Error(operationID, "setMessageToCache failed, continue ", err.Error(), len(msgList), userID) log.Error(operationID, "setMessageToCache failed, continue ", err.Error(), len(msgList), insertID)
} }
log.Debug(operationID, "batch to redis cost time ", getCurrentTimestampByMill()-newTime, userID, len(msgList)) log.Debug(operationID, "batch to redis cost time ", getCurrentTimestampByMill()-newTime, insertID, len(msgList))
return utils.Wrap(d.SetUserMaxSeq(userID, uint64(currentMaxSeq)), ""), lastMaxSeq if msgList[0].MsgData.SessionType == constant.SuperGroupChatType {
err = d.SetGroupMaxSeq(insertID, currentMaxSeq)
} else {
err = d.SetUserMaxSeq(insertID, currentMaxSeq)
}
return utils.Wrap(err, ""), lastMaxSeq
} }
//func (d *DataBases) BatchInsertChatBoth(userID string, msgList []*pbMsg.MsgDataToMQ, operationID string) (error, uint64) { //func (d *DataBases) BatchInsertChatBoth(userID string, msgList []*pbMsg.MsgDataToMQ, operationID string) (error, uint64) {