mirror of
https://github.com/openimsdk/open-im-server.git
synced 2026-05-10 20:15:59 +08:00
merge: update code from main to v3.8-js-sdk-only. (#2818)
* feat: implement merge milestone PR to target-branch. (#2796) * build: improve workflows logic. (#2801) * fix: improve time condition check mehtod. (#2804) * fix: improve time condition check mehtod. * fix * fix: webhook before online push (#2805) * fix: set own read seq in MongoDB when sender send a message. (#2808) * fix: solve err Notification when setGroupInfo. (#2806) * fix: solve err Notification when setGroupInfo. * build: update checkout version. * fix: update notification contents. * Introducing OpenIM Guru on Gurubase.io (#2788) * feat: support app update service (#2811) * 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: add ApplicationVersion * feat: add ApplicationVersion * feat: add ApplicationVersion --------- Co-authored-by: withchao <withchao@users.noreply.github.com> * feat: ApplicationVersion move chat (#2813) * 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: add ApplicationVersion * feat: ApplicationVersion move chat --------- Co-authored-by: withchao <withchao@users.noreply.github.com> * fix: improve condition check. (#2815) --------- Co-authored-by: Monet Lee <monet_lee@163.com> Co-authored-by: icey-yu <119291641+icey-yu@users.noreply.github.com> Co-authored-by: Kürşat Aktaş <kursat.ce@gmail.com> Co-authored-by: chao <48119764+withchao@users.noreply.github.com> Co-authored-by: withchao <withchao@users.noreply.github.com>
This commit is contained in:
OpenIM-Gordon
@@ -128,6 +128,7 @@ func (m *MsgTransfer) Start(index int, config *Config) error {
|
||||
|
||||
go m.historyCH.historyConsumerGroup.RegisterHandleAndConsumer(m.ctx, m.historyCH)
|
||||
go m.historyMongoCH.historyConsumerGroup.RegisterHandleAndConsumer(m.ctx, m.historyMongoCH)
|
||||
go m.historyCH.HandleUserHasReadSeqMessages(m.ctx)
|
||||
err := m.historyCH.redisMessageBatches.Start()
|
||||
if err != nil {
|
||||
return err
|
||||
@@ -157,12 +158,14 @@ func (m *MsgTransfer) Start(index int, config *Config) error {
|
||||
// graceful close kafka client.
|
||||
m.cancel()
|
||||
m.historyCH.redisMessageBatches.Close()
|
||||
m.historyCH.Close()
|
||||
m.historyCH.historyConsumerGroup.Close()
|
||||
m.historyMongoCH.historyConsumerGroup.Close()
|
||||
return nil
|
||||
case <-netDone:
|
||||
m.cancel()
|
||||
m.historyCH.redisMessageBatches.Close()
|
||||
m.historyCH.Close()
|
||||
m.historyCH.historyConsumerGroup.Close()
|
||||
m.historyMongoCH.historyConsumerGroup.Close()
|
||||
close(netDone)
|
||||
|
||||
@@ -18,8 +18,10 @@ import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/IBM/sarama"
|
||||
@@ -40,11 +42,12 @@ import (
|
||||
)
|
||||
|
||||
const (
|
||||
size = 500
|
||||
mainDataBuffer = 500
|
||||
subChanBuffer = 50
|
||||
worker = 50
|
||||
interval = 100 * time.Millisecond
|
||||
size = 500
|
||||
mainDataBuffer = 500
|
||||
subChanBuffer = 50
|
||||
worker = 50
|
||||
interval = 100 * time.Millisecond
|
||||
hasReadChanBuffer = 1000
|
||||
)
|
||||
|
||||
type ContextMsg struct {
|
||||
@@ -52,14 +55,23 @@ type ContextMsg struct {
|
||||
ctx context.Context
|
||||
}
|
||||
|
||||
// This structure is used for asynchronously writing the sender’s read sequence (seq) regarding a message into MongoDB.
|
||||
// For example, if the sender sends a message with a seq of 10, then their own read seq for this conversation should be set to 10.
|
||||
type userHasReadSeq struct {
|
||||
conversationID string
|
||||
userHasReadMap map[string]int64
|
||||
}
|
||||
|
||||
type OnlineHistoryRedisConsumerHandler struct {
|
||||
historyConsumerGroup *kafka.MConsumerGroup
|
||||
|
||||
redisMessageBatches *batcher.Batcher[sarama.ConsumerMessage]
|
||||
|
||||
msgTransferDatabase controller.MsgTransferDatabase
|
||||
conversationRpcClient *rpcclient.ConversationRpcClient
|
||||
groupRpcClient *rpcclient.GroupRpcClient
|
||||
msgTransferDatabase controller.MsgTransferDatabase
|
||||
conversationRpcClient *rpcclient.ConversationRpcClient
|
||||
groupRpcClient *rpcclient.GroupRpcClient
|
||||
conversationUserHasReadChan chan *userHasReadSeq
|
||||
wg sync.WaitGroup
|
||||
}
|
||||
|
||||
func NewOnlineHistoryRedisConsumerHandler(kafkaConf *config.Kafka, database controller.MsgTransferDatabase,
|
||||
@@ -70,6 +82,8 @@ func NewOnlineHistoryRedisConsumerHandler(kafkaConf *config.Kafka, database cont
|
||||
}
|
||||
var och OnlineHistoryRedisConsumerHandler
|
||||
och.msgTransferDatabase = database
|
||||
och.conversationUserHasReadChan = make(chan *userHasReadSeq, hasReadChanBuffer)
|
||||
och.wg.Add(1)
|
||||
|
||||
b := batcher.New[sarama.ConsumerMessage](
|
||||
batcher.WithSize(size),
|
||||
@@ -115,25 +129,25 @@ func (och *OnlineHistoryRedisConsumerHandler) do(ctx context.Context, channelID
|
||||
}
|
||||
|
||||
func (och *OnlineHistoryRedisConsumerHandler) doSetReadSeq(ctx context.Context, msgs []*ContextMsg) {
|
||||
type seqKey struct {
|
||||
conversationID string
|
||||
userID string
|
||||
}
|
||||
var readSeq map[seqKey]int64
|
||||
|
||||
var conversationID string
|
||||
var userSeqMap map[string]int64
|
||||
for _, msg := range msgs {
|
||||
if msg.message.ContentType != constant.HasReadReceipt {
|
||||
continue
|
||||
}
|
||||
var elem sdkws.NotificationElem
|
||||
if err := json.Unmarshal(msg.message.Content, &elem); err != nil {
|
||||
log.ZError(ctx, "handlerConversationRead Unmarshal NotificationElem msg err", err, "msg", msg)
|
||||
log.ZWarn(ctx, "handlerConversationRead Unmarshal NotificationElem msg err", err, "msg", msg)
|
||||
continue
|
||||
}
|
||||
var tips sdkws.MarkAsReadTips
|
||||
if err := json.Unmarshal([]byte(elem.Detail), &tips); err != nil {
|
||||
log.ZError(ctx, "handlerConversationRead Unmarshal MarkAsReadTips msg err", err, "msg", msg)
|
||||
log.ZWarn(ctx, "handlerConversationRead Unmarshal MarkAsReadTips msg err", err, "msg", msg)
|
||||
continue
|
||||
}
|
||||
//The conversation ID for each batch of messages processed by the batcher is the same.
|
||||
conversationID = tips.ConversationID
|
||||
if len(tips.Seqs) > 0 {
|
||||
for _, seq := range tips.Seqs {
|
||||
if tips.HasReadSeq < seq {
|
||||
@@ -146,26 +160,25 @@ func (och *OnlineHistoryRedisConsumerHandler) doSetReadSeq(ctx context.Context,
|
||||
if tips.HasReadSeq < 0 {
|
||||
continue
|
||||
}
|
||||
if readSeq == nil {
|
||||
readSeq = make(map[seqKey]int64)
|
||||
if userSeqMap == nil {
|
||||
userSeqMap = make(map[string]int64)
|
||||
}
|
||||
key := seqKey{
|
||||
conversationID: tips.ConversationID,
|
||||
userID: tips.MarkAsReadUserID,
|
||||
}
|
||||
if readSeq[key] > tips.HasReadSeq {
|
||||
|
||||
if userSeqMap[tips.MarkAsReadUserID] > tips.HasReadSeq {
|
||||
continue
|
||||
}
|
||||
readSeq[key] = tips.HasReadSeq
|
||||
userSeqMap[tips.MarkAsReadUserID] = tips.HasReadSeq
|
||||
}
|
||||
if readSeq == nil {
|
||||
if userSeqMap == nil {
|
||||
return
|
||||
}
|
||||
for key, seq := range readSeq {
|
||||
if err := och.msgTransferDatabase.SetHasReadSeqToDB(ctx, key.userID, key.conversationID, seq); err != nil {
|
||||
log.ZError(ctx, "set read seq to db error", err, "userID", key.userID, "conversationID", key.conversationID, "seq", seq)
|
||||
}
|
||||
if len(conversationID) == 0 {
|
||||
log.ZWarn(ctx, "conversation err", nil, "conversationID", conversationID)
|
||||
}
|
||||
if err := och.msgTransferDatabase.SetHasReadSeqToDB(ctx, conversationID, userSeqMap); err != nil {
|
||||
log.ZWarn(ctx, "set read seq to db error", err, "conversationID", conversationID, "userSeqMap", userSeqMap)
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
func (och *OnlineHistoryRedisConsumerHandler) parseConsumerMessages(ctx context.Context, consumerMessages []*sarama.ConsumerMessage) []*ContextMsg {
|
||||
@@ -250,12 +263,21 @@ func (och *OnlineHistoryRedisConsumerHandler) handleMsg(ctx context.Context, key
|
||||
}
|
||||
if len(storageMessageList) > 0 {
|
||||
msg := storageMessageList[0]
|
||||
lastSeq, isNewConversation, err := och.msgTransferDatabase.BatchInsertChat2Cache(ctx, conversationID, storageMessageList)
|
||||
lastSeq, isNewConversation, userSeqMap, err := och.msgTransferDatabase.BatchInsertChat2Cache(ctx, conversationID, storageMessageList)
|
||||
if err != nil && !errors.Is(errs.Unwrap(err), redis.Nil) {
|
||||
log.ZError(ctx, "batch data insert to redis err", err, "storageMsgList", storageMessageList)
|
||||
log.ZWarn(ctx, "batch data insert to redis err", err, "storageMsgList", storageMessageList)
|
||||
return
|
||||
}
|
||||
log.ZInfo(ctx, "BatchInsertChat2Cache end")
|
||||
err = och.msgTransferDatabase.SetHasReadSeqs(ctx, conversationID, userSeqMap)
|
||||
if err != nil {
|
||||
log.ZWarn(ctx, "SetHasReadSeqs error", err, "userSeqMap", userSeqMap, "conversationID", conversationID)
|
||||
prommetrics.SeqSetFailedCounter.Inc()
|
||||
}
|
||||
och.conversationUserHasReadChan <- &userHasReadSeq{
|
||||
conversationID: conversationID,
|
||||
userHasReadMap: userSeqMap,
|
||||
}
|
||||
|
||||
if isNewConversation {
|
||||
switch msg.SessionType {
|
||||
@@ -308,7 +330,7 @@ func (och *OnlineHistoryRedisConsumerHandler) handleNotification(ctx context.Con
|
||||
storageMessageList = append(storageMessageList, msg.message)
|
||||
}
|
||||
if len(storageMessageList) > 0 {
|
||||
lastSeq, _, err := och.msgTransferDatabase.BatchInsertChat2Cache(ctx, conversationID, storageMessageList)
|
||||
lastSeq, _, _, err := och.msgTransferDatabase.BatchInsertChat2Cache(ctx, conversationID, storageMessageList)
|
||||
if err != nil {
|
||||
log.ZError(ctx, "notification batch insert to redis error", err, "conversationID", conversationID,
|
||||
"storageList", storageMessageList)
|
||||
@@ -323,6 +345,21 @@ func (och *OnlineHistoryRedisConsumerHandler) handleNotification(ctx context.Con
|
||||
och.toPushTopic(ctx, key, conversationID, storageList)
|
||||
}
|
||||
}
|
||||
func (och *OnlineHistoryRedisConsumerHandler) HandleUserHasReadSeqMessages(ctx context.Context) {
|
||||
defer och.wg.Done()
|
||||
|
||||
for msg := range och.conversationUserHasReadChan {
|
||||
if err := och.msgTransferDatabase.SetHasReadSeqToDB(ctx, msg.conversationID, msg.userHasReadMap); err != nil {
|
||||
log.ZWarn(ctx, "set read seq to db error", err, "conversationID", msg.conversationID, "userSeqMap", msg.userHasReadMap)
|
||||
}
|
||||
}
|
||||
|
||||
log.ZInfo(ctx, "Channel closed, exiting handleUserHasReadSeqMessages")
|
||||
}
|
||||
func (och *OnlineHistoryRedisConsumerHandler) Close() {
|
||||
close(och.conversationUserHasReadChan)
|
||||
och.wg.Wait()
|
||||
}
|
||||
|
||||
func (och *OnlineHistoryRedisConsumerHandler) toPushTopic(ctx context.Context, key, conversationID string, msgs []*ContextMsg) {
|
||||
for _, v := range msgs {
|
||||
|
||||
Reference in New Issue
Block a user