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

Merge branch 'errcode' of github.com:OpenIMSDK/Open-IM-Server into errcode

Browse Source
This commit is contained in:
wangchuxiao 2023-05-10 15:57:17 +08:00
commit 6035bd5683
3 changed files with 20 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
internal/msgtransfer/online_history_msg_handler.go
View File

@ -2,7 +2,11 @@ package msgtransfer
import ( import (
"context" "context"
"fmt"
"github.com/OpenIMSDK/Open-IM-Server/pkg/errs" "github.com/OpenIMSDK/Open-IM-Server/pkg/errs"
"runtime/debug"
"strconv"
"strings"
"sync" "sync"
"time" "time"
@ -155,6 +159,7 @@ func (och *OnlineHistoryRedisConsumerHandler) handleNotification(ctx context.Con
} }
func (och *OnlineHistoryRedisConsumerHandler) toPushTopic(ctx context.Context, conversationID string, msgs []*sdkws.MsgData) { func (och *OnlineHistoryRedisConsumerHandler) toPushTopic(ctx context.Context, conversationID string, msgs []*sdkws.MsgData) {
fmt.Printf("toPushTopic Stack:\n%s\n", debug.Stack())
for _, v := range msgs { for _, v := range msgs {
och.msgDatabase.MsgToPushMQ(ctx, conversationID, v) och.msgDatabase.MsgToPushMQ(ctx, conversationID, v)
} }
@ -219,6 +224,11 @@ func (och *OnlineHistoryRedisConsumerHandler) MessagesDistributionHandle() {
log.ZError(ctx, "msg_transfer Unmarshal msg err", err, string(consumerMessages[i].Value)) log.ZError(ctx, "msg_transfer Unmarshal msg err", err, string(consumerMessages[i].Value))
return return
} }
var arr []string
for i, header := range consumerMessages[i].Headers {
arr = append(arr, strconv.Itoa(i), string(header.Key), string(header.Value))
}
log.ZInfo(ctx, "consumer.kafka.GetContextWithMQHeader", "len", len(consumerMessages[i].Headers), "header", strings.Join(arr, ", "))
ctxMsg.ctx = kafka.GetContextWithMQHeader(consumerMessages[i].Headers) ctxMsg.ctx = kafka.GetContextWithMQHeader(consumerMessages[i].Headers)
ctxMsg.message = &msgFromMQ ctxMsg.message = &msgFromMQ
log.ZDebug(ctx, "single msg come to distribution center", msgFromMQ.String(), string(consumerMessages[i].Key)) log.ZDebug(ctx, "single msg come to distribution center", msgFromMQ.String(), string(consumerMessages[i].Key))

7
pkg/common/db/controller/black.go
View File

@ -2,10 +2,9 @@ package controller
import ( import (
"context" "context"
"errors"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/cache" "github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/cache"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/table/relation" "github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/table/relation"
"github.com/OpenIMSDK/Open-IM-Server/pkg/utils" "github.com/OpenIMSDK/Open-IM-Server/pkg/errs"
"gorm.io/gorm" "gorm.io/gorm"
) )
@ -49,14 +48,14 @@ func (b *blackDatabase) FindOwnerBlacks(ctx context.Context, ownerUserID string,
func (b *blackDatabase) CheckIn(ctx context.Context, userID1, userID2 string) (inUser1Blacks bool, inUser2Blacks bool, err error) { func (b *blackDatabase) CheckIn(ctx context.Context, userID1, userID2 string) (inUser1Blacks bool, inUser2Blacks bool, err error) {
_, err = b.black.Take(ctx, userID1, userID2) _, err = b.black.Take(ctx, userID1, userID2)
if err != nil { if err != nil {
if errors.Unwrap(err) != gorm.ErrRecordNotFound { if errs.Unwrap(err) != gorm.ErrRecordNotFound {
return return
} }
} }
inUser1Blacks = err == nil inUser1Blacks = err == nil
_, err = b.black.Take(ctx, userID2, userID1) _, err = b.black.Take(ctx, userID2, userID1)
if err != nil { if err != nil {
if utils.Unwrap(err) != gorm.ErrRecordNotFound { if errs.Unwrap(err) != gorm.ErrRecordNotFound {
return return
} }
} }

7
pkg/common/kafka/producer.go
View File

@ -3,6 +3,8 @@ package kafka
import ( import (
"context" "context"
"errors" "errors"
"strconv"
"strings"
"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"
@ -87,6 +89,11 @@ func (p *Producer) SendMessage(ctx context.Context, key string, m proto.Message)
if err != nil { if err != nil {
return 0, 0, utils.Wrap(err, "") return 0, 0, utils.Wrap(err, "")
} }
var arr []string
for i, header := range header {
arr = append(arr, strconv.Itoa(i), string(header.Key), string(header.Value))
}
log.ZInfo(ctx, "producer.kafka.GetContextWithMQHeader", "len", len(header), "header", strings.Join(arr, ", "))
kMsg.Headers = header kMsg.Headers = header
partition, offset, err := p.producer.SendMessage(kMsg) partition, offset, err := p.producer.SendMessage(kMsg)
log.ZDebug(ctx, "ByteEncoder SendMessage end", "key ", kMsg.Key, "key length", kMsg.Value.Length()) log.ZDebug(ctx, "ByteEncoder SendMessage end", "key ", kMsg.Key, "key length", kMsg.Value.Length())