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

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

Browse Source
This commit is contained in:
withchao 2023-04-18 14:48:17 +08:00
commit 5432ea0869
4 changed files with 12 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
internal/msgtransfer/online_history_msg_handler.go
View File

@ -2,6 +2,9 @@ package msgtransfer
import ( import (
"context" "context"
"sync"
"time"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/config" "github.com/OpenIMSDK/Open-IM-Server/pkg/common/config"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/constant" "github.com/OpenIMSDK/Open-IM-Server/pkg/common/constant"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/controller" "github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/controller"
@ -12,8 +15,6 @@ import (
"github.com/OpenIMSDK/Open-IM-Server/pkg/utils" "github.com/OpenIMSDK/Open-IM-Server/pkg/utils"
"github.com/Shopify/sarama" "github.com/Shopify/sarama"
"github.com/golang/protobuf/proto" "github.com/golang/protobuf/proto"
"sync"
"time"
) )
const ConsumerMsgs = 3 const ConsumerMsgs = 3
@ -200,7 +201,7 @@ func (och *OnlineHistoryRedisConsumerHandler) Cleanup(_ sarama.ConsumerGroupSess
func (och *OnlineHistoryRedisConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error { // a instance in the consumer group func (och *OnlineHistoryRedisConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error { // a instance in the consumer group
for { for {
if sess == nil { if sess == nil {
log.NewWarn("", " sess == nil, waiting ") log.ZWarn(context.Background(), "sess == nil, waiting", nil)
time.Sleep(100 * time.Millisecond) time.Sleep(100 * time.Millisecond)
} else { } else {
break break

9
pkg/common/db/controller/msg.go
View File

@ -2,6 +2,9 @@ package controller
import ( import (
"fmt" "fmt"
"sync"
"time"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/config" "github.com/OpenIMSDK/Open-IM-Server/pkg/common/config"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/constant" "github.com/OpenIMSDK/Open-IM-Server/pkg/common/constant"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/cache" "github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/cache"
@ -12,11 +15,10 @@ import (
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/mcontext" "github.com/OpenIMSDK/Open-IM-Server/pkg/common/mcontext"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/prome" "github.com/OpenIMSDK/Open-IM-Server/pkg/common/prome"
"github.com/gogo/protobuf/sortkeys" "github.com/gogo/protobuf/sortkeys"
"sync"
"time"
"context" "context"
"errors" "errors"
pbMsg "github.com/OpenIMSDK/Open-IM-Server/pkg/proto/msg" pbMsg "github.com/OpenIMSDK/Open-IM-Server/pkg/proto/msg"
"github.com/OpenIMSDK/Open-IM-Server/pkg/proto/sdkws" "github.com/OpenIMSDK/Open-IM-Server/pkg/proto/sdkws"
"github.com/OpenIMSDK/Open-IM-Server/pkg/utils" "github.com/OpenIMSDK/Open-IM-Server/pkg/utils"
@ -617,7 +619,7 @@ func (db *msgDatabase) deleteMsgRecursion(ctx context.Context, sourceID string,
} }
//log.NewDebug(operationID, "ID:", sourceID, "index:", index, "uid:", msgs.UID, "len:", len(msgs.Msg)) //log.NewDebug(operationID, "ID:", sourceID, "index:", index, "uid:", msgs.UID, "len:", len(msgs.Msg))
if int64(len(msgs.Msg)) > db.msg.GetSingleGocMsgNum() { if int64(len(msgs.Msg)) > db.msg.GetSingleGocMsgNum() {
log.NewWarn(mcontext.GetOperationID(ctx), utils.GetSelfFuncName(), "msgs too large:", len(msgs.Msg), "docID:", msgs.DocID) log.ZWarn(ctx, "msgs too large", nil, "lenth", len(msgs.Msg), "docID:", msgs.DocID)
} }
if msgs.Msg[len(msgs.Msg)-1].SendTime+(remainTime*1000) < utils.GetCurrentTimestampByMill() && msgs.IsFull() { if msgs.Msg[len(msgs.Msg)-1].SendTime+(remainTime*1000) < utils.GetCurrentTimestampByMill() && msgs.IsFull() {
delStruct.delDocIDs = append(delStruct.delDocIDs, msgs.DocID) delStruct.delDocIDs = append(delStruct.delDocIDs, msgs.DocID)
@ -657,7 +659,6 @@ func (db *msgDatabase) deleteMsgRecursion(ctx context.Context, sourceID string,
} }
} }
} }
//log.NewDebug(operationID, sourceID, "continue to", delStruct)
// 继续递归 index+1 // 继续递归 index+1
seq, err := db.deleteMsgRecursion(ctx, sourceID, index+1, delStruct, remainTime) seq, err := db.deleteMsgRecursion(ctx, sourceID, index+1, delStruct, remainTime)
return seq, utils.Wrap(err, "deleteMsg failed") return seq, utils.Wrap(err, "deleteMsg failed")

1
pkg/common/log/logrus.go
View File

@ -3,6 +3,7 @@ package log
import ( import (
"bufio" "bufio"
"context" "context"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/config" "github.com/OpenIMSDK/Open-IM-Server/pkg/common/config"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/mcontext" "github.com/OpenIMSDK/Open-IM-Server/pkg/common/mcontext"

4
pkg/common/log/zap.go
View File

@ -95,14 +95,14 @@ func NewZapLogger(loggerName string, logLevel int, isStdout bool, isJson bool, l
} }
func (l *ZapLogger) cores(isStdout bool, isJson bool, logLocation string, rotateCount uint) (zap.Option, error) { func (l *ZapLogger) cores(isStdout bool, isJson bool, logLocation string, rotateCount uint) (zap.Option, error) {
c := zap.NewDevelopmentEncoderConfig() c := zap.NewProductionEncoderConfig()
c.EncodeTime = l.timeEncoder c.EncodeTime = l.timeEncoder
c.EncodeDuration = zapcore.SecondsDurationEncoder c.EncodeDuration = zapcore.SecondsDurationEncoder
c.MessageKey = "msg" c.MessageKey = "msg"
c.LevelKey = "level" c.LevelKey = "level"
c.TimeKey = "time" c.TimeKey = "time"
c.CallerKey = "caller" c.CallerKey = "caller"
c.NameKey = "logger"
var fileEncoder zapcore.Encoder var fileEncoder zapcore.Encoder
if isJson { if isJson {
c.EncodeLevel = zapcore.CapitalLevelEncoder c.EncodeLevel = zapcore.CapitalLevelEncoder