-
Notifications
You must be signed in to change notification settings - Fork 2.5k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat: support stream message (#2824)
* fix: GroupApplicationAcceptedNotification * fix: GroupApplicationAcceptedNotification * fix: NotificationUserInfoUpdate * cicd: robot automated Change * fix: component * fix: getConversationInfo * feat: cron task * feat: cron task * feat: cron task * feat: cron task * feat: cron task * fix: minio config url recognition error * update gomake version * update gomake version * fix: seq conversion bug * fix: redis pipe exec * fix: ImportFriends * fix: A large number of logs keysAndValues length is not even * feat: mark read aggregate write * feat: online status supports redis cluster * feat: online status supports redis cluster * feat: online status supports redis cluster * merge * merge * read seq is written to mongo * read seq is written to mongo * fix: invitation to join group notification * fix: friend op_user_id * feat: optimizing asynchronous context * feat: optimizing memamq size * feat: add GetSeqMessage * feat: GroupApplicationAgreeMemberEnterNotification * feat: GroupApplicationAgreeMemberEnterNotification * feat: go.mod * feat: go.mod * feat: join group notification and get seq * feat: join group notification and get seq * feat: avoid pulling messages from sessions with a large number of max seq values of 0 * feat: API supports gzip * go.mod * fix: nil pointer error on close * fix: listen error * fix: listen error * update go.mod * feat: add log * fix: token parse token value * fix: GetMsgBySeqs boundary issues * fix: sn_ not sort * fix: sn_ not sort * fix: sn_ not sort * fix: jssdk add * fix: jssdk support * fix: jssdk support * fix: jssdk support * fix: the message I sent is not set to read seq in mongodb * fix: cannot modify group member avatars * fix: MemberEnterNotification * fix: MemberEnterNotification * fix: MsgData status * feat: stream msg * feat: support stream messages --------- Co-authored-by: withchao <[email protected]>
- Loading branch information
Showing
15 changed files
with
441 additions
and
5 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,114 @@ | ||
| package msg | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "github.com/openimsdk/open-im-server/v3/pkg/common/storage/model" | ||
| "github.com/openimsdk/open-im-server/v3/pkg/msgprocessor" | ||
| "github.com/openimsdk/protocol/constant" | ||
| "github.com/openimsdk/protocol/msg" | ||
| "github.com/openimsdk/protocol/sdkws" | ||
| "github.com/openimsdk/tools/errs" | ||
| "time" | ||
| ) | ||
|
|
||
| const StreamDeadlineTime = time.Second * 60 * 10 | ||
|
|
||
| func (m *msgServer) handlerStreamMsg(ctx context.Context, msgData *sdkws.MsgData) error { | ||
| now := time.Now() | ||
| val := &model.StreamMsg{ | ||
| ClientMsgID: msgData.ClientMsgID, | ||
| ConversationID: msgprocessor.GetConversationIDByMsg(msgData), | ||
| UserID: msgData.SendID, | ||
| CreateTime: now, | ||
| DeadlineTime: now.Add(StreamDeadlineTime), | ||
| } | ||
| return m.StreamMsgDatabase.CreateStreamMsg(ctx, val) | ||
| } | ||
|
|
||
| func (m *msgServer) getStreamMsg(ctx context.Context, clientMsgID string) (*model.StreamMsg, error) { | ||
| res, err := m.StreamMsgDatabase.GetStreamMsg(ctx, clientMsgID) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| now := time.Now() | ||
| if !res.End && res.DeadlineTime.Before(now) { | ||
| res.End = true | ||
| res.DeadlineTime = now | ||
| _ = m.StreamMsgDatabase.AppendStreamMsg(ctx, res.ClientMsgID, 0, nil, true, now) | ||
| } | ||
| return res, nil | ||
| } | ||
|
|
||
| func (m *msgServer) AppendStreamMsg(ctx context.Context, req *msg.AppendStreamMsgReq) (*msg.AppendStreamMsgResp, error) { | ||
| res, err := m.getStreamMsg(ctx, req.ClientMsgID) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| if res.End { | ||
| return nil, errs.ErrNoPermission.WrapMsg("stream msg is end") | ||
| } | ||
| if len(res.Packets) < int(req.StartIndex) { | ||
| return nil, errs.ErrNoPermission.WrapMsg("start index is invalid") | ||
| } | ||
| if val := len(res.Packets) - int(req.StartIndex); val > 0 { | ||
| exist := res.Packets[int(req.StartIndex):] | ||
| for i, s := range exist { | ||
| if len(req.Packets) == 0 { | ||
| break | ||
| } | ||
| if s != req.Packets[i] { | ||
| return nil, errs.ErrNoPermission.WrapMsg(fmt.Sprintf("packet %d has been written and is inconsistent", i)) | ||
| } | ||
| req.StartIndex++ | ||
| req.Packets = req.Packets[1:] | ||
| } | ||
| } | ||
| if len(req.Packets) == 0 && res.End == req.End { | ||
| return &msg.AppendStreamMsgResp{}, nil | ||
| } | ||
| deadlineTime := time.Now().Add(StreamDeadlineTime) | ||
| if err := m.StreamMsgDatabase.AppendStreamMsg(ctx, req.ClientMsgID, int(req.StartIndex), req.Packets, req.End, deadlineTime); err != nil { | ||
| return nil, err | ||
| } | ||
| conversation, err := m.Conversation.GetConversation(ctx, res.UserID, res.ConversationID) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| tips := &sdkws.StreamMsgTips{ | ||
| ConversationID: res.ConversationID, | ||
| ClientMsgID: res.ClientMsgID, | ||
| StartIndex: req.StartIndex, | ||
| Packets: req.Packets, | ||
| End: req.End, | ||
| } | ||
| var ( | ||
| recvID string | ||
| sessionType int32 | ||
| ) | ||
| if conversation.GroupID == "" { | ||
| sessionType = constant.SingleChatType | ||
| recvID = conversation.UserID | ||
| } else { | ||
| sessionType = constant.ReadGroupChatType | ||
| recvID = conversation.GroupID | ||
| } | ||
| m.msgNotificationSender.StreamMsgNotification(ctx, res.UserID, recvID, sessionType, tips) | ||
| return &msg.AppendStreamMsgResp{}, nil | ||
| } | ||
|
|
||
| func (m *msgServer) GetStreamMsg(ctx context.Context, req *msg.GetStreamMsgReq) (*msg.GetStreamMsgResp, error) { | ||
| res, err := m.getStreamMsg(ctx, req.ClientMsgID) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| return &msg.GetStreamMsgResp{ | ||
| ClientMsgID: res.ClientMsgID, | ||
| ConversationID: res.ConversationID, | ||
| UserID: res.UserID, | ||
| Packets: res.Packets, | ||
| End: res.End, | ||
| CreateTime: res.CreateTime.UnixMilli(), | ||
| DeadlineTime: res.DeadlineTime.UnixMilli(), | ||
| }, nil | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,34 @@ | ||
| package controller | ||
|
|
||
| import ( | ||
| "context" | ||
| "github.com/openimsdk/open-im-server/v3/pkg/common/storage/database" | ||
| "github.com/openimsdk/open-im-server/v3/pkg/common/storage/model" | ||
| "time" | ||
| ) | ||
|
|
||
| type StreamMsgDatabase interface { | ||
| CreateStreamMsg(ctx context.Context, model *model.StreamMsg) error | ||
| AppendStreamMsg(ctx context.Context, clientMsgID string, startIndex int, packets []string, end bool, deadlineTime time.Time) error | ||
| GetStreamMsg(ctx context.Context, clientMsgID string) (*model.StreamMsg, error) | ||
| } | ||
|
|
||
| func NewStreamMsgDatabase(db database.StreamMsg) StreamMsgDatabase { | ||
| return &streamMsgDatabase{db: db} | ||
| } | ||
|
|
||
| type streamMsgDatabase struct { | ||
| db database.StreamMsg | ||
| } | ||
|
|
||
| func (m *streamMsgDatabase) CreateStreamMsg(ctx context.Context, model *model.StreamMsg) error { | ||
| return m.db.CreateStreamMsg(ctx, model) | ||
| } | ||
|
|
||
| func (m *streamMsgDatabase) AppendStreamMsg(ctx context.Context, clientMsgID string, startIndex int, packets []string, end bool, deadlineTime time.Time) error { | ||
| return m.db.AppendStreamMsg(ctx, clientMsgID, startIndex, packets, end, deadlineTime) | ||
| } | ||
|
|
||
| func (m *streamMsgDatabase) GetStreamMsg(ctx context.Context, clientMsgID string) (*model.StreamMsg, error) { | ||
| return m.db.GetStreamMsg(ctx, clientMsgID) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,60 @@ | ||
| package mgo | ||
|
|
||
| import ( | ||
| "context" | ||
| "github.com/openimsdk/open-im-server/v3/pkg/common/storage/database" | ||
| "github.com/openimsdk/open-im-server/v3/pkg/common/storage/model" | ||
| "github.com/openimsdk/tools/db/mongoutil" | ||
| "github.com/openimsdk/tools/errs" | ||
| "go.mongodb.org/mongo-driver/bson" | ||
| "go.mongodb.org/mongo-driver/mongo" | ||
| "go.mongodb.org/mongo-driver/mongo/options" | ||
| "time" | ||
| ) | ||
|
|
||
| func NewStreamMsgMongo(db *mongo.Database) (*StreamMsgMongo, error) { | ||
| coll := db.Collection(database.StreamMsgName) | ||
| _, err := coll.Indexes().CreateOne(context.Background(), mongo.IndexModel{ | ||
| Keys: bson.D{ | ||
| {Key: "client_msg_id", Value: 1}, | ||
| }, | ||
| Options: options.Index().SetUnique(true), | ||
| }) | ||
| if err != nil { | ||
| return nil, errs.Wrap(err) | ||
| } | ||
| return &StreamMsgMongo{coll: coll}, nil | ||
| } | ||
|
|
||
| type StreamMsgMongo struct { | ||
| coll *mongo.Collection | ||
| } | ||
|
|
||
| func (m *StreamMsgMongo) CreateStreamMsg(ctx context.Context, val *model.StreamMsg) error { | ||
| if val.Packets == nil { | ||
| val.Packets = []string{} | ||
| } | ||
| return mongoutil.InsertMany(ctx, m.coll, []*model.StreamMsg{val}) | ||
| } | ||
|
|
||
| func (m *StreamMsgMongo) AppendStreamMsg(ctx context.Context, clientMsgID string, startIndex int, packets []string, end bool, deadlineTime time.Time) error { | ||
| update := bson.M{ | ||
| "$set": bson.M{ | ||
| "end": end, | ||
| "deadline_time": deadlineTime, | ||
| }, | ||
| } | ||
| if len(packets) > 0 { | ||
| update["$push"] = bson.M{ | ||
| "packets": bson.M{ | ||
| "$each": packets, | ||
| "$position": startIndex, | ||
| }, | ||
| } | ||
| } | ||
| return mongoutil.UpdateOne(ctx, m.coll, bson.M{"client_msg_id": clientMsgID, "end": false}, update, true) | ||
| } | ||
|
|
||
| func (m *StreamMsgMongo) GetStreamMsg(ctx context.Context, clientMsgID string) (*model.StreamMsg, error) { | ||
| return mongoutil.FindOne[*model.StreamMsg](ctx, m.coll, bson.M{"client_msg_id": clientMsgID}) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -17,4 +17,5 @@ const ( | |
| UserName = "user" | ||
| SeqConversationName = "seq" | ||
| SeqUserName = "seq_user" | ||
| StreamMsgName = "stream_msg" | ||
| ) | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,13 @@ | ||
| package database | ||
|
|
||
| import ( | ||
| "context" | ||
| "github.com/openimsdk/open-im-server/v3/pkg/common/storage/model" | ||
| "time" | ||
| ) | ||
|
|
||
| type StreamMsg interface { | ||
| CreateStreamMsg(ctx context.Context, model *model.StreamMsg) error | ||
| AppendStreamMsg(ctx context.Context, clientMsgID string, startIndex int, packets []string, end bool, deadlineTime time.Time) error | ||
| GetStreamMsg(ctx context.Context, clientMsgID string) (*model.StreamMsg, error) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,21 @@ | ||
| package model | ||
|
|
||
| import ( | ||
| "time" | ||
| ) | ||
|
|
||
| const ( | ||
| StreamMsgStatusWait = 0 | ||
| StreamMsgStatusDone = 1 | ||
| StreamMsgStatusFail = 2 | ||
| ) | ||
|
|
||
| type StreamMsg struct { | ||
| ClientMsgID string `bson:"client_msg_id"` | ||
| ConversationID string `bson:"conversation_id"` | ||
| UserID string `bson:"user_id"` | ||
| Packets []string `bson:"packets"` | ||
| End bool `bson:"end"` | ||
| CreateTime time.Time `bson:"create_time"` | ||
| DeadlineTime time.Time `bson:"deadline_time"` | ||
| } |
Oops, something went wrong.