You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
79 lines
2.3 KiB
79 lines
2.3 KiB
3 years ago
|
package db
|
||
|
|
|
||
|
|
import (
|
||
|
|
"Open_IM/pkg/common/config"
|
||
|
|
"Open_IM/pkg/common/log"
|
||
|
|
pbMsg "Open_IM/pkg/proto/chat"
|
||
|
|
"Open_IM/pkg/utils"
|
||
|
|
"context"
|
||
|
|
"errors"
|
||
|
|
"github.com/garyburd/redigo/redis"
|
||
|
|
"github.com/golang/protobuf/proto"
|
||
|
|
"go.mongodb.org/mongo-driver/bson"
|
||
|
|
"time"
|
||
|
|
)
|
||
|
|
|
||
|
|
func (d *DataBases) BatchInsertChat(userID string, msgList []*pbMsg.MsgDataToMQ, operationID string) error {
|
||
|
|
if len(msgList) > GetSingleGocMsgNum() {
|
||
|
|
return errors.New("too large")
|
||
|
|
}
|
||
|
|
currentMaxSeq, err := d.GetUserMaxSeq(userID)
|
||
|
|
if err == nil {
|
||
|
|
|
||
|
|
} else if err == redis.ErrNil {
|
||
|
|
currentMaxSeq = 0
|
||
|
|
} else {
|
||
|
|
return utils.Wrap(err, "")
|
||
|
|
}
|
||
|
|
|
||
|
|
remain := currentMaxSeq % uint64(GetSingleGocMsgNum())
|
||
|
|
insertCounter := uint64(0)
|
||
|
|
msgListToMongo := make([]MsgInfo, 0)
|
||
|
|
msgListToMongoNext := make([]MsgInfo, 0)
|
||
|
|
seqUid := ""
|
||
|
|
seqUidNext := ""
|
||
|
|
for _, m := range msgList {
|
||
|
|
currentMaxSeq++
|
||
|
|
sMsg := MsgInfo{}
|
||
|
|
sMsg.SendTime = m.MsgData.SendTime
|
||
|
|
m.MsgData.Seq = uint32(currentMaxSeq)
|
||
|
|
if sMsg.Msg, err = proto.Marshal(m.MsgData); err != nil {
|
||
|
|
return utils.Wrap(err, "")
|
||
|
|
}
|
||
|
|
if insertCounter < remain {
|
||
|
|
msgListToMongo = append(msgListToMongo, sMsg)
|
||
|
|
insertCounter++
|
||
|
|
seqUid = getSeqUid(userID, uint32(currentMaxSeq))
|
||
|
|
} else {
|
||
|
|
msgListToMongoNext = append(msgListToMongoNext, sMsg)
|
||
|
|
seqUidNext = getSeqUid(userID, uint32(currentMaxSeq))
|
||
|
|
}
|
||
|
|
}
|
||
|
|
ctx, _ := context.WithTimeout(context.Background(), time.Duration(config.Config.Mongo.DBTimeout)*time.Second)
|
||
|
|
c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cChat)
|
||
|
|
newTime := getCurrentTimestampByMill()
|
||
|
|
|
||
|
|
if seqUid != "" {
|
||
|
|
filter := bson.M{"uid": seqUid}
|
||
|
|
log.NewDebug(operationID, "filter ", seqUid)
|
||
|
|
err := c.FindOneAndUpdate(ctx, filter, bson.M{"$push": bson.M{"msg": bson.M{"$each": msgListToMongo}}}).Err()
|
||
|
|
if err != nil {
|
||
|
|
log.Error(operationID, "FindOneAndUpdate failed ", err.Error(), filter)
|
||
|
|
return utils.Wrap(err, "")
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
if seqUidNext != "" {
|
||
|
|
filter := bson.M{"uid": seqUidNext}
|
||
|
|
sChat := UserChat{}
|
||
|
|
sChat.UID = seqUidNext
|
||
|
|
sChat.Msg = msgListToMongoNext
|
||
|
|
if _, err = c.InsertOne(ctx, &sChat); err != nil {
|
||
|
|
log.NewError(operationID, "InsertOne failed", filter, err.Error(), sChat)
|
||
|
|
return utils.Wrap(err, "")
|
||
|
|
}
|
||
|
|
}
|
||
|
|
log.NewDebug(operationID, "find mgo uid cost time", getCurrentTimestampByMill()-newTime)
|
||
|
|
return nil
|
||
|
|
}
|