blackteay/open-im-server
1
0
Fork 0
mirror of https://github.com/openimsdk/open-im-server.git synced 2025-04-25 19:22:46 +08:00
Code Issues Packages Projects Releases Wiki Activity

log

Browse Source
This commit is contained in:
wangchuxiao 2023-05-10 20:27:39 +08:00
parent 042d4ca670
commit 150089c2b9
11 changed files with 26 additions and 41 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
go.mod
View File

@ -43,7 +43,7 @@ require (
require github.com/google/uuid v1.3.0
require (
github.com/OpenIMSDK/openKeeper v0.0.8
github.com/OpenIMSDK/openKeeper v0.0.9
github.com/go-redis/redis v6.15.9+incompatible
github.com/go-sql-driver/mysql v1.6.0
k8s.io/apimachinery v0.27.1

4
go.sum
View File

@ -379,8 +379,8 @@ firebase.google.com/go v3.13.0+incompatible/go.mod h1:xlah6XbEyW6tbfSklcfe5FHJIw
github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
github.com/BurntSushi/xgb v0.0.0-20160522181843-27f122750802/go.mod h1:IVnqGOEym/WlBOVXweHU+Q+/VP0lqqI8lqeDx9IjBqo=
github.com/OneOfOne/xxhash v1.2.2/go.mod h1:HSdplMjZKSmBqAxg5vPj2TmRDmfkzw+cTzAElWljhcU=
github.com/OpenIMSDK/openKeeper v0.0.8 h1:HUL8ZPsrIA4fJ86TZ+cBbsRnUA1ZFgGs0YiXRpiuM+E=
github.com/OpenIMSDK/openKeeper v0.0.8/go.mod h1:HwHgo+p3E+aAeiLOSyDwVB2zFJtu+QvnFNlfNjFG9vA=
github.com/OpenIMSDK/openKeeper v0.0.9 h1:KzXHvWivCCOa573DKd6L38Usr1NLFT29gqRupkNOZHM=
github.com/OpenIMSDK/openKeeper v0.0.9/go.mod h1:HwHgo+p3E+aAeiLOSyDwVB2zFJtu+QvnFNlfNjFG9vA=
github.com/OpenIMSDK/open_utils v1.0.8 h1:IopxWgJwEF5ZAPsRuiZZOfcxNOQOCt/p8VDENcHN9r4=
github.com/OpenIMSDK/open_utils v1.0.8/go.mod h1:FLoaQblWUVKQgqt2LrNzfSZLT6D3DICBn1kcOMDLUOI=
github.com/Shopify/sarama v1.29.0 h1:ARid8o8oieau9XrHI55f/L3EoRAhm9px6sonbD7yuUE=

2
internal/msgtransfer/online_history_msg_handler.go
View File

@ -20,7 +20,7 @@ import (
"github.com/OpenIMSDK/Open-IM-Server/pkg/utils"
"github.com/Shopify/sarama"
"github.com/go-redis/redis"
"github.com/golang/protobuf/proto"
"google.golang.org/protobuf/proto"
)
const ConsumerMsgs = 3

1
internal/msgtransfer/online_msg_to_mongo_handler.go
View File

@ -104,7 +104,6 @@ func (mc *OnlineHistoryMongoConsumerHandler) ConsumeClaim(sess sarama.ConsumerGr
for msg := range claim.Messages() {
ctx := mc.historyConsumerGroup.GetContextFromMsg(msg)
if len(msg.Value) != 0 {
log.ZDebug(ctx, "mongo consumer recv new msg", "conversationID", msg.Key, "offset", msg.Offset)
mc.handleChatWs2Mongo(ctx, msg, string(msg.Key), sess)
} else {
log.ZError(ctx, "mongo msg get from kafka but is nil", nil, "conversationID", msg.Key)

1
pkg/a2r/api2rpc.go
View File

@ -15,6 +15,7 @@ func Call[A, B, C any](
client func(ctx context.Context) (C, error),
c *gin.Context,
) {
log.ZDebug(c, "before bind")
var req A
if err := c.BindJSON(&req); err != nil {
log.ZWarn(c, "gin bind json error", err, "req", req)

2
pkg/common/db/controller/common_msg.go
View File

@ -26,7 +26,7 @@ import (
"github.com/go-redis/redis/v8"
"go.mongodb.org/mongo-driver/mongo"
"github.com/golang/protobuf/proto"
"google.golang.org/protobuf/proto"
)
type CommonMsgDatabase interface {

14
pkg/common/db/controller/notification.go
View File

@ -25,7 +25,7 @@ import (
"github.com/go-redis/redis/v8"
"go.mongodb.org/mongo-driver/mongo"
"github.com/golang/protobuf/proto"
"google.golang.org/protobuf/proto"
)
type NotificationDatabase interface {
@ -229,7 +229,6 @@ func (db *notificationDatabase) GetGroupMinSeq(ctx context.Context, groupID stri
}
func (db *notificationDatabase) BatchInsertChat2DB(ctx context.Context, conversationID string, msgList []*sdkws.MsgData, currentMaxSeq int64) error {
//newTime := utils.GetCurrentTimestampByMill()
if int64(len(msgList)) > db.msg.GetsingleGocNotificationNum() {
return errors.New("too large")
}
@ -251,7 +250,6 @@ func (db *notificationDatabase) BatchInsertChat2DB(ctx context.Context, conversa
docIDNext := ""
var err error
for _, m := range msgList {
//log.Debug(operationID, "msg node ", m.String(), m.MsgData.ClientMsgID)
currentMaxSeq++
sMsg := unRelationTb.NotificationInfoModel{}
sMsg.SendTime = m.SendTime
@ -263,18 +261,13 @@ func (db *notificationDatabase) BatchInsertChat2DB(ctx context.Context, conversa
msgsToMongo = append(msgsToMongo, sMsg)
insertCounter++
docID = db.msg.GetDocID(conversationID, currentMaxSeq)
//log.Debug(operationID, "msgListToMongo ", seqUid, m.MsgData.Seq, m.MsgData.ClientMsgID, insertCounter, remain, "userID: ", userID)
} else {
msgsToMongoNext = append(msgsToMongoNext, sMsg)
docIDNext = db.msg.GetDocID(conversationID, currentMaxSeq)
//log.Debug(operationID, "msgListToMongoNext ", seqUidNext, m.MsgData.Seq, m.MsgData.ClientMsgID, insertCounter, remain, "userID: ", userID)
}
}
if docID != "" {
//filter := bson.M{"uid": seqUid}
//log.NewDebug(operationID, "filter ", seqUid, "list ", msgListToMongo, "userID: ", userID)
//err := c.FindOneAndUpdate(ctx, filter, bson.M{"$push": bson.M{"msg": bson.M{"$each": msgsToMongo}}}).Err()
err = db.msgDocDatabase.PushMsgsToDoc(ctx, docID, msgsToMongo)
if err != nil {
if err == mongo.ErrNoDocuments {
@ -283,13 +276,11 @@ func (db *notificationDatabase) BatchInsertChat2DB(ctx context.Context, conversa
doc.Msg = msgsToMongo
if err = db.msgDocDatabase.Create(ctx, doc); err != nil {
prome.Inc(prome.MsgInsertMongoFailedCounter)
//log.NewError(operationID, "InsertOne failed", filter, err.Error(), sChat)
return utils.Wrap(err, "")
}
prome.Inc(prome.MsgInsertMongoSuccessCounter)
} else {
prome.Inc(prome.MsgInsertMongoFailedCounter)
//log.Error(operationID, "FindOneAndUpdate failed ", err.Error(), filter)
return utils.Wrap(err, "")
}
} else {
@ -300,15 +291,12 @@ func (db *notificationDatabase) BatchInsertChat2DB(ctx context.Context, conversa
nextDoc := &unRelationTb.NotificationDocModel{}
nextDoc.DocID = docIDNext
nextDoc.Msg = msgsToMongoNext
//log.NewDebug(operationID, "filter ", seqUidNext, "list ", msgListToMongoNext, "userID: ", userID)
if err = db.msgDocDatabase.Create(ctx, nextDoc); err != nil {
prome.Inc(prome.MsgInsertMongoFailedCounter)
//log.NewError(operationID, "InsertOne failed", filter, err.Error(), sChat)
return utils.Wrap(err, "")
}
prome.Inc(prome.MsgInsertMongoSuccessCounter)
}
//log.Debug(operationID, "batch mgo cost time ", mongo2.getCurrentTimestampByMill()-newTime, userID, len(msgList))
return nil
}

2
pkg/common/db/unrelation/extend_msg.go
View File

@ -36,7 +36,7 @@ func (e *ExtendMsgSetMongoDriver) GetAllExtendMsgSet(ctx context.Context, ID str
findOpts.SetProjection(bson.M{"extend_msgs": 0})
}
}
cursor, err := e.ExtendMsgSetCollection.Find(ctx, bson.M{"uid": primitive.Regex{Pattern: regex}}, findOpts)
cursor, err := e.ExtendMsgSetCollection.Find(ctx, bson.M{"doc_id": primitive.Regex{Pattern: regex}}, findOpts)
if err != nil {
return nil, utils.Wrap(err, "")
}

19
pkg/common/db/unrelation/msg.go
View File

@ -30,8 +30,7 @@ func NewMsgMongoDriver(database *mongo.Database) table.MsgDocModelInterface {
}
func (m *MsgMongoDriver) PushMsgsToDoc(ctx context.Context, docID string, msgsToMongo []table.MsgInfoModel) error {
filter := bson.M{"uid": docID}
return m.MsgCollection.FindOneAndUpdate(ctx, filter, bson.M{"$push": bson.M{"msg": bson.M{"$each": msgsToMongo}}}).Err()
return m.MsgCollection.FindOneAndUpdate(ctx, bson.M{"doc_id": docID}, bson.M{"$push": bson.M{"msg": bson.M{"$each": msgsToMongo}}}).Err()
}
func (m *MsgMongoDriver) Create(ctx context.Context, model *table.MsgDocModel) error {
@ -45,7 +44,7 @@ func (m *MsgMongoDriver) UpdateMsgStatusByIndexInOneDoc(ctx context.Context, doc
if err != nil {
return utils.Wrap(err, "")
}
_, err = m.MsgCollection.UpdateOne(ctx, bson.M{"uid": docID}, bson.M{"$set": bson.M{fmt.Sprintf("msg.%d.msg", seqIndex): bytes}})
_, err = m.MsgCollection.UpdateOne(ctx, bson.M{"doc_id": docID}, bson.M{"$set": bson.M{fmt.Sprintf("msg.%d.msg", seqIndex): bytes}})
if err != nil {
return utils.Wrap(err, "")
}
@ -54,13 +53,13 @@ func (m *MsgMongoDriver) UpdateMsgStatusByIndexInOneDoc(ctx context.Context, doc
func (m *MsgMongoDriver) FindOneByDocID(ctx context.Context, docID string) (*table.MsgDocModel, error) {
doc := &table.MsgDocModel{}
err := m.MsgCollection.FindOne(ctx, bson.M{"uid": docID}).Decode(doc)
err := m.MsgCollection.FindOne(ctx, bson.M{"doc_id": docID}).Decode(doc)
return doc, err
}
func (m *MsgMongoDriver) GetMsgsByIndex(ctx context.Context, conversationID string, index int64) (*table.MsgDocModel, error) {
findOpts := options.Find().SetLimit(1).SetSkip(index).SetSort(bson.M{"uid": 1})
cursor, err := m.MsgCollection.Find(ctx, bson.M{"uid": primitive.Regex{Pattern: fmt.Sprintf("^%s:", conversationID)}}, findOpts)
findOpts := options.Find().SetLimit(1).SetSkip(index).SetSort(bson.M{"doc_id": 1})
cursor, err := m.MsgCollection.Find(ctx, bson.M{"doc_id": primitive.Regex{Pattern: fmt.Sprintf("^%s:", conversationID)}}, findOpts)
if err != nil {
return nil, utils.Wrap(err, "")
}
@ -77,7 +76,7 @@ func (m *MsgMongoDriver) GetMsgsByIndex(ctx context.Context, conversationID stri
func (m *MsgMongoDriver) GetNewestMsg(ctx context.Context, conversationID string) (*table.MsgInfoModel, error) {
var msgDocs []table.MsgDocModel
cursor, err := m.MsgCollection.Find(ctx, bson.M{"uid": bson.M{"$regex": fmt.Sprintf("^%s:", conversationID)}}, options.Find().SetLimit(1).SetSort(bson.M{"uid": -1}))
cursor, err := m.MsgCollection.Find(ctx, bson.M{"doc_id": bson.M{"$regex": fmt.Sprintf("^%s:", conversationID)}}, options.Find().SetLimit(1).SetSort(bson.M{"doc_id": -1}))
if err != nil {
return nil, utils.Wrap(err, "")
}
@ -96,7 +95,7 @@ func (m *MsgMongoDriver) GetNewestMsg(ctx context.Context, conversationID string
func (m *MsgMongoDriver) GetOldestMsg(ctx context.Context, conversationID string) (*table.MsgInfoModel, error) {
var msgDocs []table.MsgDocModel
cursor, err := m.MsgCollection.Find(ctx, bson.M{"uid": bson.M{"$regex": fmt.Sprintf("^%s:", conversationID)}}, options.Find().SetLimit(1).SetSort(bson.M{"uid": 1}))
cursor, err := m.MsgCollection.Find(ctx, bson.M{"doc_id": bson.M{"$regex": fmt.Sprintf("^%s:", conversationID)}}, options.Find().SetLimit(1).SetSort(bson.M{"doc_id": 1}))
if err != nil {
return nil, err
}
@ -126,12 +125,12 @@ func (m *MsgMongoDriver) Delete(ctx context.Context, docIDs []string) error {
if docIDs == nil {
return nil
}
_, err := m.MsgCollection.DeleteMany(ctx, bson.M{"uid": bson.M{"$in": docIDs}})
_, err := m.MsgCollection.DeleteMany(ctx, bson.M{"doc_id": bson.M{"$in": docIDs}})
return err
}
func (m *MsgMongoDriver) UpdateOneDoc(ctx context.Context, msg *table.MsgDocModel) error {
_, err := m.MsgCollection.UpdateOne(ctx, bson.M{"uid": msg.DocID}, bson.M{"$set": bson.M{"msg": msg.Msg}})
_, err := m.MsgCollection.UpdateOne(ctx, bson.M{"doc_id": msg.DocID}, bson.M{"$set": bson.M{"msg": msg.Msg}})
return err
}

18
pkg/common/db/unrelation/notification.go
View File

@ -28,7 +28,7 @@ func NewNotificationMongoDriver(database *mongo.Database) table.NotificationDocM
}
func (m *NotificationMongoDriver) PushMsgsToDoc(ctx context.Context, docID string, msgsToMongo []table.NotificationInfoModel) error {
filter := bson.M{"uid": docID}
filter := bson.M{"doc_id": docID}
return m.MsgCollection.FindOneAndUpdate(ctx, filter, bson.M{"$push": bson.M{"msg": bson.M{"$each": msgsToMongo}}}).Err()
}
@ -43,7 +43,7 @@ func (m *NotificationMongoDriver) UpdateMsgStatusByIndexInOneDoc(ctx context.Con
if err != nil {
return utils.Wrap(err, "")
}
_, err = m.MsgCollection.UpdateOne(ctx, bson.M{"uid": docID}, bson.M{"$set": bson.M{fmt.Sprintf("msg.%d.msg", seqIndex): bytes}})
_, err = m.MsgCollection.UpdateOne(ctx, bson.M{"doc_id": docID}, bson.M{"$set": bson.M{fmt.Sprintf("msg.%d.msg", seqIndex): bytes}})
if err != nil {
return utils.Wrap(err, "")
}
@ -52,13 +52,13 @@ func (m *NotificationMongoDriver) UpdateMsgStatusByIndexInOneDoc(ctx context.Con
func (m *NotificationMongoDriver) FindOneByDocID(ctx context.Context, docID string) (*table.NotificationDocModel, error) {
doc := &table.NotificationDocModel{}
err := m.MsgCollection.FindOne(ctx, bson.M{"uid": docID}).Decode(doc)
err := m.MsgCollection.FindOne(ctx, bson.M{"doc_id": docID}).Decode(doc)
return doc, err
}
func (m *NotificationMongoDriver) GetMsgsByIndex(ctx context.Context, conversationID string, index int64) (*table.NotificationDocModel, error) {
findOpts := options.Find().SetLimit(1).SetSkip(index).SetSort(bson.M{"uid": 1})
cursor, err := m.MsgCollection.Find(ctx, bson.M{"uid": primitive.Regex{Pattern: fmt.Sprintf("^%s:", conversationID)}}, findOpts)
findOpts := options.Find().SetLimit(1).SetSkip(index).SetSort(bson.M{"doc_id": 1})
cursor, err := m.MsgCollection.Find(ctx, bson.M{"doc_id": primitive.Regex{Pattern: fmt.Sprintf("^%s:", conversationID)}}, findOpts)
if err != nil {
return nil, utils.Wrap(err, "")
}
@ -75,7 +75,7 @@ func (m *NotificationMongoDriver) GetMsgsByIndex(ctx context.Context, conversati
func (m *NotificationMongoDriver) GetNewestMsg(ctx context.Context, conversationID string) (*table.NotificationInfoModel, error) {
var msgDocs []table.NotificationDocModel
cursor, err := m.MsgCollection.Find(ctx, bson.M{"uid": bson.M{"$regex": fmt.Sprintf("^%s:", conversationID)}}, options.Find().SetLimit(1).SetSort(bson.M{"uid": -1}))
cursor, err := m.MsgCollection.Find(ctx, bson.M{"doc_id": bson.M{"$regex": fmt.Sprintf("^%s:", conversationID)}}, options.Find().SetLimit(1).SetSort(bson.M{"doc_id": -1}))
if err != nil {
return nil, utils.Wrap(err, "")
}
@ -94,7 +94,7 @@ func (m *NotificationMongoDriver) GetNewestMsg(ctx context.Context, conversation
func (m *NotificationMongoDriver) GetOldestMsg(ctx context.Context, conversationID string) (*table.NotificationInfoModel, error) {
var msgDocs []table.NotificationDocModel
cursor, err := m.MsgCollection.Find(ctx, bson.M{"uid": bson.M{"$regex": fmt.Sprintf("^%s:", conversationID)}}, options.Find().SetLimit(1).SetSort(bson.M{"uid": 1}))
cursor, err := m.MsgCollection.Find(ctx, bson.M{"doc_id": bson.M{"$regex": fmt.Sprintf("^%s:", conversationID)}}, options.Find().SetLimit(1).SetSort(bson.M{"doc_id": 1}))
if err != nil {
return nil, err
}
@ -124,11 +124,11 @@ func (m *NotificationMongoDriver) Delete(ctx context.Context, docIDs []string) e
if docIDs == nil {
return nil
}
_, err := m.MsgCollection.DeleteMany(ctx, bson.M{"uid": bson.M{"$in": docIDs}})
_, err := m.MsgCollection.DeleteMany(ctx, bson.M{"doc_id": bson.M{"$in": docIDs}})
return err
}
func (m *NotificationMongoDriver) UpdateOneDoc(ctx context.Context, msg *table.NotificationDocModel) error {
_, err := m.MsgCollection.UpdateOne(ctx, bson.M{"uid": msg.DocID}, bson.M{"$set": bson.M{"msg": msg.Msg}})
_, err := m.MsgCollection.UpdateOne(ctx, bson.M{"doc_id": msg.DocID}, bson.M{"$set": bson.M{"msg": msg.Msg}})
return err
}

2
pkg/common/kafka/producer.go
View File

@ -4,7 +4,6 @@ import (
"context"
"errors"
"strconv"
"strings"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/config"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/constant"
@ -93,7 +92,6 @@ func (p *Producer) SendMessage(ctx context.Context, key string, m proto.Message)
for i, header := range header {
arr = append(arr, strconv.Itoa(i), string(header.Key), string(header.Value))
}
log.ZInfo(ctx, "producer.kafka.GetContextWithMQHeader", "len", len(header), "header", strings.Join(arr, ", "))
kMsg.Headers = header
partition, offset, err := p.producer.SendMessage(kMsg)
log.ZDebug(ctx, "ByteEncoder SendMessage end", "key ", kMsg.Key, "key length", kMsg.Value.Length())