IM/open-im-server
1
0
Fork 0
mirror of https://github.com/openimsdk/open-im-server.git synced 2026-05-06 01:55:58 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'tuoyun' of github.com:OpenIMSDK/Open-IM-Server into tuoyun

Browse Source
This commit is contained in:
wangchuxiao
2022-02-21 16:25:36 +08:00
parent 4d28ccebad 2171acc909
commit 079efe5de6
14 changed files with 447 additions and 203 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/common/db/model.go
+34 -4
View File
@@ -3,9 +3,16 @@ package db
import (
"Open_IM/pkg/common/config"
"Open_IM/pkg/common/log"
// "context"
// "fmt"
"github.com/garyburd/redigo/redis"
"gopkg.in/mgo.v2"
"time"
//"go.mongodb.org/mongo-driver/bson"
"go.mongodb.org/mongo-driver/mongo"
// "go.mongodb.org/mongo-driver/mongo/options"
)
var DB DataBases
@@ -14,6 +21,7 @@ type DataBases struct {
MysqlDB mysqlDB
mgoSession *mgo.Session
redisPool *redis.Pool
mongoClient *mongo.Client
}
func key(dbAddress, dbName string) string {
@@ -22,10 +30,31 @@ func key(dbAddress, dbName string) string {
func init() {
var mgoSession *mgo.Session
var mongoClient *mongo.Client
var err1 error
//mysql init
initMysqlDB()
// mongo init
// "mongodb://sysop:moon@localhost/records"
// uri := "mongodb://user:pass@sample.host:27017/?maxPoolSize=20&w=majority"
//uri := fmt.Sprintf("mongodb://%s:%s@%s/%s/?maxPoolSize=%d",
// config.Config.Mongo.DBUserName, config.Config.Mongo.DBPassword,
// config.Config.Mongo.DBAddress[0],config.Config.Mongo.DBDatabase,
// config.Config.Mongo.DBMaxPoolSize)
//
//mongoClient, err := mongo.Connect(context.TODO(), options.Client().ApplyURI(uri))
//if err != nil{
// log.NewError(" mongo.Connect failed, try ", err.Error(), uri)
// time.Sleep(time.Duration(30) * time.Second)
// mongoClient, err1 = mongo.Connect(context.TODO(), options.Client().ApplyURI(uri))
// if err1 != nil {
// log.NewError(" mongo.Connect failed, panic", err.Error(), uri)
// panic(err1.Error())
// }
//}
mgoDailInfo := &mgo.DialInfo{
Addrs: config.Config.Mongo.DBAddress,
Direct: config.Config.Mongo.DBDirect,
@@ -37,16 +66,16 @@ func init() {
PoolLimit: config.Config.Mongo.DBMaxPoolSize,
}
mgoSession, err := mgo.DialWithInfo(mgoDailInfo)
if err != nil {
log.NewError("mgo init err", err.Error(), mgoDailInfo)
}
if err != nil {
time.Sleep(time.Duration(30) * time.Second)
mgoSession, err1 = mgo.DialWithInfo(mgoDailInfo)
if err1 != nil {
log.NewError(" mongo.Connect failed, panic", err.Error())
panic(err1.Error())
}
}
DB.mongoClient = mongoClient
DB.mgoSession = mgoSession
DB.mgoSession.SetMode(mgo.Monotonic, true)
c := DB.mgoSession.DB(config.Config.Mongo.DBDatabase).C(cChat)
@@ -55,6 +84,7 @@ func init() {
panic(err.Error())
}
// redis pool init
DB.redisPool = &redis.Pool{
MaxIdle: config.Config.Redis.DBMaxIdle,
pkg/common/db/model_struct.go
+1 -1
View File
@@ -187,7 +187,7 @@ type ChatLog struct {
SessionType int32 `gorm:"column:session_type" json:"sessionType"`
MsgFrom int32 `gorm:"column:msg_from" json:"msgFrom"`
ContentType int32 `gorm:"column:content_type" json:"contentType"`
Content string `gorm:"column:content;type:varchar(1000)" json:"content"`
Content string `gorm:"column:content;type:varchar(3000)" json:"content"`
Status int32 `gorm:"column:status" json:"status"`
SendTime time.Time `gorm:"column:send_time" json:"sendTime"`
CreateTime time.Time `gorm:"column:create_time" json:"createTime"`
pkg/common/db/mongoModel.go
+226 -114
View File
@@ -6,10 +6,12 @@ import (
pbMsg "Open_IM/pkg/proto/chat"
open_im_sdk "Open_IM/pkg/proto/sdk_ws"
"Open_IM/pkg/utils"
"context"
"errors"
"github.com/garyburd/redigo/redis"
//"github.com/garyburd/redigo/redis"
"github.com/golang/protobuf/proto"
"gopkg.in/mgo.v2/bson"
"strconv"
"time"
)
@@ -34,33 +36,35 @@ type GroupMember_x struct {
}
func (d *DataBases) GetMinSeqFromMongo(uid string) (MinSeq uint32, err error) {
var i, NB uint32
var seqUid string
session := d.mgoSession.Clone()
if session == nil {
return MinSeq, errors.New("session == nil")
}
defer session.Close()
c := session.DB(config.Config.Mongo.DBDatabase).C(cChat)
MaxSeq, err := d.GetUserMaxSeq(uid)
if err != nil && err != redis.ErrNil {
return MinSeq, err
}
NB = uint32(MaxSeq / singleGocMsgNum)
for i = 0; i <= NB; i++ {
seqUid = indexGen(uid, i)
n, err := c.Find(bson.M{"uid": seqUid}).Count()
if err == nil && n != 0 {
if i == 0 {
MinSeq = 1
} else {
MinSeq = uint32(i * singleGocMsgNum)
}
break
}
}
return MinSeq, nil
return 1, nil
//var i, NB uint32
//var seqUid string
//session := d.mgoSession.Clone()
//if session == nil {
// return MinSeq, errors.New("session == nil")
//}
//defer session.Close()
//c := session.DB(config.Config.Mongo.DBDatabase).C(cChat)
//MaxSeq, err := d.GetUserMaxSeq(uid)
//if err != nil && err != redis.ErrNil {
// return MinSeq, err
//}
//NB = uint32(MaxSeq / singleGocMsgNum)
//for i = 0; i <= NB; i++ {
// seqUid = indexGen(uid, i)
// n, err := c.Find(bson.M{"uid": seqUid}).Count()
// if err == nil && n != 0 {
// if i == 0 {
// MinSeq = 1
// } else {
// MinSeq = uint32(i * singleGocMsgNum)
// }
// break
// }
//}
//return MinSeq, nil
}
func (d *DataBases) GetMsgBySeqList(uid string, seqList []uint32, operationID string) (seqMsg []*open_im_sdk.MsgData, err error) {
var hasSeqList []uint32
singleCount := 0
@@ -115,6 +119,61 @@ func (d *DataBases) GetMsgBySeqList(uid string, seqList []uint32, operationID st
}
return seqMsg, nil
}
func (d *DataBases) GetMsgBySeqListMongo2(uid string, seqList []uint32, operationID string) (seqMsg []*open_im_sdk.MsgData, err error) {
var hasSeqList []uint32
singleCount := 0
ctx, _ := context.WithTimeout(context.Background(), time.Duration(config.Config.Mongo.DBTimeout)*time.Second)
c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cChat)
m := func(uid string, seqList []uint32) map[string][]uint32 {
t := make(map[string][]uint32)
for i := 0; i < len(seqList); i++ {
seqUid := getSeqUid(uid, seqList[i])
if value, ok := t[seqUid]; !ok {
var temp []uint32
t[seqUid] = append(temp, seqList[i])
} else {
t[seqUid] = append(value, seqList[i])
}
}
return t
}(uid, seqList)
sChat := UserChat{}
for seqUid, value := range m {
if err = c.FindOne(ctx, bson.M{"uid": seqUid}).Decode(&sChat); err != nil {
log.NewError(operationID, "not find seqUid", seqUid, value, uid, seqList, err.Error())
continue
}
singleCount = 0
for i := 0; i < len(sChat.Msg); i++ {
msg := new(open_im_sdk.MsgData)
if err = proto.Unmarshal(sChat.Msg[i].Msg, msg); err != nil {
log.NewError(operationID, "Unmarshal err", seqUid, value, uid, seqList, err.Error())
return nil, err
}
if isContainInt32(msg.Seq, value) {
seqMsg = append(seqMsg, msg)
hasSeqList = append(hasSeqList, msg.Seq)
singleCount++
if singleCount == len(value) {
break
}
}
}
}
if len(hasSeqList) != len(seqList) {
var diff []uint32
diff = utils.Difference(hasSeqList, seqList)
exceptionMSg := genExceptionMessageBySeqList(diff)
seqMsg = append(seqMsg, exceptionMSg...)
}
return seqMsg, nil
}
func genExceptionMessageBySeqList(seqList []uint32) (exceptionMsg []*open_im_sdk.MsgData) {
for _, v := range seqList {
msg := new(open_im_sdk.MsgData)
@@ -124,6 +183,37 @@ func genExceptionMessageBySeqList(seqList []uint32) (exceptionMsg []*open_im_sdk
return exceptionMsg
}
func (d *DataBases) SaveUserChatMongo2(uid string, sendTime int64, m *pbMsg.MsgDataToDB) error {
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()
operationID := ""
seqUid := getSeqUid(uid, m.MsgData.Seq)
filter := bson.M{"uid": seqUid}
var err error
sMsg := MsgInfo{}
sMsg.SendTime = sendTime
if sMsg.Msg, err = proto.Marshal(m.MsgData); err != nil {
return utils.Wrap(err,"")
}
err = c.FindOneAndUpdate(ctx, filter, bson.M{"$push": bson.M{"msg": sMsg}}).Err()
log.NewDebug(operationID, "get mgoSession cost time", getCurrentTimestampByMill()-newTime)
if err != nil {
sChat := UserChat{}
sChat.UID = seqUid
sChat.Msg = append(sChat.Msg, sMsg)
if _, err = c.InsertOne(ctx, &sChat) ; err != nil{
log.NewDebug(operationID, "InsertOne failed", filter)
return utils.Wrap(err, "")
}
}else{
log.NewDebug(operationID, "FindOneAndUpdate ok", filter)
}
log.NewDebug(operationID, "find mgo uid cost time", getCurrentTimestampByMill()-newTime)
return nil
}
func (d *DataBases) SaveUserChat(uid string, sendTime int64, m *pbMsg.MsgDataToDB) error {
var seqUid string
newTime := getCurrentTimestampByMill()
@@ -163,115 +253,137 @@ func (d *DataBases) SaveUserChat(uid string, sendTime int64, m *pbMsg.MsgDataToD
return nil
}
func (d *DataBases) DelUserChat(uid string) error {
session := d.mgoSession.Clone()
if session == nil {
return errors.New("session == nil")
}
defer session.Close()
c := session.DB(config.Config.Mongo.DBDatabase).C(cChat)
func (d *DataBases) DelUserChat(uid string) error {
return nil
//session := d.mgoSession.Clone()
//if session == nil {
// return errors.New("session == nil")
//}
//defer session.Close()
//
//c := session.DB(config.Config.Mongo.DBDatabase).C(cChat)
//
//delTime := time.Now().Unix() - int64(config.Config.Mongo.DBRetainChatRecords)*24*3600
//if err := c.Update(bson.M{"uid": uid}, bson.M{"$pull": bson.M{"msg": bson.M{"sendtime": bson.M{"$lte": delTime}}}}); err != nil {
// return err
//}
//
//return nil
}
func (d *DataBases) DelUserChatMongo2(uid string) error {
ctx, _ := context.WithTimeout(context.Background(), time.Duration(config.Config.Mongo.DBTimeout)*time.Second)
c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cChat)
filter := bson.M{"uid": uid}
delTime := time.Now().Unix() - int64(config.Config.Mongo.DBRetainChatRecords)*24*3600
if err := c.Update(bson.M{"uid": uid}, bson.M{"$pull": bson.M{"msg": bson.M{"sendtime": bson.M{"$lte": delTime}}}}); err != nil {
return err
if _, err := c.UpdateOne(ctx, filter, bson.M{"$pull": bson.M{"msg": bson.M{"sendtime": bson.M{"$lte": delTime}}}}); err != nil {
return utils.Wrap(err, "")
}
return nil
}
func (d *DataBases) MgoUserCount() (int, error) {
session := d.mgoSession.Clone()
if session == nil {
return 0, errors.New("session == nil")
}
defer session.Close()
c := session.DB(config.Config.Mongo.DBDatabase).C(cChat)
return c.Find(nil).Count()
return 0, nil
//session := d.mgoSession.Clone()
//if session == nil {
// return 0, errors.New("session == nil")
//}
//defer session.Close()
//
//c := session.DB(config.Config.Mongo.DBDatabase).C(cChat)
//
//return c.Find(nil).Count()
}
func (d *DataBases) MgoSkipUID(count int) (string, error) {
session := d.mgoSession.Clone()
if session == nil {
return "", errors.New("session == nil")
}
defer session.Close()
c := session.DB(config.Config.Mongo.DBDatabase).C(cChat)
sChat := UserChat{}
c.Find(nil).Skip(count).Limit(1).One(&sChat)
return sChat.UID, nil
return "", nil
//session := d.mgoSession.Clone()
//if session == nil {
// return "", errors.New("session == nil")
//}
//defer session.Close()
//
//c := session.DB(config.Config.Mongo.DBDatabase).C(cChat)
//
//sChat := UserChat{}
//c.Find(nil).Skip(count).Limit(1).One(&sChat)
//return sChat.UID, nil
}
func (d *DataBases) GetGroupMember(groupID string) []string {
groupInfo := GroupMember_x{}
groupInfo.GroupID = groupID
groupInfo.UIDList = make([]string, 0)
session := d.mgoSession.Clone()
if session == nil {
return groupInfo.UIDList
}
defer session.Close()
c := session.DB(config.Config.Mongo.DBDatabase).C(cGroup)
if err := c.Find(bson.M{"groupid": groupInfo.GroupID}).One(&groupInfo); err != nil {
return groupInfo.UIDList
}
return groupInfo.UIDList
return nil
//groupInfo := GroupMember_x{}
//groupInfo.GroupID = groupID
//groupInfo.UIDList = make([]string, 0)
//
//session := d.mgoSession.Clone()
//if session == nil {
// return groupInfo.UIDList
//}
//defer session.Close()
//
//c := session.DB(config.Config.Mongo.DBDatabase).C(cGroup)
//
//if err := c.Find(bson.M{"groupid": groupInfo.GroupID}).One(&groupInfo); err != nil {
// return groupInfo.UIDList
//}
//
//return groupInfo.UIDList
}
func (d *DataBases) AddGroupMember(groupID, uid string) error {
session := d.mgoSession.Clone()
if session == nil {
return errors.New("session == nil")
}
defer session.Close()
c := session.DB(config.Config.Mongo.DBDatabase).C(cGroup)
n, err := c.Find(bson.M{"groupid": groupID}).Count()
if err != nil {
return err
}
if n == 0 {
groupInfo := GroupMember_x{}
groupInfo.GroupID = groupID
groupInfo.UIDList = append(groupInfo.UIDList, uid)
err = c.Insert(&groupInfo)
if err != nil {
return err
}
} else {
err = c.Update(bson.M{"groupid": groupID}, bson.M{"$addToSet": bson.M{"uidlist": uid}})
if err != nil {
return err
}
}
return nil
//session := d.mgoSession.Clone()
//if session == nil {
// return errors.New("session == nil")
//}
//defer session.Close()
//
//c := session.DB(config.Config.Mongo.DBDatabase).C(cGroup)
//
//n, err := c.Find(bson.M{"groupid": groupID}).Count()
//if err != nil {
// return err
//}
//
//if n == 0 {
// groupInfo := GroupMember_x{}
// groupInfo.GroupID = groupID
// groupInfo.UIDList = append(groupInfo.UIDList, uid)
// err = c.Insert(&groupInfo)
// if err != nil {
// return err
// }
//} else {
// err = c.Update(bson.M{"groupid": groupID}, bson.M{"$addToSet": bson.M{"uidlist": uid}})
// if err != nil {
// return err
// }
//}
//
//return nil
}
func (d *DataBases) DelGroupMember(groupID, uid string) error {
session := d.mgoSession.Clone()
if session == nil {
return errors.New("session == nil")
}
defer session.Close()
c := session.DB(config.Config.Mongo.DBDatabase).C(cGroup)
if err := c.Update(bson.M{"groupid": groupID}, bson.M{"$pull": bson.M{"uidlist": uid}}); err != nil {
return err
}
return nil
//session := d.mgoSession.Clone()
//if session == nil {
// return errors.New("session == nil")
//}
//defer session.Close()
//
//c := session.DB(config.Config.Mongo.DBDatabase).C(cGroup)
//
//if err := c.Update(bson.M{"groupid": groupID}, bson.M{"$pull": bson.M{"uidlist": uid}}); err != nil {
// return err
//}
//
//return nil
}
func getCurrentTimestampByMill() int64 {