Skip to content

Commit

Permalink
fix 细节
Browse files Browse the repository at this point in the history 
1、统一结构体方法 receiver，都用 pointer
2、使用 errors.Is 来做错误判断
3、修复单词拼写的错误
  • Loading branch information
hughyu committed Aug 16, 2024
1 parent 7110183 commit 65dfe56
Show file tree
Hide file tree
Showing 9 changed files with 35 additions and 33 deletions.
6 changes: 3 additions & 3 deletions internal/api/msg.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -49,14 +49,14 @@ func NewMessageApi(msgRpcClient *rpcclient.Message, userRpcClient *rpcclient.Use
userRpcClient: rpcclient.NewUserRpcClientByUser(userRpcClient), imAdminUserID: imAdminUserID}
}

func (MessageApi) SetOptions(options map[string]bool, value bool) {
func (*MessageApi) SetOptions(options map[string]bool, value bool) {
datautil.SetSwitchFromOptions(options, constant.IsHistory, value)
datautil.SetSwitchFromOptions(options, constant.IsPersistent, value)
datautil.SetSwitchFromOptions(options, constant.IsSenderSync, value)
datautil.SetSwitchFromOptions(options, constant.IsConversationUpdate, value)
}

func (m MessageApi) newUserSendMsgReq(_ *gin.Context, params *apistruct.SendMsg) *msg.SendMsgReq {
func (m *MessageApi) newUserSendMsgReq(_ *gin.Context, params *apistruct.SendMsg) *msg.SendMsgReq {
var newContent string
options := make(map[string]bool, 5)
switch params.ContentType {
Expand Down Expand Up @@ -231,7 +231,7 @@ func (m *MessageApi) SendMessage(c *gin.Context) {
}

// Set the status to successful if the message is sent.
var status int = constant.MsgSendSuccessed
var status = constant.MsgSendSuccessed

// Attempt to update the message sending status in the system.
_, err = m.Client.SetSendMsgStatus(c, &msg.SetSendMsgStatusReq{
Expand Down
3 changes: 2 additions & 1 deletion internal/msgtransfer/init.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@ package msgtransfer

import (
"context"
"errors"
"fmt"
"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/redis"
Expand Down Expand Up @@ -137,7 +138,7 @@ func (m *MsgTransfer) Start(index int, config *Config) error {
return
}

if err := prommetrics.TransferInit(prometheusPort); err != nil && err != http.ErrServerClosed {
if err := prommetrics.TransferInit(prometheusPort); err != nil && !errors.Is(err, http.ErrServerClosed) {
netErr = errs.WrapMsg(err, "prometheus start error", "prometheusPort", prometheusPort)
netDone <- struct{}{}
}
Expand Down
3 changes: 2 additions & 1 deletion internal/msgtransfer/online_history_msg_handler.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@ package msgtransfer

import (
"context"
"errors"
"github.com/IBM/sarama"
"github.com/go-redis/redis"
"github.com/openimsdk/open-im-server/v3/pkg/common/config"
Expand Down Expand Up @@ -187,7 +188,7 @@ func (och *OnlineHistoryRedisConsumerHandler) handleMsg(ctx context.Context, key
if len(storageMessageList) > 0 {
msg := storageMessageList[0]
lastSeq, isNewConversation, err := och.msgDatabase.BatchInsertChat2Cache(ctx, conversationID, storageMessageList)
if err != nil && errs.Unwrap(err) != redis.Nil {
if err != nil && !errors.Is(errs.Unwrap(err), redis.Nil) {
log.ZError(ctx, "batch data insert to redis err", err, "storageMsgList", storageMessageList)
return
}
Expand Down
6 changes: 3 additions & 3 deletions internal/msgtransfer/online_msg_to_mongo_handler.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -91,13 +91,13 @@ func (mc *OnlineHistoryMongoConsumerHandler) handleChatWs2Mongo(ctx context.Cont
}
}

func (OnlineHistoryMongoConsumerHandler) Setup(_ sarama.ConsumerGroupSession) error { return nil }
func (OnlineHistoryMongoConsumerHandler) Cleanup(_ sarama.ConsumerGroupSession) error { return nil }
func (*OnlineHistoryMongoConsumerHandler) Setup(_ sarama.ConsumerGroupSession) error { return nil }
func (*OnlineHistoryMongoConsumerHandler) Cleanup(_ sarama.ConsumerGroupSession) error { return nil }

func (mc *OnlineHistoryMongoConsumerHandler) ConsumeClaim(
sess sarama.ConsumerGroupSession,
claim sarama.ConsumerGroupClaim,
) error { // a instance in the consumer group
) error { // an instance in the consumer group
log.ZDebug(context.Background(), "online new session msg come", "highWaterMarkOffset",
claim.HighWaterMarkOffset(), "topic", claim.Topic(), "partition", claim.Partition())
for msg := range claim.Messages() {
Expand Down
16 changes: 8 additions & 8 deletions internal/push/onlinepusher.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -19,20 +19,20 @@ type OnlinePusher interface {
pushToUserIDs *[]string) []string
}

type emptyOnlinePUsher struct{}
type emptyOnlinePusher struct{}

func newEmptyOnlinePUsher() *emptyOnlinePUsher {
return &emptyOnlinePUsher{}
func newEmptyOnlinePusher() *emptyOnlinePusher {
return &emptyOnlinePusher{}
}

func (emptyOnlinePUsher) GetConnsAndOnlinePush(ctx context.Context, msg *sdkws.MsgData,
func (emptyOnlinePusher) GetConnsAndOnlinePush(ctx context.Context, msg *sdkws.MsgData,
pushToUserIDs []string) (wsResults []*msggateway.SingleMsgToUserResults, err error) {
log.ZWarn(ctx, "emptyOnlinePUsher GetConnsAndOnlinePush", nil)
log.ZWarn(ctx, "emptyOnlinePusher GetConnsAndOnlinePush", nil)
return nil, nil
}
func (u emptyOnlinePUsher) GetOnlinePushFailedUserIDs(ctx context.Context, msg *sdkws.MsgData,
func (u emptyOnlinePusher) GetOnlinePushFailedUserIDs(ctx context.Context, msg *sdkws.MsgData,
wsResults []*msggateway.SingleMsgToUserResults, pushToUserIDs *[]string) []string {
log.ZWarn(ctx, "emptyOnlinePUsher GetOnlinePushFailedUserIDs", nil)
log.ZWarn(ctx, "emptyOnlinePusher GetOnlinePushFailedUserIDs", nil)
return nil
}

Expand All @@ -45,7 +45,7 @@ func NewOnlinePusher(disCov discovery.SvcDiscoveryRegistry, config *Config) Onli
case "etcd":
return NewDefaultAllNode(disCov, config)
default:
return newEmptyOnlinePUsher()
return newEmptyOnlinePusher()
}
}

Expand Down
8 changes: 4 additions & 4 deletions internal/push/push_handler.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -154,17 +154,17 @@ func (c *ConsumerHandler) Push2User(ctx context.Context, userIDs []string, msg *
return nil
}
}
offlinePUshUserID := []string{msg.RecvID}
offlinePushUserID := []string{msg.RecvID}

//receiver offline push
if err = c.webhookBeforeOfflinePush(ctx, &c.config.WebhooksConfig.BeforeOfflinePush,
offlinePUshUserID, msg, nil); err != nil {
offlinePushUserID, msg, nil); err != nil {
return err
}

err = c.offlinePushMsg(ctx, msg, offlinePUshUserID)
err = c.offlinePushMsg(ctx, msg, offlinePushUserID)
if err != nil {
log.ZWarn(ctx, "offlinePushMsg failed", err, "offlinePUshUserID", offlinePUshUserID, "msg", msg)
log.ZWarn(ctx, "offlinePushMsg failed", err, "offlinePushUserID", offlinePushUserID, "msg", msg)
return nil
}

Expand Down
4 changes: 2 additions & 2 deletions pkg/common/storage/controller/msg.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -89,8 +89,8 @@ type CommonMsgDatabase interface {

// to mq
MsgToMQ(ctx context.Context, key string, msg2mq *sdkws.MsgData) error
MsgToPushMQ(ctx context.Context, key, conversarionID string, msg2mq *sdkws.MsgData) (int32, int64, error)
MsgToMongoMQ(ctx context.Context, key, conversarionID string, msgs []*sdkws.MsgData, lastSeq int64) error
MsgToPushMQ(ctx context.Context, key, conversationID string, msg2mq *sdkws.MsgData) (int32, int64, error)
MsgToMongoMQ(ctx context.Context, key, conversationID string, msgs []*sdkws.MsgData, lastSeq int64) error

RangeUserSendCount(ctx context.Context, start time.Time, end time.Time, group bool, ase bool, pageNumber int32, showNumber int32) (msgCount int64, userCount int64, users []*model.UserCount, dateCount map[string]int64, err error)
RangeGroupSendCount(ctx context.Context, start time.Time, end time.Time, ase bool, pageNumber int32, showNumber int32) (msgCount int64, userCount int64, groups []*model.GroupCount, dateCount map[string]int64, err error)
Expand Down
6 changes: 3 additions & 3 deletions pkg/common/storage/database/mgo/msg.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -117,9 +117,9 @@ func (m *MsgMgo) GetMsgBySeqIndexIn1Doc(ctx context.Context, userID, docID strin
}

func (m *MsgMgo) getMsgBySeqIndexIn1Doc(ctx context.Context, userID, docID string, seqs []int64) ([]*model.MsgInfoModel, error) {
indexs := make([]int64, 0, len(seqs))
indexes := make([]int64, 0, len(seqs))
for _, seq := range seqs {
indexs = append(indexs, m.model.GetMsgIndex(seq))
indexes = append(indexes, m.model.GetMsgIndex(seq))
}
pipeline := mongo.Pipeline{
bson.D{{Key: "$match", Value: bson.D{
Expand All @@ -130,7 +130,7 @@ func (m *MsgMgo) getMsgBySeqIndexIn1Doc(ctx context.Context, userID, docID strin
{Key: "doc_id", Value: 1},
{Key: "msgs", Value: bson.D{
{Key: "$map", Value: bson.D{
{Key: "input", Value: indexs},
{Key: "input", Value: indexes},
{Key: "as", Value: "index"},
{Key: "in", Value: bson.D{
{Key: "$arrayElemAt", Value: bson.A{"$msgs", "$$index"}},
Expand Down
16 changes: 8 additions & 8 deletions pkg/common/storage/model/msg.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -92,28 +92,28 @@ type GroupCount struct {
Count int64 `bson:"count"`
}

func (MsgDocModel) TableName() string {
func (*MsgDocModel) TableName() string {
return MsgTableName
}

func (MsgDocModel) GetSingleGocMsgNum() int64 {
func (*MsgDocModel) GetSingleGocMsgNum() int64 {
return singleGocMsgNum
}

func (MsgDocModel) GetSingleGocMsgNum5000() int64 {
func (*MsgDocModel) GetSingleGocMsgNum5000() int64 {
return singleGocMsgNum5000
}

func (m *MsgDocModel) IsFull() bool {
return m.Msg[len(m.Msg)-1].Msg != nil
}

func (m MsgDocModel) GetDocID(conversationID string, seq int64) string {
func (m *MsgDocModel) GetDocID(conversationID string, seq int64) string {
seqSuffix := (seq - 1) / singleGocMsgNum
return m.indexGen(conversationID, seqSuffix)
}

func (m MsgDocModel) GetDocIDSeqsMap(conversationID string, seqs []int64) map[string][]int64 {
func (m *MsgDocModel) GetDocIDSeqsMap(conversationID string, seqs []int64) map[string][]int64 {
t := make(map[string][]int64)
for i := 0; i < len(seqs); i++ {
docID := m.GetDocID(conversationID, seqs[i])
Expand All @@ -127,15 +127,15 @@ func (m MsgDocModel) GetDocIDSeqsMap(conversationID string, seqs []int64) map[st
return t
}

func (MsgDocModel) GetMsgIndex(seq int64) int64 {
func (*MsgDocModel) GetMsgIndex(seq int64) int64 {
return (seq - 1) % singleGocMsgNum
}

func (MsgDocModel) indexGen(conversationID string, seqSuffix int64) string {
func (*MsgDocModel) indexGen(conversationID string, seqSuffix int64) string {
return conversationID + ":" + strconv.FormatInt(seqSuffix, 10)
}

func (MsgDocModel) GenExceptionMessageBySeqs(seqs []int64) (exceptionMsg []*sdkws.MsgData) {
func (*MsgDocModel) GenExceptionMessageBySeqs(seqs []int64) (exceptionMsg []*sdkws.MsgData) {
for _, v := range seqs {
msgModel := new(sdkws.MsgData)
msgModel.Seq = v
Expand Down

0 comments on commit 65dfe56

Please sign in to comment.