blackteay/open-im-server
1
0
Fork 0
mirror of https://github.com/openimsdk/open-im-server.git synced 2025-12-02 18:34:29 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix: fix the unsolve error

Browse Source
This commit is contained in:
luhaoling 2024-02-18 18:19:13 +08:00
parent f6fdfee759
commit 455ac082ef
4 changed files with 24 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
internal/msggateway/init.go
View File

@ -49,6 +49,9 @@ func RunWsAndServer(rpcPort, wsPort, prometheusPort int) error {
if err != nil { if err != nil {
netDone <- err netDone <- err
} }
if err == nil {
fmt.Println("review SIGNAL but err is nil")
}
}() }()
return hubServer.LongConnServer.Run(netDone) return hubServer.LongConnServer.Run(netDone)
} }

8
internal/msgtransfer/init.go
View File

@ -19,6 +19,7 @@ import (
"errors" "errors"
"fmt" "fmt"
"github.com/OpenIMSDK/tools/errs" "github.com/OpenIMSDK/tools/errs"
"github.com/OpenIMSDK/tools/log"
util "github.com/openimsdk/open-im-server/v3/pkg/util/genutil" util "github.com/openimsdk/open-im-server/v3/pkg/util/genutil"
"github.com/OpenIMSDK/tools/mw" "github.com/OpenIMSDK/tools/mw"
@ -113,8 +114,11 @@ func (m *MsgTransfer) Start(prometheusPort int) error {
netErr error netErr error
) )
go m.historyCH.historyConsumerGroup.RegisterHandleAndConsumer(m.ctx, m.historyCH) onError := func(ctx context.Context, err error, errInfo string) {
go m.historyMongoCH.historyConsumerGroup.RegisterHandleAndConsumer(m.ctx, m.historyMongoCH) log.ZWarn(ctx, errInfo, err)
}
go m.historyCH.historyConsumerGroup.RegisterHandleAndConsumer(m.ctx, m.historyCH, onError)
go m.historyMongoCH.historyConsumerGroup.RegisterHandleAndConsumer(m.ctx, m.historyMongoCH, onError)
if config.Config.Prometheus.Enable { if config.Config.Prometheus.Enable {
go func() { go func() {

10
internal/push/consumer_init.go
View File

@ -14,7 +14,10 @@
package push package push
import "context" import (
"context"
"github.com/OpenIMSDK/tools/log"
)
type Consumer struct { type Consumer struct {
pushCh ConsumerHandler pushCh ConsumerHandler
@ -32,6 +35,9 @@ func NewConsumer(pusher *Pusher) (*Consumer, error) {
} }
func (c *Consumer) Start() { func (c *Consumer) Start() {
onError := func(ctx context.Context, err error, errInfo string) {
log.ZWarn(ctx, errInfo, err)
}
go c.pushCh.pushConsumerGroup.RegisterHandleAndConsumer(context.Background(), &c.pushCh, onError)
go c.pushCh.pushConsumerGroup.RegisterHandleAndConsumer(context.Background(), &c.pushCh)
} }

14
pkg/common/kafka/consumer_group.go
View File

@ -17,6 +17,7 @@ package kafka
import ( import (
"context" "context"
"errors" "errors"
"fmt"
"github.com/IBM/sarama" "github.com/IBM/sarama"
"github.com/OpenIMSDK/tools/errs" "github.com/OpenIMSDK/tools/errs"
"github.com/OpenIMSDK/tools/log" "github.com/OpenIMSDK/tools/log"
@ -64,18 +65,17 @@ func (mc *MConsumerGroup) GetContextFromMsg(cMsg *sarama.ConsumerMessage) contex
return GetContextWithMQHeader(cMsg.Headers) return GetContextWithMQHeader(cMsg.Headers)
} }
func (mc *MConsumerGroup) RegisterHandleAndConsumer(ctx context.Context, handler sarama.ConsumerGroupHandler) error { func (mc *MConsumerGroup) RegisterHandleAndConsumer(ctx context.Context, handler sarama.ConsumerGroupHandler, onError func(context.Context, error, string)) {
log.ZDebug(ctx, "register consumer group", "groupID", mc.groupID) log.ZDebug(ctx, "register consumer group", "groupID", mc.groupID)
for { for {
err := mc.ConsumerGroup.Consume(ctx, mc.topics, handler) err := mc.ConsumerGroup.Consume(ctx, mc.topics, handler)
if errors.Is(err, sarama.ErrClosedConsumerGroup) { if errors.Is(err, sarama.ErrClosedConsumerGroup) || errors.Is(err, context.Canceled) {
return nil return
}
if errors.Is(err, context.Canceled) {
return nil
} }
if err != nil { if err != nil {
log.ZWarn(ctx, "consume err", err, "topic", mc.topics, "groupID", mc.groupID) errInfo := fmt.Sprintf("consume err: %v, topic: %v, groupID: %s", err, strings.Join(mc.topics, ", "), mc.groupID)
onError(ctx, err, errInfo) // 调用回调函数处理错误
return
} }
} }
} }