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

Merge remote-tracking branch 'origin/tuoyun' into tuoyun

Browse Source
This commit is contained in:
wenxu12345 2022-05-11 19:25:33 +08:00
commit 35a184827a
4 changed files with 22 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
config/config.yaml
View File

@ -49,7 +49,7 @@ kafka:
topic: "ms2ps_chat"
consumergroupid:
msgToMongo: mongo
MsgToMongoOffline: mongo_offline
msgToMongoOffline: mongo_offline
msgToMySql: mysql
msgToPush: push

3
internal/msg_transfer/logic/init.go
View File

@ -4,6 +4,7 @@ import (
"Open_IM/pkg/common/config"
"Open_IM/pkg/common/constant"
"Open_IM/pkg/common/kafka"
"Open_IM/pkg/common/log"
"Open_IM/pkg/statistics"
"fmt"
"sync"
@ -27,8 +28,10 @@ var (
func Init() {
cmdCh = make(chan Cmd2Value, 10000)
w = new(sync.Mutex)
persistentCH.Init()
historyCH.Init(cmdCh)
log.Debug("come msg transfer ts", config.Config.Kafka.ConsumerGroupID.MsgToMongoOffline, config.Config.Kafka.Ws2mschatOffline.Topic)
offlineHistoryCH.Init(cmdCh)
statistics.NewStatistics(&singleMsgSuccessCount, config.Config.ModuleName.MsgTransferName, fmt.Sprintf("%d second singleMsgCount insert to mongo", constant.StatisticsTimeInterval), constant.StatisticsTimeInterval)
statistics.NewStatistics(&groupMsgCount, config.Config.ModuleName.MsgTransferName, fmt.Sprintf("%d second groupMsgCount insert to mongo", constant.StatisticsTimeInterval), constant.StatisticsTimeInterval)

29
internal/msg_transfer/logic/offline_history_msg_handler.go
View File

@ -98,19 +98,24 @@ func (OfflineHistoryConsumerHandler) Setup(_ sarama.ConsumerGroupSession) error
func (OfflineHistoryConsumerHandler) Cleanup(_ sarama.ConsumerGroupSession) error { return nil }
func (mc *OfflineHistoryConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupSession,
claim sarama.ConsumerGroupClaim) error { // a instance in the consumer group
log.NewDebug("", "new session msg come", claim.HighWaterMarkOffset(), claim.Topic(), claim.Partition())
select {
case cmd := <-mc.cmdCh:
if cmd.Cmd == OnlineTopicVacancy {
for msg := range claim.Messages() {
if GetOnlineTopicStatus() == OnlineTopicVacancy {
log.NewDebug("", "kafka get info to mongo", "msgTopic", msg.Topic, "msgPartition", msg.Partition, "msg", string(msg.Value))
mc.msgHandle[msg.Topic](msg.Value, string(msg.Key))
sess.MarkMessage(msg, "")
}
}
}
log.NewDebug("", "new session msg come", claim.HighWaterMarkOffset(), claim.Topic(), claim.Partition())
for msg := range claim.Messages() {
log.NewDebug("", "kafka get info to delay mongo", "msgTopic", msg.Topic, "msgPartition", msg.Partition, "msg", "offline")
mc.msgHandle[msg.Topic](msg.Value, string(msg.Key))
}
//select {
//case cmd := <-mc.cmdCh:
// if cmd.Cmd == OnlineTopicVacancy {
// for msg := range claim.Messages() {
// if GetOnlineTopicStatus() == OnlineTopicVacancy {
// log.NewDebug("", "kafka get info to mongo", "msgTopic", msg.Topic, "msgPartition", msg.Partition, "msg", string(msg.Value))
// mc.msgHandle[msg.Topic](msg.Value, string(msg.Key))
// sess.MarkMessage(msg, "")
// }
// }
// }
//
//}
return nil
}

2
internal/msg_transfer/logic/online_history_msg_handler.go
View File

@ -136,7 +136,7 @@ func (och *OnlineHistoryConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupS
och.TriggerCmd(OnlineTopicBusy)
SetOnlineTopicStatus(OnlineTopicBusy)
for msg := range claim.Messages() {
log.NewDebug("", "kafka get info to mongo", "msgTopic", msg.Topic, "msgPartition", msg.Partition, "msg", string(msg.Value))
log.NewDebug("", "kafka get info to mongo", "msgTopic", msg.Topic, "msgPartition", msg.Partition, "online")
och.msgHandle[msg.Topic](msg.Value, string(msg.Key))
sess.MarkMessage(msg, "")
if claim.HighWaterMarkOffset()-msg.Offset <= 1 {