msb-public
/
Open-IM-Server
mirror of https://github.com/OpenIMSDK/Open-IM-Server
2
0
Code Issues Projects Releases Wiki Activity

fix: pkg update.

Browse Source
pull/2148/head
Gordon 2 years ago
parent a097295894
commit ca89004086
20 changed files with 253 additions and 244 deletions
Show Stats Download Patch File Download Diff File

2
go.mod
Unescape View File

@ -16,7 +16,7 @@ require (
github.com/mitchellh/mapstructure v1.5.0 github.com/mitchellh/mapstructure v1.5.0
github.com/openimsdk/localcache v0.0.1 github.com/openimsdk/localcache v0.0.1
github.com/openimsdk/protocol v0.0.58-google github.com/openimsdk/protocol v0.0.58-google
github.com/openimsdk/tools v0.0.47-alpha.5 github.com/openimsdk/tools v0.0.47-alpha.8
github.com/pkg/errors v0.9.1 // indirect github.com/pkg/errors v0.9.1 // indirect
github.com/prometheus/client_golang v1.18.0 github.com/prometheus/client_golang v1.18.0
github.com/sirupsen/logrus v1.9.3 // indirect github.com/sirupsen/logrus v1.9.3 // indirect

6
internal/push/offlinepush/getui/push.go
Unescape View File

@ -19,6 +19,7 @@ import (
"crypto/sha256" "crypto/sha256"
"encoding/hex" "encoding/hex"
"errors" "errors"
"github.com/openimsdk/tools/utils/httputil"
"strconv" "strconv"
"sync" "sync"
"time" "time"
@ -26,7 +27,6 @@ import (
"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush" "github.com/openimsdk/open-im-server/v3/internal/push/offlinepush"
"github.com/openimsdk/open-im-server/v3/pkg/common/config" "github.com/openimsdk/open-im-server/v3/pkg/common/config"
"github.com/openimsdk/open-im-server/v3/pkg/common/db/cache" "github.com/openimsdk/open-im-server/v3/pkg/common/db/cache"
http2 "github.com/openimsdk/open-im-server/v3/pkg/common/http"
"github.com/openimsdk/tools/errs" "github.com/openimsdk/tools/errs"
"github.com/openimsdk/tools/log" "github.com/openimsdk/tools/log"
"github.com/openimsdk/tools/mcontext" "github.com/openimsdk/tools/mcontext"
@ -56,6 +56,7 @@ type Client struct {
tokenExpireTime int64 tokenExpireTime int64
taskIDTTL int64 taskIDTTL int64
pushConf *config.Push pushConf *config.Push
httpClient *httputil.HTTPClient
} }
func NewClient(pushConf *config.Push, cache cache.MsgModel) *Client { func NewClient(pushConf *config.Push, cache cache.MsgModel) *Client {
@ -63,6 +64,7 @@ func NewClient(pushConf *config.Push, cache cache.MsgModel) *Client {
tokenExpireTime: tokenExpireTime, tokenExpireTime: tokenExpireTime,
taskIDTTL: taskIDTTL, taskIDTTL: taskIDTTL,
pushConf: pushConf, pushConf: pushConf,
httpClient: httputil.NewHTTPClient(httputil.NewClientConfig()),
} }
} }
@ -174,7 +176,7 @@ func (g *Client) postReturn(
output RespI, output RespI,
timeout int, timeout int,
) error { ) error {
err := http2.PostReturn(ctx, url, header, input, output, timeout) err := g.httpClient.PostReturn(ctx, url, header, input, output, timeout)
if err != nil { if err != nil {
return err return err
} }

13
internal/rpc/auth/auth.go
Unescape View File

@ -16,6 +16,8 @@ package auth
import ( import (
"context" "context"
"github.com/openimsdk/open-im-server/v3/pkg/common/servererrs"
"github.com/openimsdk/tools/discovery"
"github.com/openimsdk/open-im-server/v3/pkg/authverify" "github.com/openimsdk/open-im-server/v3/pkg/authverify"
"github.com/openimsdk/open-im-server/v3/pkg/common/config" "github.com/openimsdk/open-im-server/v3/pkg/common/config"
@ -26,7 +28,6 @@ import (
pbauth "github.com/openimsdk/protocol/auth" pbauth "github.com/openimsdk/protocol/auth"
"github.com/openimsdk/protocol/constant" "github.com/openimsdk/protocol/constant"
"github.com/openimsdk/protocol/msggateway" "github.com/openimsdk/protocol/msggateway"
"github.com/openimsdk/tools/discoveryregistry"
"github.com/openimsdk/tools/errs" "github.com/openimsdk/tools/errs"
"github.com/openimsdk/tools/log" "github.com/openimsdk/tools/log"
"github.com/openimsdk/tools/mcontext" "github.com/openimsdk/tools/mcontext"
@ -37,11 +38,11 @@ import (
type authServer struct { type authServer struct {
authDatabase controller.AuthDatabase authDatabase controller.AuthDatabase
userRpcClient *rpcclient.UserRpcClient userRpcClient *rpcclient.UserRpcClient
RegisterCenter discoveryregistry.SvcDiscoveryRegistry RegisterCenter discovery.SvcDiscoveryRegistry
config *config.GlobalConfig config *config.GlobalConfig
} }
func Start(ctx context.Context, config *config.GlobalConfig, client discoveryregistry.SvcDiscoveryRegistry, server *grpc.Server) error { func Start(ctx context.Context, config *config.GlobalConfig, client discovery.SvcDiscoveryRegistry, server *grpc.Server) error {
rdb, err := cache.NewRedis(ctx, &config.Redis) rdb, err := cache.NewRedis(ctx, &config.Redis)
if err != nil { if err != nil {
return err return err
@ -109,19 +110,19 @@ func (s *authServer) parseToken(ctx context.Context, tokensString string) (claim
return nil, err return nil, err
} }
if len(m) == 0 { if len(m) == 0 {
return nil, errs.ErrTokenNotExist.Wrap() return nil, servererrs.ErrTokenNotExist.Wrap()
} }
if v, ok := m[tokensString]; ok { if v, ok := m[tokensString]; ok {
switch v { switch v {
case constant.NormalToken: case constant.NormalToken:
return claims, nil return claims, nil
case constant.KickedToken: case constant.KickedToken:
return nil, errs.ErrTokenKicked.Wrap() return nil, servererrs.ErrTokenKicked.Wrap()
default: default:
return nil, errs.Wrap(errs.ErrTokenUnknown) return nil, errs.Wrap(errs.ErrTokenUnknown)
} }
} }
return nil, errs.ErrTokenNotExist.Wrap() return nil, servererrs.ErrTokenNotExist.Wrap()
} }
func (s *authServer) ParseToken( func (s *authServer) ParseToken(

21
internal/rpc/conversation/conversaion.go
Unescape View File

@ -17,6 +17,9 @@ package conversation
import ( import (
"context" "context"
"errors" "errors"
"github.com/openimsdk/open-im-server/v3/pkg/common/servererrs"
"github.com/openimsdk/tools/db/mongoutil"
"github.com/openimsdk/tools/utils/datautil"
"sort" "sort"
"github.com/openimsdk/open-im-server/v3/pkg/common/config" "github.com/openimsdk/open-im-server/v3/pkg/common/config"
@ -31,11 +34,9 @@ import (
"github.com/openimsdk/protocol/constant" "github.com/openimsdk/protocol/constant"
pbconversation "github.com/openimsdk/protocol/conversation" pbconversation "github.com/openimsdk/protocol/conversation"
"github.com/openimsdk/protocol/sdkws" "github.com/openimsdk/protocol/sdkws"
"github.com/openimsdk/tools/discoveryregistry" "github.com/openimsdk/tools/discovery"
"github.com/openimsdk/tools/errs" "github.com/openimsdk/tools/errs"
"github.com/openimsdk/tools/log" "github.com/openimsdk/tools/log"
"github.com/openimsdk/tools/tx"
"github.com/openimsdk/tools/utils"
"google.golang.org/grpc" "google.golang.org/grpc"
) )
@ -47,7 +48,7 @@ type conversationServer struct {
conversationNotificationSender *notification.ConversationNotificationSender conversationNotificationSender *notification.ConversationNotificationSender
} }
func Start(ctx context.Context, config *config.GlobalConfig, client discoveryregistry.SvcDiscoveryRegistry, server *grpc.Server) error { func Start(ctx context.Context, config *config.GlobalConfig, client discovery.SvcDiscoveryRegistry, server *grpc.Server) error {
rdb, err := cache.NewRedis(ctx, &config.Redis) rdb, err := cache.NewRedis(ctx, &config.Redis)
if err != nil { if err != nil {
return err return err
@ -68,7 +69,7 @@ func Start(ctx context.Context, config *config.GlobalConfig, client discoveryreg
user: &userRpcClient, user: &userRpcClient,
conversationNotificationSender: notification.NewConversationNotificationSender(&config.Notification, &msgRpcClient), conversationNotificationSender: notification.NewConversationNotificationSender(&config.Notification, &msgRpcClient),
groupRpcClient: &groupRpcClient, groupRpcClient: &groupRpcClient,
conversationDatabase: controller.NewConversationDatabase(conversationDB, cache.NewConversationRedis(rdb, cache.GetDefaultOpt(), conversationDB), tx.NewMongo(mongo.GetClient())), conversationDatabase: controller.NewConversationDatabase(conversationDB, cache.NewConversationRedis(rdb, cache.GetDefaultOpt(), conversationDB), mongoutil.NewMongo(mongo.GetClient())),
}) })
return nil return nil
} }
@ -156,7 +157,7 @@ func (c *conversationServer) GetSortedConversationList(ctx context.Context, req
c.conversationSort(conversation_isPinTime, resp, conversation_unreadCount, conversationMsg) c.conversationSort(conversation_isPinTime, resp, conversation_unreadCount, conversationMsg)
c.conversationSort(conversation_notPinTime, resp, conversation_unreadCount, conversationMsg) c.conversationSort(conversation_notPinTime, resp, conversation_unreadCount, conversationMsg)
resp.ConversationElems = utils.Paginate(resp.ConversationElems, int(req.Pagination.GetPageNumber()), int(req.Pagination.GetShowNumber())) resp.ConversationElems = datautil.Paginate(resp.ConversationElems, int(req.Pagination.GetPageNumber()), int(req.Pagination.GetShowNumber()))
return resp, nil return resp, nil
} }
@ -182,7 +183,7 @@ func (c *conversationServer) GetConversations(ctx context.Context, req *pbconver
func (c *conversationServer) SetConversation(ctx context.Context, req *pbconversation.SetConversationReq) (*pbconversation.SetConversationResp, error) { func (c *conversationServer) SetConversation(ctx context.Context, req *pbconversation.SetConversationReq) (*pbconversation.SetConversationResp, error) {
var conversation tablerelation.ConversationModel var conversation tablerelation.ConversationModel
if err := utils.CopyStructFields(&conversation, req.Conversation); err != nil { if err := datautil.CopyStructFields(&conversation, req.Conversation); err != nil {
return nil, err return nil, err
} }
err := c.conversationDatabase.SetUserConversations(ctx, req.Conversation.OwnerUserID, []*tablerelation.ConversationModel{&conversation}) err := c.conversationDatabase.SetUserConversations(ctx, req.Conversation.OwnerUserID, []*tablerelation.ConversationModel{&conversation})
@ -205,7 +206,7 @@ func (c *conversationServer) SetConversations(ctx context.Context, req *pbconver
return nil, err return nil, err
} }
if groupInfo.Status == constant.GroupStatusDismissed { if groupInfo.Status == constant.GroupStatusDismissed {
return nil, errs.ErrDismissedAlready.WrapMsg("group dismissed") return nil, servererrs.ErrDismissedAlready.WrapMsg("group dismissed")
} }
} }
var unequal int var unequal int
@ -424,7 +425,7 @@ func (c *conversationServer) GetConversationOfflinePushUserIDs(ctx context.Conte
for _, userID := range userIDs { for _, userID := range userIDs {
delete(userIDSet, userID) delete(userIDSet, userID)
} }
return &pbconversation.GetConversationOfflinePushUserIDsResp{UserIDs: utils.Keys(userIDSet)}, nil return &pbconversation.GetConversationOfflinePushUserIDsResp{UserIDs: datautil.Keys(userIDSet)}, nil
} }
func (c *conversationServer) conversationSort(conversations map[int64]string, resp *pbconversation.GetSortedConversationListResp, conversation_unreadCount map[string]int64, conversationMsg map[string]*pbconversation.ConversationElem) { func (c *conversationServer) conversationSort(conversations map[int64]string, resp *pbconversation.GetSortedConversationListResp, conversation_unreadCount map[string]int64, conversationMsg map[string]*pbconversation.ConversationElem) {
@ -493,7 +494,7 @@ func (c *conversationServer) getConversationInfo(
for conversationID, chatLog := range chatLogs { for conversationID, chatLog := range chatLogs {
pbchatLog := &pbconversation.ConversationElem{} pbchatLog := &pbconversation.ConversationElem{}
msgInfo := &pbconversation.MsgInfo{} msgInfo := &pbconversation.MsgInfo{}
if err := utils.CopyStructFields(msgInfo, chatLog); err != nil { if err := datautil.CopyStructFields(msgInfo, chatLog); err != nil {
return nil, err return nil, err
} }
switch chatLog.SessionType { switch chatLog.SessionType {

4
internal/rpc/friend/callback.go
Unescape View File

@ -16,12 +16,12 @@ package friend
import ( import (
"context" "context"
"github.com/openimsdk/tools/utils/datautil"
cbapi "github.com/openimsdk/open-im-server/v3/pkg/callbackstruct" cbapi "github.com/openimsdk/open-im-server/v3/pkg/callbackstruct"
"github.com/openimsdk/open-im-server/v3/pkg/common/config" "github.com/openimsdk/open-im-server/v3/pkg/common/config"
"github.com/openimsdk/open-im-server/v3/pkg/common/http" "github.com/openimsdk/open-im-server/v3/pkg/common/http"
pbfriend "github.com/openimsdk/protocol/friend" pbfriend "github.com/openimsdk/protocol/friend"
"github.com/openimsdk/tools/utils"
) )
func CallbackBeforeAddFriend(ctx context.Context, callback *config.Callback, req *pbfriend.ApplyToAddFriendReq) error { func CallbackBeforeAddFriend(ctx context.Context, callback *config.Callback, req *pbfriend.ApplyToAddFriendReq) error {
@ -56,7 +56,7 @@ func CallbackBeforeSetFriendRemark(ctx context.Context, callback *config.Callbac
if err := http.CallBackPostReturn(ctx, callback.CallbackUrl, cbReq, resp, callback.CallbackBeforeAddFriend); err != nil { if err := http.CallBackPostReturn(ctx, callback.CallbackUrl, cbReq, resp, callback.CallbackBeforeAddFriend); err != nil {
return err return err
} }
utils.NotNilReplace(&req.Remark, &resp.Remark) datautil.NotNilReplace(&req.Remark, &resp.Remark)
return nil return nil
} }

43
internal/rpc/friend/friend.go
Unescape View File

@ -16,6 +16,9 @@ package friend
import ( import (
"context" "context"
"github.com/openimsdk/open-im-server/v3/pkg/common/servererrs"
"github.com/openimsdk/tools/db/mongoutil"
"github.com/openimsdk/tools/utils/datautil"
"github.com/openimsdk/open-im-server/v3/pkg/authverify" "github.com/openimsdk/open-im-server/v3/pkg/authverify"
"github.com/openimsdk/open-im-server/v3/pkg/common/config" "github.com/openimsdk/open-im-server/v3/pkg/common/config"
@ -30,10 +33,8 @@ import (
"github.com/openimsdk/protocol/constant" "github.com/openimsdk/protocol/constant"
pbfriend "github.com/openimsdk/protocol/friend" pbfriend "github.com/openimsdk/protocol/friend"
"github.com/openimsdk/protocol/sdkws" "github.com/openimsdk/protocol/sdkws"
registry "github.com/openimsdk/tools/discoveryregistry" "github.com/openimsdk/tools/discovery"
"github.com/openimsdk/tools/errs" "github.com/openimsdk/tools/errs"
"github.com/openimsdk/tools/tx"
"github.com/openimsdk/tools/utils"
"google.golang.org/grpc" "google.golang.org/grpc"
) )
@ -43,11 +44,11 @@ type friendServer struct {
userRpcClient *rpcclient.UserRpcClient userRpcClient *rpcclient.UserRpcClient
notificationSender *notification.FriendNotificationSender notificationSender *notification.FriendNotificationSender
conversationRpcClient rpcclient.ConversationRpcClient conversationRpcClient rpcclient.ConversationRpcClient
RegisterCenter registry.SvcDiscoveryRegistry RegisterCenter discovery.SvcDiscoveryRegistry
config *config.GlobalConfig config *config.GlobalConfig
} }
func Start(ctx context.Context, config *config.GlobalConfig, client registry.SvcDiscoveryRegistry, server *grpc.Server) error { func Start(ctx context.Context, config *config.GlobalConfig, client discovery.SvcDiscoveryRegistry, server *grpc.Server) error {
// Initialize MongoDB // Initialize MongoDB
mongo, err := unrelation.NewMongoDB(ctx, &config.Mongo) mongo, err := unrelation.NewMongoDB(ctx, &config.Mongo)
if err != nil { if err != nil {
@ -92,7 +93,7 @@ func Start(ctx context.Context, config *config.GlobalConfig, client registry.Svc
friendMongoDB, friendMongoDB,
friendRequestMongoDB, friendRequestMongoDB,
cache.NewFriendCacheRedis(rdb, friendMongoDB, cache.GetDefaultOpt()), cache.NewFriendCacheRedis(rdb, friendMongoDB, cache.GetDefaultOpt()),
tx.NewMongo(mongo.GetClient()), mongoutil.NewMongo(mongo.GetClient()),
), ),
blackDatabase: controller.NewBlackDatabase( blackDatabase: controller.NewBlackDatabase(
blackMongoDB, blackMongoDB,
@ -116,9 +117,9 @@ func (s *friendServer) ApplyToAddFriend(ctx context.Context, req *pbfriend.Apply
} }
if req.ToUserID == req.FromUserID { if req.ToUserID == req.FromUserID {
return nil, errs.ErrCanNotAddYourself.WrapMsg("req.ToUserID", req.ToUserID) return nil, servererrs.ErrCanNotAddYourself.WrapMsg("req.ToUserID", req.ToUserID)
} }
if err = CallbackBeforeAddFriend(ctx, &s.config.Callback, req); err != nil && err != errs.ErrCallbackContinue { if err = CallbackBeforeAddFriend(ctx, &s.config.Callback, req); err != nil && err != servererrs.ErrCallbackContinue {
return nil, err return nil, err
} }
@ -132,7 +133,7 @@ func (s *friendServer) ApplyToAddFriend(ctx context.Context, req *pbfriend.Apply
} }
if in1 && in2 { if in1 && in2 {
return nil, errs.ErrRelationshipAlready.WrapMsg("already friends has f") return nil, servererrs.ErrRelationshipAlready.WrapMsg("already friends has f")
} }
if err = s.friendDatabase.AddFriendRequest(ctx, req.FromUserID, req.ToUserID, req.ReqMsg, req.Ex); err != nil { if err = s.friendDatabase.AddFriendRequest(ctx, req.FromUserID, req.ToUserID, req.ReqMsg, req.Ex); err != nil {
@ -143,7 +144,7 @@ func (s *friendServer) ApplyToAddFriend(ctx context.Context, req *pbfriend.Apply
return nil, err return nil, err
} }
if err = CallbackAfterAddFriend(ctx, &s.config.Callback, req); err != nil && err != errs.ErrCallbackContinue { if err = CallbackAfterAddFriend(ctx, &s.config.Callback, req); err != nil && err != servererrs.ErrCallbackContinue {
return nil, err return nil, err
} }
return resp, nil return resp, nil
@ -157,10 +158,10 @@ func (s *friendServer) ImportFriends(ctx context.Context, req *pbfriend.ImportFr
if _, err := s.userRpcClient.GetUsersInfo(ctx, append([]string{req.OwnerUserID}, req.FriendUserIDs...)); err != nil { if _, err := s.userRpcClient.GetUsersInfo(ctx, append([]string{req.OwnerUserID}, req.FriendUserIDs...)); err != nil {
return nil, err return nil, err
} }
if utils.Contain(req.OwnerUserID, req.FriendUserIDs...) { if datautil.Contain(req.OwnerUserID, req.FriendUserIDs...) {
return nil, errs.ErrCanNotAddYourself.WrapMsg("can not add yourself") return nil, servererrs.ErrCanNotAddYourself.WrapMsg("can not add yourself")
} }
if utils.Duplicate(req.FriendUserIDs) { if datautil.Duplicate(req.FriendUserIDs) {
return nil, errs.ErrArgs.WrapMsg("friend userID repeated") return nil, errs.ErrArgs.WrapMsg("friend userID repeated")
} }
@ -198,7 +199,7 @@ func (s *friendServer) RespondFriendApply(ctx context.Context, req *pbfriend.Res
HandleResult: req.HandleResult, HandleResult: req.HandleResult,
} }
if req.HandleResult == constant.FriendResponseAgree { if req.HandleResult == constant.FriendResponseAgree {
if err := CallbackBeforeAddFriendAgree(ctx, &s.config.Callback, req); err != nil && err != errs.ErrCallbackContinue { if err := CallbackBeforeAddFriendAgree(ctx, &s.config.Callback, req); err != nil && err != servererrs.ErrCallbackContinue {
return nil, err return nil, err
} }
err := s.friendDatabase.AgreeFriendRequest(ctx, &friendRequest) err := s.friendDatabase.AgreeFriendRequest(ctx, &friendRequest)
@ -244,7 +245,7 @@ func (s *friendServer) DeleteFriend(ctx context.Context, req *pbfriend.DeleteFri
// ok. // ok.
func (s *friendServer) SetFriendRemark(ctx context.Context, req *pbfriend.SetFriendRemarkReq) (resp *pbfriend.SetFriendRemarkResp, err error) { func (s *friendServer) SetFriendRemark(ctx context.Context, req *pbfriend.SetFriendRemarkReq) (resp *pbfriend.SetFriendRemarkResp, err error) {
if err = CallbackBeforeSetFriendRemark(ctx, &s.config.Callback, req); err != nil && err != errs.ErrCallbackContinue { if err = CallbackBeforeSetFriendRemark(ctx, &s.config.Callback, req); err != nil && err != servererrs.ErrCallbackContinue {
return nil, err return nil, err
} }
resp = &pbfriend.SetFriendRemarkResp{} resp = &pbfriend.SetFriendRemarkResp{}
@ -258,7 +259,7 @@ func (s *friendServer) SetFriendRemark(ctx context.Context, req *pbfriend.SetFri
if err := s.friendDatabase.UpdateRemark(ctx, req.OwnerUserID, req.FriendUserID, req.Remark); err != nil { if err := s.friendDatabase.UpdateRemark(ctx, req.OwnerUserID, req.FriendUserID, req.Remark); err != nil {
return nil, err return nil, err
} }
if err := CallbackAfterSetFriendRemark(ctx, &s.config.Callback, req); err != nil && err != errs.ErrCallbackContinue { if err := CallbackAfterSetFriendRemark(ctx, &s.config.Callback, req); err != nil && err != servererrs.ErrCallbackContinue {
return nil, err return nil, err
} }
s.notificationSender.FriendRemarkSetNotification(ctx, req.OwnerUserID, req.FriendUserID) s.notificationSender.FriendRemarkSetNotification(ctx, req.OwnerUserID, req.FriendUserID)
@ -268,7 +269,7 @@ func (s *friendServer) SetFriendRemark(ctx context.Context, req *pbfriend.SetFri
// ok. // ok.
func (s *friendServer) GetDesignatedFriends(ctx context.Context, req *pbfriend.GetDesignatedFriendsReq) (resp *pbfriend.GetDesignatedFriendsResp, err error) { func (s *friendServer) GetDesignatedFriends(ctx context.Context, req *pbfriend.GetDesignatedFriendsReq) (resp *pbfriend.GetDesignatedFriendsResp, err error) {
resp = &pbfriend.GetDesignatedFriendsResp{} resp = &pbfriend.GetDesignatedFriendsResp{}
if utils.Duplicate(req.FriendUserIDs) { if datautil.Duplicate(req.FriendUserIDs) {
return nil, errs.ErrArgs.WrapMsg("friend userID repeated") return nil, errs.ErrArgs.WrapMsg("friend userID repeated")
} }
friends, err := s.friendDatabase.FindFriendsWithError(ctx, req.OwnerUserID, req.FriendUserIDs) friends, err := s.friendDatabase.FindFriendsWithError(ctx, req.OwnerUserID, req.FriendUserIDs)
@ -374,7 +375,7 @@ func (s *friendServer) GetSpecifiedFriendsInfo(ctx context.Context, req *pbfrien
if len(req.UserIDList) == 0 { if len(req.UserIDList) == 0 {
return nil, errs.ErrArgs.WrapMsg("userIDList is empty") return nil, errs.ErrArgs.WrapMsg("userIDList is empty")
} }
if utils.Duplicate(req.UserIDList) { if datautil.Duplicate(req.UserIDList) {
return nil, errs.ErrArgs.WrapMsg("userIDList repeated") return nil, errs.ErrArgs.WrapMsg("userIDList repeated")
} }
userMap, err := s.userRpcClient.GetUsersInfoMap(ctx, req.UserIDList) userMap, err := s.userRpcClient.GetUsersInfoMap(ctx, req.UserIDList)
@ -389,10 +390,10 @@ func (s *friendServer) GetSpecifiedFriendsInfo(ctx context.Context, req *pbfrien
if err != nil { if err != nil {
return nil, err return nil, err
} }
friendMap := utils.SliceToMap(friends, func(e *tablerelation.FriendModel) string { friendMap := datautil.SliceToMap(friends, func(e *tablerelation.FriendModel) string {
return e.FriendUserID return e.FriendUserID
}) })
blackMap := utils.SliceToMap(blacks, func(e *tablerelation.BlackModel) string { blackMap := datautil.SliceToMap(blacks, func(e *tablerelation.BlackModel) string {
return e.BlockUserID return e.BlockUserID
}) })
resp := &pbfriend.GetSpecifiedFriendsInfoResp{ resp := &pbfriend.GetSpecifiedFriendsInfoResp{
@ -441,7 +442,7 @@ func (s *friendServer) UpdateFriends(
if len(req.FriendUserIDs) == 0 { if len(req.FriendUserIDs) == 0 {
return nil, errs.ErrArgs.WrapMsg("friendIDList is empty") return nil, errs.ErrArgs.WrapMsg("friendIDList is empty")
} }
if utils.Duplicate(req.FriendUserIDs) { if datautil.Duplicate(req.FriendUserIDs) {
return nil, errs.ErrArgs.WrapMsg("friendIDList repeated") return nil, errs.ErrArgs.WrapMsg("friendIDList repeated")
} }

42
internal/rpc/group/callback.go
Unescape View File

@ -16,6 +16,7 @@ package group
import ( import (
"context" "context"
"github.com/openimsdk/tools/utils/datautil"
"time" "time"
"github.com/openimsdk/open-im-server/v3/pkg/apistruct" "github.com/openimsdk/open-im-server/v3/pkg/apistruct"
@ -28,7 +29,6 @@ import (
pbgroup "github.com/openimsdk/protocol/group" pbgroup "github.com/openimsdk/protocol/group"
"github.com/openimsdk/tools/log" "github.com/openimsdk/tools/log"
"github.com/openimsdk/tools/mcontext" "github.com/openimsdk/tools/mcontext"
"github.com/openimsdk/tools/utils"
"google.golang.org/protobuf/types/known/wrapperspb" "google.golang.org/protobuf/types/known/wrapperspb"
) )
@ -69,18 +69,18 @@ func CallbackBeforeCreateGroup(ctx context.Context, cfg *GroupEventCallbackConfi
return err return err
} }
utils.NotNilReplace(&req.GroupInfo.GroupID, resp.GroupID) datautil.NotNilReplace(&req.GroupInfo.GroupID, resp.GroupID)
utils.NotNilReplace(&req.GroupInfo.GroupName, resp.GroupName) datautil.NotNilReplace(&req.GroupInfo.GroupName, resp.GroupName)
utils.NotNilReplace(&req.GroupInfo.Notification, resp.Notification) datautil.NotNilReplace(&req.GroupInfo.Notification, resp.Notification)
utils.NotNilReplace(&req.GroupInfo.Introduction, resp.Introduction) datautil.NotNilReplace(&req.GroupInfo.Introduction, resp.Introduction)
utils.NotNilReplace(&req.GroupInfo.FaceURL, resp.FaceURL) datautil.NotNilReplace(&req.GroupInfo.FaceURL, resp.FaceURL)
utils.NotNilReplace(&req.GroupInfo.OwnerUserID, resp.OwnerUserID) datautil.NotNilReplace(&req.GroupInfo.OwnerUserID, resp.OwnerUserID)
utils.NotNilReplace(&req.GroupInfo.Ex, resp.Ex) datautil.NotNilReplace(&req.GroupInfo.Ex, resp.Ex)
utils.NotNilReplace(&req.GroupInfo.Status, resp.Status) datautil.NotNilReplace(&req.GroupInfo.Status, resp.Status)
utils.NotNilReplace(&req.GroupInfo.CreatorUserID, resp.CreatorUserID) datautil.NotNilReplace(&req.GroupInfo.CreatorUserID, resp.CreatorUserID)
utils.NotNilReplace(&req.GroupInfo.GroupType, resp.GroupType) datautil.NotNilReplace(&req.GroupInfo.GroupType, resp.GroupType)
utils.NotNilReplace(&req.GroupInfo.NeedVerification, resp.NeedVerification) datautil.NotNilReplace(&req.GroupInfo.NeedVerification, resp.NeedVerification)
utils.NotNilReplace(&req.GroupInfo.LookMemberInfo, resp.LookMemberInfo) datautil.NotNilReplace(&req.GroupInfo.LookMemberInfo, resp.LookMemberInfo)
return nil return nil
} }
@ -137,10 +137,10 @@ func CallbackBeforeMemberJoinGroup(ctx context.Context, cfg *GroupEventCallbackC
groupMember.MuteEndTime = time.UnixMilli(*resp.MuteEndTime) groupMember.MuteEndTime = time.UnixMilli(*resp.MuteEndTime)
} }
utils.NotNilReplace(&groupMember.FaceURL, resp.FaceURL) datautil.NotNilReplace(&groupMember.FaceURL, resp.FaceURL)
utils.NotNilReplace(&groupMember.Ex, resp.Ex) datautil.NotNilReplace(&groupMember.Ex, resp.Ex)
utils.NotNilReplace(&groupMember.Nickname, resp.Nickname) datautil.NotNilReplace(&groupMember.Nickname, resp.Nickname)
utils.NotNilReplace(&groupMember.RoleLevel, resp.RoleLevel) datautil.NotNilReplace(&groupMember.RoleLevel, resp.RoleLevel)
return nil return nil
} }
@ -394,10 +394,10 @@ func CallbackBeforeSetGroupInfo(ctx context.Context, cfg *GroupEventCallbackConf
if resp.ApplyMemberFriend != nil { if resp.ApplyMemberFriend != nil {
req.GroupInfoForSet.ApplyMemberFriend = wrapperspb.Int32(*resp.ApplyMemberFriend) req.GroupInfoForSet.ApplyMemberFriend = wrapperspb.Int32(*resp.ApplyMemberFriend)
} }
utils.NotNilReplace(&req.GroupInfoForSet.GroupID, &resp.GroupID) datautil.NotNilReplace(&req.GroupInfoForSet.GroupID, &resp.GroupID)
utils.NotNilReplace(&req.GroupInfoForSet.GroupName, &resp.GroupName) datautil.NotNilReplace(&req.GroupInfoForSet.GroupName, &resp.GroupName)
utils.NotNilReplace(&req.GroupInfoForSet.FaceURL, &resp.FaceURL) datautil.NotNilReplace(&req.GroupInfoForSet.FaceURL, &resp.FaceURL)
utils.NotNilReplace(&req.GroupInfoForSet.Introduction, &resp.Introduction) datautil.NotNilReplace(&req.GroupInfoForSet.Introduction, &resp.Introduction)
return nil return nil
} }

4
internal/rpc/msg/as_read.go
Unescape View File

@ -16,7 +16,7 @@ package msg
import ( import (
"context" "context"
"github.com/openimsdk/tools/utils/goassist" "github.com/openimsdk/tools/utils/datautil"
cbapi "github.com/openimsdk/open-im-server/v3/pkg/callbackstruct" cbapi "github.com/openimsdk/open-im-server/v3/pkg/callbackstruct"
"github.com/openimsdk/protocol/constant" "github.com/openimsdk/protocol/constant"
@ -157,7 +157,7 @@ func (m *msgServer) MarkConversationAsRead(ctx context.Context, req *msg.MarkCon
} }
// avoid client missed call MarkConversationMessageAsRead by order // avoid client missed call MarkConversationMessageAsRead by order
for _, val := range req.Seqs { for _, val := range req.Seqs {
if !goassist.Contain(val, seqs...) { if !datautil.Contain(val, seqs...) {
seqs = append(seqs, val) seqs = append(seqs, val)
} }
} }

34
internal/rpc/msg/callback.go
Unescape View File

@ -16,6 +16,7 @@ package msg
import ( import (
"context" "context"
"github.com/openimsdk/tools/utils/datautil"
cbapi "github.com/openimsdk/open-im-server/v3/pkg/callbackstruct" cbapi "github.com/openimsdk/open-im-server/v3/pkg/callbackstruct"
"github.com/openimsdk/open-im-server/v3/pkg/common/config" "github.com/openimsdk/open-im-server/v3/pkg/common/config"
@ -25,7 +26,6 @@ import (
"github.com/openimsdk/protocol/sdkws" "github.com/openimsdk/protocol/sdkws"
"github.com/openimsdk/tools/log" "github.com/openimsdk/tools/log"
"github.com/openimsdk/tools/mcontext" "github.com/openimsdk/tools/mcontext"
"github.com/openimsdk/tools/utils"
"google.golang.org/protobuf/proto" "google.golang.org/protobuf/proto"
) )
@ -136,22 +136,22 @@ func callbackMsgModify(ctx context.Context, globalConfig *config.GlobalConfig, m
if resp.Content != nil { if resp.Content != nil {
msg.MsgData.Content = []byte(*resp.Content) msg.MsgData.Content = []byte(*resp.Content)
} }
utils.NotNilReplace(msg.MsgData.OfflinePushInfo, resp.OfflinePushInfo) datautil.NotNilReplace(msg.MsgData.OfflinePushInfo, resp.OfflinePushInfo)
utils.NotNilReplace(&msg.MsgData.RecvID, resp.RecvID) datautil.NotNilReplace(&msg.MsgData.RecvID, resp.RecvID)
utils.NotNilReplace(&msg.MsgData.GroupID, resp.GroupID) datautil.NotNilReplace(&msg.MsgData.GroupID, resp.GroupID)
utils.NotNilReplace(&msg.MsgData.ClientMsgID, resp.ClientMsgID) datautil.NotNilReplace(&msg.MsgData.ClientMsgID, resp.ClientMsgID)
utils.NotNilReplace(&msg.MsgData.ServerMsgID, resp.ServerMsgID) datautil.NotNilReplace(&msg.MsgData.ServerMsgID, resp.ServerMsgID)
utils.NotNilReplace(&msg.MsgData.SenderPlatformID, resp.SenderPlatformID) datautil.NotNilReplace(&msg.MsgData.SenderPlatformID, resp.SenderPlatformID)
utils.NotNilReplace(&msg.MsgData.SenderNickname, resp.SenderNickname) datautil.NotNilReplace(&msg.MsgData.SenderNickname, resp.SenderNickname)
utils.NotNilReplace(&msg.MsgData.SenderFaceURL, resp.SenderFaceURL) datautil.NotNilReplace(&msg.MsgData.SenderFaceURL, resp.SenderFaceURL)
utils.NotNilReplace(&msg.MsgData.SessionType, resp.SessionType) datautil.NotNilReplace(&msg.MsgData.SessionType, resp.SessionType)
utils.NotNilReplace(&msg.MsgData.MsgFrom, resp.MsgFrom) datautil.NotNilReplace(&msg.MsgData.MsgFrom, resp.MsgFrom)
utils.NotNilReplace(&msg.MsgData.ContentType, resp.ContentType) datautil.NotNilReplace(&msg.MsgData.ContentType, resp.ContentType)
utils.NotNilReplace(&msg.MsgData.Status, resp.Status) datautil.NotNilReplace(&msg.MsgData.Status, resp.Status)
utils.NotNilReplace(&msg.MsgData.Options, resp.Options) datautil.NotNilReplace(&msg.MsgData.Options, resp.Options)
utils.NotNilReplace(&msg.MsgData.AtUserIDList, resp.AtUserIDList) datautil.NotNilReplace(&msg.MsgData.AtUserIDList, resp.AtUserIDList)
utils.NotNilReplace(&msg.MsgData.AttachedInfo, resp.AttachedInfo) datautil.NotNilReplace(&msg.MsgData.AttachedInfo, resp.AttachedInfo)
utils.NotNilReplace(&msg.MsgData.Ex, resp.Ex) datautil.NotNilReplace(&msg.MsgData.Ex, resp.Ex)
log.ZDebug(ctx, "callbackMsgModify", "msg", msg.MsgData) log.ZDebug(ctx, "callbackMsgModify", "msg", msg.MsgData)
return nil return nil
} }

4
internal/rpc/msg/delete.go
Unescape View File

@ -16,6 +16,7 @@ package msg
import ( import (
"context" "context"
"github.com/openimsdk/tools/utils/timeutil"
"github.com/openimsdk/open-im-server/v3/pkg/authverify" "github.com/openimsdk/open-im-server/v3/pkg/authverify"
"github.com/openimsdk/protocol/constant" "github.com/openimsdk/protocol/constant"
@ -23,7 +24,6 @@ import (
"github.com/openimsdk/protocol/msg" "github.com/openimsdk/protocol/msg"
"github.com/openimsdk/protocol/sdkws" "github.com/openimsdk/protocol/sdkws"
"github.com/openimsdk/tools/log" "github.com/openimsdk/tools/log"
"github.com/openimsdk/tools/utils"
) )
func (m *msgServer) getMinSeqs(maxSeqs map[string]int64) map[string]int64 { func (m *msgServer) getMinSeqs(maxSeqs map[string]int64) map[string]int64 {
@ -112,7 +112,7 @@ func (m *msgServer) DeleteMsgPhysical(ctx context.Context, req *msg.DeleteMsgPhy
if err := authverify.CheckAdmin(ctx, &m.config.Manager, &m.config.IMAdmin); err != nil { if err := authverify.CheckAdmin(ctx, &m.config.Manager, &m.config.IMAdmin); err != nil {
return nil, err return nil, err
} }
remainTime := utils.GetCurrentTimestampBySecond() - req.Timestamp remainTime := timeutil.GetCurrentTimestampBySecond() - req.Timestamp
for _, conversationID := range req.ConversationIDs { for _, conversationID := range req.ConversationIDs {
if err := m.MsgDatabase.DeleteConversationMsgsAndSetMinSeq(ctx, conversationID, remainTime); err != nil { if err := m.MsgDatabase.DeleteConversationMsgsAndSetMinSeq(ctx, conversationID, remainTime); err != nil {
log.ZWarn( log.ZWarn(

6
internal/rpc/msg/message_interceptor.go
Unescape View File

@ -16,12 +16,12 @@ package msg
import ( import (
"context" "context"
"github.com/openimsdk/open-im-server/v3/pkg/common/servererrs"
"github.com/openimsdk/open-im-server/v3/pkg/common/config" "github.com/openimsdk/open-im-server/v3/pkg/common/config"
"github.com/openimsdk/protocol/constant" "github.com/openimsdk/protocol/constant"
"github.com/openimsdk/protocol/msg" "github.com/openimsdk/protocol/msg"
"github.com/openimsdk/protocol/sdkws" "github.com/openimsdk/protocol/sdkws"
"github.com/openimsdk/tools/errs"
) )
type MessageInterceptorFunc func(ctx context.Context, globalConfig *config.GlobalConfig, req *msg.SendMsgReq) (*sdkws.MsgData, error) type MessageInterceptorFunc func(ctx context.Context, globalConfig *config.GlobalConfig, req *msg.SendMsgReq) (*sdkws.MsgData, error)
@ -30,12 +30,12 @@ func MessageHasReadEnabled(_ context.Context, globalConfig *config.GlobalConfig,
switch { switch {
case req.MsgData.ContentType == constant.HasReadReceipt && req.MsgData.SessionType == constant.SingleChatType: case req.MsgData.ContentType == constant.HasReadReceipt && req.MsgData.SessionType == constant.SingleChatType:
if !globalConfig.SingleMessageHasReadReceiptEnable { if !globalConfig.SingleMessageHasReadReceiptEnable {
return nil, errs.ErrMessageHasReadDisable.Wrap() return nil, servererrs.ErrMessageHasReadDisable.Wrap()
} }
return req.MsgData, nil return req.MsgData, nil
case req.MsgData.ContentType == constant.HasReadReceipt && req.MsgData.SessionType == constant.SuperGroupChatType: case req.MsgData.ContentType == constant.HasReadReceipt && req.MsgData.SessionType == constant.SuperGroupChatType:
if !globalConfig.GroupMessageHasReadReceiptEnable { if !globalConfig.GroupMessageHasReadReceiptEnable {
return nil, errs.ErrMessageHasReadDisable.Wrap() return nil, servererrs.ErrMessageHasReadDisable.Wrap()
} }
return req.MsgData, nil return req.MsgData, nil
} }

11
internal/rpc/msg/revoke.go
Unescape View File

@ -18,6 +18,8 @@ import (
"context" "context"
"encoding/json" "encoding/json"
"github.com/openimsdk/open-im-server/v3/pkg/common/db/table/relation" "github.com/openimsdk/open-im-server/v3/pkg/common/db/table/relation"
"github.com/openimsdk/open-im-server/v3/pkg/common/servererrs"
"github.com/openimsdk/tools/utils/datautil"
"time" "time"
"github.com/openimsdk/open-im-server/v3/pkg/authverify" "github.com/openimsdk/open-im-server/v3/pkg/authverify"
@ -27,7 +29,6 @@ import (
"github.com/openimsdk/tools/errs" "github.com/openimsdk/tools/errs"
"github.com/openimsdk/tools/log" "github.com/openimsdk/tools/log"
"github.com/openimsdk/tools/mcontext" "github.com/openimsdk/tools/mcontext"
"github.com/openimsdk/tools/utils"
) )
func (m *msgServer) RevokeMsg(ctx context.Context, req *msg.RevokeMsgReq) (*msg.RevokeMsgResp, error) { func (m *msgServer) RevokeMsg(ctx context.Context, req *msg.RevokeMsgReq) (*msg.RevokeMsgResp, error) {
@ -56,7 +57,7 @@ func (m *msgServer) RevokeMsg(ctx context.Context, req *msg.RevokeMsgReq) (*msg.
return nil, errs.ErrRecordNotFound.WrapMsg("msg not found") return nil, errs.ErrRecordNotFound.WrapMsg("msg not found")
} }
if msgs[0].ContentType == constant.MsgRevokeNotification { if msgs[0].ContentType == constant.MsgRevokeNotification {
return nil, errs.ErrMsgAlreadyRevoke.WrapMsg("msg already revoke") return nil, servererrs.ErrMsgAlreadyRevoke.WrapMsg("msg already revoke")
} }
data, _ := json.Marshal(msgs[0]) data, _ := json.Marshal(msgs[0])
@ -70,7 +71,7 @@ func (m *msgServer) RevokeMsg(ctx context.Context, req *msg.RevokeMsgReq) (*msg.
} }
role = user.AppMangerLevel role = user.AppMangerLevel
case constant.SuperGroupChatType: case constant.SuperGroupChatType:
members, err := m.GroupLocalCache.GetGroupMemberInfoMap(ctx, msgs[0].GroupID, utils.Distinct([]string{req.UserID, msgs[0].SendID})) members, err := m.GroupLocalCache.GetGroupMemberInfoMap(ctx, msgs[0].GroupID, datautil.Distinct([]string{req.UserID, msgs[0].SendID}))
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -105,10 +106,10 @@ func (m *msgServer) RevokeMsg(ctx context.Context, req *msg.RevokeMsgReq) (*msg.
revokerUserID := mcontext.GetOpUserID(ctx) revokerUserID := mcontext.GetOpUserID(ctx)
var flag bool var flag bool
if len(m.config.Manager.UserID) > 0 { if len(m.config.Manager.UserID) > 0 {
flag = utils.Contain(revokerUserID, m.config.Manager.UserID...) flag = datautil.Contain(revokerUserID, m.config.Manager.UserID...)
} }
if len(m.config.Manager.UserID) == 0 && len(m.config.IMAdmin.UserID) > 0 { if len(m.config.Manager.UserID) == 0 && len(m.config.IMAdmin.UserID) > 0 {
flag = utils.Contain(revokerUserID, m.config.IMAdmin.UserID...) flag = datautil.Contain(revokerUserID, m.config.IMAdmin.UserID...)
} }
tips := sdkws.RevokeMsgTips{ tips := sdkws.RevokeMsgTips{
RevokerUserID: revokerUserID, RevokerUserID: revokerUserID,

21
internal/rpc/msg/send.go
Unescape View File

@ -16,6 +16,10 @@ package msg
import ( import (
"context" "context"
"github.com/openimsdk/open-im-server/v3/pkg/common/servererrs"
"github.com/openimsdk/open-im-server/v3/pkg/util/conversationutil"
"github.com/openimsdk/tools/utils/datautil"
"github.com/openimsdk/tools/utils/stringutil"
"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics" "github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
"github.com/openimsdk/open-im-server/v3/pkg/msgprocessor" "github.com/openimsdk/open-im-server/v3/pkg/msgprocessor"
@ -26,7 +30,6 @@ import (
"github.com/openimsdk/tools/errs" "github.com/openimsdk/tools/errs"
"github.com/openimsdk/tools/log" "github.com/openimsdk/tools/log"
"github.com/openimsdk/tools/mcontext" "github.com/openimsdk/tools/mcontext"
"github.com/openimsdk/tools/utils"
"google.golang.org/protobuf/types/known/wrapperspb" "google.golang.org/protobuf/types/known/wrapperspb"
) )
@ -35,7 +38,7 @@ func (m *msgServer) SendMsg(ctx context.Context, req *pbmsg.SendMsgReq) (resp *p
if req.MsgData != nil { if req.MsgData != nil {
flag := isMessageHasReadEnabled(req.MsgData, m.config) flag := isMessageHasReadEnabled(req.MsgData, m.config)
if !flag { if !flag {
return nil, errs.ErrMessageHasReadDisable.Wrap() return nil, servererrs.ErrMessageHasReadDisable.Wrap()
} }
m.encapsulateMsgData(req.MsgData) m.encapsulateMsgData(req.MsgData)
switch req.MsgData.SessionType { switch req.MsgData.SessionType {
@ -64,7 +67,7 @@ func (m *msgServer) sendMsgSuperGroupChat(ctx context.Context, req *pbmsg.SendMs
if err := callbackMsgModify(ctx, m.config, req); err != nil { if err := callbackMsgModify(ctx, m.config, req); err != nil {
return nil, err return nil, err
} }
err = m.MsgDatabase.MsgToMQ(ctx, utils.GenConversationUniqueKeyForGroup(req.MsgData.GroupID), req.MsgData) err = m.MsgDatabase.MsgToMQ(ctx, conversationutil.GenConversationUniqueKeyForGroup(req.MsgData.GroupID), req.MsgData)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -91,14 +94,14 @@ func (m *msgServer) setConversationAtInfo(nctx context.Context, msg *sdkws.MsgDa
ConversationType: msg.SessionType, ConversationType: msg.SessionType,
GroupID: msg.GroupID, GroupID: msg.GroupID,
} }
tagAll := utils.Contain(constant.AtAllString, msg.AtUserIDList...) tagAll := datautil.Contain(constant.AtAllString, msg.AtUserIDList...)
if tagAll { if tagAll {
memberUserIDList, err := m.GroupLocalCache.GetGroupMemberIDs(ctx, msg.GroupID) memberUserIDList, err := m.GroupLocalCache.GetGroupMemberIDs(ctx, msg.GroupID)
if err != nil { if err != nil {
log.ZWarn(ctx, "GetGroupMemberIDs", err) log.ZWarn(ctx, "GetGroupMemberIDs", err)
return return
} }
atUserID = utils.DifferenceString([]string{constant.AtAllString}, msg.AtUserIDList) atUserID = stringutil.DifferenceString([]string{constant.AtAllString}, msg.AtUserIDList)
if len(atUserID) == 0 { // just @everyone if len(atUserID) == 0 { // just @everyone
conversation.GroupAtType = &wrapperspb.Int32Value{Value: constant.AtAll} conversation.GroupAtType = &wrapperspb.Int32Value{Value: constant.AtAll}
} else { // @Everyone and @other people } else { // @Everyone and @other people
@ -107,7 +110,7 @@ func (m *msgServer) setConversationAtInfo(nctx context.Context, msg *sdkws.MsgDa
if err != nil { if err != nil {
log.ZWarn(ctx, "SetConversations", err, "userID", atUserID, "conversation", conversation) log.ZWarn(ctx, "SetConversations", err, "userID", atUserID, "conversation", conversation)
} }
memberUserIDList = utils.DifferenceString(atUserID, memberUserIDList) memberUserIDList = stringutil.DifferenceString(atUserID, memberUserIDList)
} }
conversation.GroupAtType = &wrapperspb.Int32Value{Value: constant.AtAll} conversation.GroupAtType = &wrapperspb.Int32Value{Value: constant.AtAll}
err = m.Conversation.SetConversations(ctx, memberUserIDList, conversation) err = m.Conversation.SetConversations(ctx, memberUserIDList, conversation)
@ -123,7 +126,7 @@ func (m *msgServer) setConversationAtInfo(nctx context.Context, msg *sdkws.MsgDa
} }
func (m *msgServer) sendMsgNotification(ctx context.Context, req *pbmsg.SendMsgReq) (resp *pbmsg.SendMsgResp, err error) { func (m *msgServer) sendMsgNotification(ctx context.Context, req *pbmsg.SendMsgReq) (resp *pbmsg.SendMsgResp, err error) {
if err := m.MsgDatabase.MsgToMQ(ctx, utils.GenConversationUniqueKeyForSingle(req.MsgData.SendID, req.MsgData.RecvID), req.MsgData); err != nil { if err := m.MsgDatabase.MsgToMQ(ctx, conversationutil.GenConversationUniqueKeyForSingle(req.MsgData.SendID, req.MsgData.RecvID), req.MsgData); err != nil {
return nil, err return nil, err
} }
resp = &pbmsg.SendMsgResp{ resp = &pbmsg.SendMsgResp{
@ -145,7 +148,7 @@ func (m *msgServer) sendMsgSingleChat(ctx context.Context, req *pbmsg.SendMsgReq
isSend, err = m.modifyMessageByUserMessageReceiveOpt( isSend, err = m.modifyMessageByUserMessageReceiveOpt(
ctx, ctx,
req.MsgData.RecvID, req.MsgData.RecvID,
utils.GenConversationIDForSingle(req.MsgData.SendID, req.MsgData.RecvID), conversationutil.GenConversationIDForSingle(req.MsgData.SendID, req.MsgData.RecvID),
constant.SingleChatType, constant.SingleChatType,
req, req,
) )
@ -164,7 +167,7 @@ func (m *msgServer) sendMsgSingleChat(ctx context.Context, req *pbmsg.SendMsgReq
if err := callbackMsgModify(ctx, m.config, req); err != nil { if err := callbackMsgModify(ctx, m.config, req); err != nil {
return nil, err return nil, err
} }
if err := m.MsgDatabase.MsgToMQ(ctx, utils.GenConversationUniqueKeyForSingle(req.MsgData.SendID, req.MsgData.RecvID), req.MsgData); err != nil { if err := m.MsgDatabase.MsgToMQ(ctx, conversationutil.GenConversationUniqueKeyForSingle(req.MsgData.SendID, req.MsgData.RecvID), req.MsgData); err != nil {
prommetrics.SingleChatMsgProcessFailedCounter.Inc() prommetrics.SingleChatMsgProcessFailedCounter.Inc()
return nil, err return nil, err
} }

24
internal/rpc/msg/server.go
Unescape View File

@ -26,7 +26,7 @@ import (
"github.com/openimsdk/protocol/constant" "github.com/openimsdk/protocol/constant"
"github.com/openimsdk/protocol/conversation" "github.com/openimsdk/protocol/conversation"
"github.com/openimsdk/protocol/msg" "github.com/openimsdk/protocol/msg"
"github.com/openimsdk/tools/discoveryregistry" "github.com/openimsdk/tools/discovery"
"google.golang.org/grpc" "google.golang.org/grpc"
) )
@ -36,16 +36,16 @@ type (
// MsgServer encapsulates dependencies required for message handling. // MsgServer encapsulates dependencies required for message handling.
msgServer struct { msgServer struct {
RegisterCenter discoveryregistry.SvcDiscoveryRegistry // Service discovery registry for service registration. RegisterCenter discovery.SvcDiscoveryRegistry // Service discovery registry for service registration.
MsgDatabase controller.CommonMsgDatabase // Interface for message database operations. MsgDatabase controller.CommonMsgDatabase // Interface for message database operations.
Conversation *rpcclient.ConversationRpcClient // RPC client for conversation service. Conversation *rpcclient.ConversationRpcClient // RPC client for conversation service.
UserLocalCache *rpccache.UserLocalCache // Local cache for user data. UserLocalCache *rpccache.UserLocalCache // Local cache for user data.
FriendLocalCache *rpccache.FriendLocalCache // Local cache for friend data. FriendLocalCache *rpccache.FriendLocalCache // Local cache for friend data.
GroupLocalCache *rpccache.GroupLocalCache // Local cache for group data. GroupLocalCache *rpccache.GroupLocalCache // Local cache for group data.
ConversationLocalCache *rpccache.ConversationLocalCache // Local cache for conversation data. ConversationLocalCache *rpccache.ConversationLocalCache // Local cache for conversation data.
Handlers MessageInterceptorChain // Chain of handlers for processing messages. Handlers MessageInterceptorChain // Chain of handlers for processing messages.
notificationSender *rpcclient.NotificationSender // RPC client for sending notifications. notificationSender *rpcclient.NotificationSender // RPC client for sending notifications.
config *config.GlobalConfig // Global configuration settings. config *config.GlobalConfig // Global configuration settings.
} }
) )
@ -64,7 +64,7 @@ func (m *msgServer) addInterceptorHandler(interceptorFunc ...MessageInterceptorF
// return nil // return nil
//} //}
func Start(ctx context.Context, config *config.GlobalConfig, client discoveryregistry.SvcDiscoveryRegistry, server *grpc.Server) error { func Start(ctx context.Context, config *config.GlobalConfig, client discovery.SvcDiscoveryRegistry, server *grpc.Server) error {
rdb, err := cache.NewRedis(ctx, &config.Redis) rdb, err := cache.NewRedis(ctx, &config.Redis)
if err != nil { if err != nil {
return err return err

6
internal/rpc/msg/statistics.go
Unescape View File

@ -19,7 +19,7 @@ import (
"github.com/openimsdk/open-im-server/v3/pkg/common/db/table/relation" "github.com/openimsdk/open-im-server/v3/pkg/common/db/table/relation"
"github.com/openimsdk/protocol/msg" "github.com/openimsdk/protocol/msg"
"github.com/openimsdk/protocol/sdkws" "github.com/openimsdk/protocol/sdkws"
"github.com/openimsdk/tools/utils" "github.com/openimsdk/tools/utils/datautil"
"time" "time"
) )
@ -38,7 +38,7 @@ func (m *msgServer) GetActiveUser(ctx context.Context, req *msg.GetActiveUserReq
} }
var pbUsers []*msg.ActiveUser var pbUsers []*msg.ActiveUser
if len(users) > 0 { if len(users) > 0 {
userIDs := utils.Slice(users, func(e *relation.UserCount) string { return e.UserID }) userIDs := datautil.Slice(users, func(e *relation.UserCount) string { return e.UserID })
userMap, err := m.UserLocalCache.GetUsersInfoMap(ctx, userIDs) userMap, err := m.UserLocalCache.GetUsersInfoMap(ctx, userIDs)
if err != nil { if err != nil {
return nil, err return nil, err
@ -80,7 +80,7 @@ func (m *msgServer) GetActiveGroup(ctx context.Context, req *msg.GetActiveGroupR
} }
var pbgroups []*msg.ActiveGroup var pbgroups []*msg.ActiveGroup
if len(groups) > 0 { if len(groups) > 0 {
groupIDs := utils.Slice(groups, func(e *relation.GroupCount) string { return e.GroupID }) groupIDs := datautil.Slice(groups, func(e *relation.GroupCount) string { return e.GroupID })
resp, err := m.GroupLocalCache.GetGroupInfos(ctx, groupIDs) resp, err := m.GroupLocalCache.GetGroupInfos(ctx, groupIDs)
if err != nil { if err != nil {
return nil, err return nil, err

12
internal/rpc/msg/sync_msg.go
Unescape View File

@ -16,6 +16,9 @@ package msg
import ( import (
"context" "context"
"github.com/openimsdk/open-im-server/v3/pkg/util/conversationutil"
"github.com/openimsdk/tools/utils/datautil"
"github.com/openimsdk/tools/utils/timeutil"
"github.com/openimsdk/open-im-server/v3/pkg/authverify" "github.com/openimsdk/open-im-server/v3/pkg/authverify"
"github.com/openimsdk/open-im-server/v3/pkg/msgprocessor" "github.com/openimsdk/open-im-server/v3/pkg/msgprocessor"
@ -23,7 +26,6 @@ import (
"github.com/openimsdk/protocol/msg" "github.com/openimsdk/protocol/msg"
"github.com/openimsdk/protocol/sdkws" "github.com/openimsdk/protocol/sdkws"
"github.com/openimsdk/tools/log" "github.com/openimsdk/tools/log"
"github.com/openimsdk/tools/utils"
) )
func (m *msgServer) PullMessageBySeqs(ctx context.Context, req *sdkws.PullMessageBySeqsReq) (*sdkws.PullMessageBySeqsResp, error) { func (m *msgServer) PullMessageBySeqs(ctx context.Context, req *sdkws.PullMessageBySeqsReq) (*sdkws.PullMessageBySeqsResp, error) {
@ -93,9 +95,9 @@ func (m *msgServer) GetMaxSeq(ctx context.Context, req *sdkws.GetMaxSeqReq) (*sd
return nil, err return nil, err
} }
for _, conversationID := range conversationIDs { for _, conversationID := range conversationIDs {
conversationIDs = append(conversationIDs, utils.GetNotificationConversationIDByConversationID(conversationID)) conversationIDs = append(conversationIDs, conversationutil.GetNotificationConversationIDByConversationID(conversationID))
} }
conversationIDs = append(conversationIDs, utils.GetSelfNotificationConversationID(req.UserID)) conversationIDs = append(conversationIDs, conversationutil.GetSelfNotificationConversationID(req.UserID))
log.ZDebug(ctx, "GetMaxSeq", "conversationIDs", conversationIDs) log.ZDebug(ctx, "GetMaxSeq", "conversationIDs", conversationIDs)
maxSeqs, err := m.MsgDatabase.GetMaxSeqs(ctx, conversationIDs) maxSeqs, err := m.MsgDatabase.GetMaxSeqs(ctx, conversationIDs)
if err != nil { if err != nil {
@ -172,7 +174,7 @@ func (m *msgServer) SearchMessage(ctx context.Context, req *msg.SearchMessageReq
// Construct response with updated information // Construct response with updated information
for _, chatLog := range chatLogs { for _, chatLog := range chatLogs {
pbchatLog := &msg.ChatLog{} pbchatLog := &msg.ChatLog{}
utils.CopyStructFields(pbchatLog, chatLog) datautil.CopyStructFields(pbchatLog, chatLog)
pbchatLog.SendTime = chatLog.SendTime pbchatLog.SendTime = chatLog.SendTime
pbchatLog.CreateTime = chatLog.CreateTime pbchatLog.CreateTime = chatLog.CreateTime
if chatLog.SenderNickname == "" { if chatLog.SenderNickname == "" {
@ -197,5 +199,5 @@ func (m *msgServer) SearchMessage(ctx context.Context, req *msg.SearchMessageReq
} }
func (m *msgServer) GetServerTime(ctx context.Context, _ *msg.GetServerTimeReq) (*msg.GetServerTimeResp, error) { func (m *msgServer) GetServerTime(ctx context.Context, _ *msg.GetServerTimeReq) (*msg.GetServerTimeResp, error) {
return &msg.GetServerTimeResp{ServerTime: utils.GetCurrentTimestampByMill()}, nil return &msg.GetServerTimeResp{ServerTime: timeutil.GetCurrentTimestampByMill()}, nil
} }

4
internal/rpc/msg/utils.go
Unescape View File

@ -18,7 +18,7 @@ import (
"github.com/openimsdk/open-im-server/v3/pkg/common/config" "github.com/openimsdk/open-im-server/v3/pkg/common/config"
"github.com/openimsdk/protocol/constant" "github.com/openimsdk/protocol/constant"
"github.com/openimsdk/protocol/sdkws" "github.com/openimsdk/protocol/sdkws"
"github.com/openimsdk/tools/utils" "github.com/openimsdk/tools/errs"
"github.com/redis/go-redis/v9" "github.com/redis/go-redis/v9"
"go.mongodb.org/mongo-driver/mongo" "go.mongodb.org/mongo-driver/mongo"
) )
@ -42,7 +42,7 @@ func isMessageHasReadEnabled(msgData *sdkws.MsgData, config *config.GlobalConfig
} }
func IsNotFound(err error) bool { func IsNotFound(err error) bool {
switch utils.Unwrap(err) { switch errs.Unwrap(err) {
case redis.Nil, mongo.ErrNoDocuments: case redis.Nil, mongo.ErrNoDocuments:
return true return true
default: default:

66
internal/rpc/msg/verify.go
Unescape View File

@ -16,17 +16,19 @@ package msg
import ( import (
"context" "context"
"github.com/openimsdk/open-im-server/v3/pkg/common/servererrs"
"github.com/openimsdk/tools/utils/datautil"
"github.com/openimsdk/tools/utils/encrypt"
"github.com/openimsdk/tools/utils/timeutil"
"math/rand" "math/rand"
"strconv" "strconv"
"time" "time"
"github.com/openimsdk/open-im-server/v3/pkg/util/genutil"
"github.com/openimsdk/protocol/constant" "github.com/openimsdk/protocol/constant"
"github.com/openimsdk/protocol/msg" "github.com/openimsdk/protocol/msg"
"github.com/openimsdk/protocol/sdkws" "github.com/openimsdk/protocol/sdkws"
"github.com/openimsdk/tools/errs" "github.com/openimsdk/tools/errs"
"github.com/openimsdk/tools/log" "github.com/openimsdk/tools/log"
"github.com/openimsdk/tools/utils"
) )
var ExcludeContentType = []int{constant.HasReadReceipt} var ExcludeContentType = []int{constant.HasReadReceipt}
@ -51,10 +53,10 @@ type MessageRevoked struct {
func (m *msgServer) messageVerification(ctx context.Context, data *msg.SendMsgReq) error { func (m *msgServer) messageVerification(ctx context.Context, data *msg.SendMsgReq) error {
switch data.MsgData.SessionType { switch data.MsgData.SessionType {
case constant.SingleChatType: case constant.SingleChatType:
if len(m.config.Manager.UserID) > 0 && utils.IsContain(data.MsgData.SendID, m.config.Manager.UserID) { if len(m.config.Manager.UserID) > 0 && datautil.Contain(data.MsgData.SendID, m.config.Manager.UserID...) {
return nil return nil
} }
if utils.IsContain(data.MsgData.SendID, m.config.IMAdmin.UserID) { if datautil.Contain(data.MsgData.SendID, m.config.IMAdmin.UserID...) {
return nil return nil
} }
if data.MsgData.ContentType <= constant.NotificationEnd && if data.MsgData.ContentType <= constant.NotificationEnd &&
@ -66,7 +68,7 @@ func (m *msgServer) messageVerification(ctx context.Context, data *msg.SendMsgRe
return err return err
} }
if black { if black {
return errs.ErrBlockedByPeer.Wrap() return servererrs.ErrBlockedByPeer.Wrap()
} }
if m.config.MessageVerify.FriendVerify != nil && *m.config.MessageVerify.FriendVerify { if m.config.MessageVerify.FriendVerify != nil && *m.config.MessageVerify.FriendVerify {
friend, err := m.FriendLocalCache.IsFriend(ctx, data.MsgData.SendID, data.MsgData.RecvID) friend, err := m.FriendLocalCache.IsFriend(ctx, data.MsgData.SendID, data.MsgData.RecvID)
@ -74,7 +76,7 @@ func (m *msgServer) messageVerification(ctx context.Context, data *msg.SendMsgRe
return err return err
} }
if !friend { if !friend {
return errs.ErrNotPeersFriend.Wrap() return servererrs.ErrNotPeersFriend.Wrap()
} }
return nil return nil
} }
@ -86,15 +88,15 @@ func (m *msgServer) messageVerification(ctx context.Context, data *msg.SendMsgRe
} }
if groupInfo.Status == constant.GroupStatusDismissed && if groupInfo.Status == constant.GroupStatusDismissed &&
data.MsgData.ContentType != constant.GroupDismissedNotification { data.MsgData.ContentType != constant.GroupDismissedNotification {
return errs.ErrDismissedAlready.Wrap() return servererrs.ErrDismissedAlready.Wrap()
} }
if groupInfo.GroupType == constant.SuperGroup { if groupInfo.GroupType == constant.SuperGroup {
return nil return nil
} }
if len(m.config.Manager.UserID) > 0 && utils.IsContain(data.MsgData.SendID, m.config.Manager.UserID) { if len(m.config.Manager.UserID) > 0 && datautil.Contain(data.MsgData.SendID, m.config.Manager.UserID...) {
return nil return nil
} }
if utils.IsContain(data.MsgData.SendID, m.config.IMAdmin.UserID) { if datautil.Contain(data.MsgData.SendID, m.config.IMAdmin.UserID...) {
return nil return nil
} }
if data.MsgData.ContentType <= constant.NotificationEnd && if data.MsgData.ContentType <= constant.NotificationEnd &&
@ -106,13 +108,13 @@ func (m *msgServer) messageVerification(ctx context.Context, data *msg.SendMsgRe
return err return err
} }
if _, ok := memberIDs[data.MsgData.SendID]; !ok { if _, ok := memberIDs[data.MsgData.SendID]; !ok {
return errs.ErrNotInGroupYet.Wrap() return servererrs.ErrNotInGroupYet.Wrap()
} }
groupMemberInfo, err := m.GroupLocalCache.GetGroupMember(ctx, data.MsgData.GroupID, data.MsgData.SendID) groupMemberInfo, err := m.GroupLocalCache.GetGroupMember(ctx, data.MsgData.GroupID, data.MsgData.SendID)
if err != nil { if err != nil {
if errs.ErrRecordNotFound.Is(err) { if errs.ErrRecordNotFound.Is(err) {
return errs.ErrNotInGroupYet.WrapMsg(err.Error()) return servererrs.ErrNotInGroupYet.WrapMsg(err.Error())
} }
return err return err
} }
@ -120,10 +122,10 @@ func (m *msgServer) messageVerification(ctx context.Context, data *msg.SendMsgRe
return nil return nil
} else { } else {
if groupMemberInfo.MuteEndTime >= time.Now().UnixMilli() { if groupMemberInfo.MuteEndTime >= time.Now().UnixMilli() {
return errs.ErrMutedInGroup.Wrap() return servererrs.ErrMutedInGroup.Wrap()
} }
if groupInfo.Status == constant.GroupStatusMuted && groupMemberInfo.RoleLevel != constant.GroupAdmin { if groupInfo.Status == constant.GroupStatusMuted && groupMemberInfo.RoleLevel != constant.GroupAdmin {
return errs.ErrMutedGroup.Wrap() return servererrs.ErrMutedGroup.Wrap()
} }
} }
return nil return nil
@ -135,7 +137,7 @@ func (m *msgServer) messageVerification(ctx context.Context, data *msg.SendMsgRe
func (m *msgServer) encapsulateMsgData(msg *sdkws.MsgData) { func (m *msgServer) encapsulateMsgData(msg *sdkws.MsgData) {
msg.ServerMsgID = GetMsgID(msg.SendID) msg.ServerMsgID = GetMsgID(msg.SendID)
if msg.SendTime == 0 { if msg.SendTime == 0 {
msg.SendTime = utils.GetCurrentTimestampByMill() msg.SendTime = timeutil.GetCurrentTimestampByMill()
} }
switch msg.ContentType { switch msg.ContentType {
case constant.Text: case constant.Text:
@ -160,27 +162,27 @@ func (m *msgServer) encapsulateMsgData(msg *sdkws.MsgData) {
fallthrough fallthrough
case constant.Quote: case constant.Quote:
case constant.Revoke: case constant.Revoke:
utils.SetSwitchFromOptions(msg.Options, constant.IsUnreadCount, false) datautil.SetSwitchFromOptions(msg.Options, constant.IsUnreadCount, false)
utils.SetSwitchFromOptions(msg.Options, constant.IsOfflinePush, false) datautil.SetSwitchFromOptions(msg.Options, constant.IsOfflinePush, false)
case constant.HasReadReceipt: case constant.HasReadReceipt:
utils.SetSwitchFromOptions(msg.Options, constant.IsConversationUpdate, false) datautil.SetSwitchFromOptions(msg.Options, constant.IsConversationUpdate, false)
utils.SetSwitchFromOptions(msg.Options, constant.IsSenderConversationUpdate, false) datautil.SetSwitchFromOptions(msg.Options, constant.IsSenderConversationUpdate, false)
utils.SetSwitchFromOptions(msg.Options, constant.IsUnreadCount, false) datautil.SetSwitchFromOptions(msg.Options, constant.IsUnreadCount, false)
utils.SetSwitchFromOptions(msg.Options, constant.IsOfflinePush, false) datautil.SetSwitchFromOptions(msg.Options, constant.IsOfflinePush, false)
case constant.Typing: case constant.Typing:
utils.SetSwitchFromOptions(msg.Options, constant.IsHistory, false) datautil.SetSwitchFromOptions(msg.Options, constant.IsHistory, false)
utils.SetSwitchFromOptions(msg.Options, constant.IsPersistent, false) datautil.SetSwitchFromOptions(msg.Options, constant.IsPersistent, false)
utils.SetSwitchFromOptions(msg.Options, constant.IsSenderSync, false) datautil.SetSwitchFromOptions(msg.Options, constant.IsSenderSync, false)
utils.SetSwitchFromOptions(msg.Options, constant.IsConversationUpdate, false) datautil.SetSwitchFromOptions(msg.Options, constant.IsConversationUpdate, false)
utils.SetSwitchFromOptions(msg.Options, constant.IsSenderConversationUpdate, false) datautil.SetSwitchFromOptions(msg.Options, constant.IsSenderConversationUpdate, false)
utils.SetSwitchFromOptions(msg.Options, constant.IsUnreadCount, false) datautil.SetSwitchFromOptions(msg.Options, constant.IsUnreadCount, false)
utils.SetSwitchFromOptions(msg.Options, constant.IsOfflinePush, false) datautil.SetSwitchFromOptions(msg.Options, constant.IsOfflinePush, false)
} }
} }
func GetMsgID(sendID string) string { func GetMsgID(sendID string) string {
t := genutil.GetCurrentTimeFormatted() t := timeutil.GetCurrentTimeFormatted()
return utils.Md5(t + "-" + sendID + "-" + strconv.Itoa(rand.Int())) return encrypt.Md5(t + "-" + sendID + "-" + strconv.Itoa(rand.Int()))
} }
func (m *msgServer) modifyMessageByUserMessageReceiveOpt(ctx context.Context, userID, conversationID string, sessionType int, pb *msg.SendMsgReq) (bool, error) { func (m *msgServer) modifyMessageByUserMessageReceiveOpt(ctx context.Context, userID, conversationID string, sessionType int, pb *msg.SendMsgReq) (bool, error) {
@ -197,7 +199,7 @@ func (m *msgServer) modifyMessageByUserMessageReceiveOpt(ctx context.Context, us
if pb.MsgData.Options == nil { if pb.MsgData.Options == nil {
pb.MsgData.Options = make(map[string]bool, 10) pb.MsgData.Options = make(map[string]bool, 10)
} }
utils.SetSwitchFromOptions(pb.MsgData.Options, constant.IsOfflinePush, false) datautil.SetSwitchFromOptions(pb.MsgData.Options, constant.IsOfflinePush, false)
return true, nil return true, nil
} }
// conversationID := utils.GetConversationIDBySessionType(conversationID, sessionType) // conversationID := utils.GetConversationIDBySessionType(conversationID, sessionType)
@ -211,7 +213,7 @@ func (m *msgServer) modifyMessageByUserMessageReceiveOpt(ctx context.Context, us
case constant.ReceiveMessage: case constant.ReceiveMessage:
return true, nil return true, nil
case constant.NotReceiveMessage: case constant.NotReceiveMessage:
if utils.IsContainInt(int(pb.MsgData.ContentType), ExcludeContentType) { if datautil.Contain(int(pb.MsgData.ContentType), ExcludeContentType...) {
return true, nil return true, nil
} }
return false, nil return false, nil
@ -219,7 +221,7 @@ func (m *msgServer) modifyMessageByUserMessageReceiveOpt(ctx context.Context, us
if pb.MsgData.Options == nil { if pb.MsgData.Options == nil {
pb.MsgData.Options = make(map[string]bool, 10) pb.MsgData.Options = make(map[string]bool, 10)
} }
utils.SetSwitchFromOptions(pb.MsgData.Options, constant.IsOfflinePush, false) datautil.SetSwitchFromOptions(pb.MsgData.Options, constant.IsOfflinePush, false)
return true, nil return true, nil
} }
return true, nil return true, nil

17
pkg/common/http/http_client.go
Unescape View File

@ -17,20 +17,17 @@ package http
import ( import (
"context" "context"
"encoding/json" "encoding/json"
"net/http"
"time"
"github.com/openimsdk/open-im-server/v3/pkg/callbackstruct" "github.com/openimsdk/open-im-server/v3/pkg/callbackstruct"
"github.com/openimsdk/open-im-server/v3/pkg/common/config" "github.com/openimsdk/open-im-server/v3/pkg/common/config"
"github.com/openimsdk/tools/errs" "github.com/openimsdk/open-im-server/v3/pkg/common/servererrs"
"github.com/openimsdk/tools/log" "github.com/openimsdk/tools/log"
"github.com/openimsdk/tools/utils/httputil"
"net/http"
) )
var ( var (
// Define http client. // Define http client.
client = &http.Client{ client = httputil.NewHTTPClient(httputil.NewClientConfig())
Timeout: 15 * time.Second, // max timeout is 15s
}
) )
func init() { func init() {
@ -41,14 +38,14 @@ func init() {
func callBackPostReturn(ctx context.Context, url, command string, input interface{}, output callbackstruct.CallbackResp, callbackConfig config.CallBackConfig) error { func callBackPostReturn(ctx context.Context, url, command string, input interface{}, output callbackstruct.CallbackResp, callbackConfig config.CallBackConfig) error {
url = url + "/" + command url = url + "/" + command
log.ZInfo(ctx, "callback", "url", url, "input", input, "config", callbackConfig) log.ZInfo(ctx, "callback", "url", url, "input", input, "config", callbackConfig)
b, err := Post(ctx, url, nil, input, callbackConfig.CallbackTimeOut) b, err := client.Post(ctx, url, nil, input, callbackConfig.CallbackTimeOut)
if err != nil { if err != nil {
if callbackConfig.CallbackFailedContinue != nil && *callbackConfig.CallbackFailedContinue { if callbackConfig.CallbackFailedContinue != nil && *callbackConfig.CallbackFailedContinue {
log.ZInfo(ctx, "callback failed but continue", err, "url", url) log.ZInfo(ctx, "callback failed but continue", err, "url", url)
return nil return nil
} }
log.ZWarn(ctx, "callback network failed", err, "url", url, "input", input) log.ZWarn(ctx, "callback network failed", err, "url", url, "input", input)
return errs.ErrNetwork.WrapMsg(err.Error()) return servererrs.ErrNetwork.WrapMsg(err.Error())
} }
if err = json.Unmarshal(b, output); err != nil { if err = json.Unmarshal(b, output); err != nil {
if callbackConfig.CallbackFailedContinue != nil && *callbackConfig.CallbackFailedContinue { if callbackConfig.CallbackFailedContinue != nil && *callbackConfig.CallbackFailedContinue {
@ -56,7 +53,7 @@ func callBackPostReturn(ctx context.Context, url, command string, input interfac
return nil return nil
} }
log.ZWarn(ctx, "callback json unmarshal failed", err, "url", url, "input", input, "response", string(b)) log.ZWarn(ctx, "callback json unmarshal failed", err, "url", url, "input", input, "response", string(b))
return errs.ErrData.WithDetail(err.Error() + "response format error") return servererrs.ErrData.WithDetail(err.Error() + "response format error")
} }
if err := output.Parse(); err != nil { if err := output.Parse(); err != nil {
log.ZWarn(ctx, "callback parse failed", err, "url", url, "input", input, "response", string(b)) log.ZWarn(ctx, "callback parse failed", err, "url", url, "input", input, "response", string(b))

157
pkg/common/http/http_client_test.go
Unescape View File

@ -16,93 +16,92 @@ package http
import ( import (
"context" "context"
"reflect"
"testing" "testing"
"github.com/openimsdk/open-im-server/v3/pkg/callbackstruct" "github.com/openimsdk/open-im-server/v3/pkg/callbackstruct"
"github.com/openimsdk/open-im-server/v3/pkg/common/config" "github.com/openimsdk/open-im-server/v3/pkg/common/config"
) )
func TestGet(t *testing.T) { //func TestGet(t *testing.T) {
type args struct { // type args struct {
url string // url string
} // }
tests := []struct { // tests := []struct {
name string // name string
args args // args args
wantResponse []byte // wantResponse []byte
wantErr bool // wantErr bool
}{ // }{
// TODO: Add test cases. // // TODO: Add test cases.
} // }
for _, tt := range tests { // for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) { // t.Run(tt.name, func(t *testing.T) {
gotResponse, err := Get(tt.args.url) // gotResponse, err := Get(tt.args.url)
if (err != nil) != tt.wantErr { // if (err != nil) != tt.wantErr {
t.Errorf("Get() error = %v, wantErr %v", err, tt.wantErr) // t.Errorf("Get() error = %v, wantErr %v", err, tt.wantErr)
return // return
} // }
if !reflect.DeepEqual(gotResponse, tt.wantResponse) { // if !reflect.DeepEqual(gotResponse, tt.wantResponse) {
t.Errorf("Get() = %v, want %v", gotResponse, tt.wantResponse) // t.Errorf("Get() = %v, want %v", gotResponse, tt.wantResponse)
} // }
}) // })
} // }
} //}
func TestPost(t *testing.T) { //func TestPost(t *testing.T) {
type args struct { // type args struct {
ctx context.Context // ctx context.Context
url string // url string
header map[string]string // header map[string]string
data any // data any
timeout int // timeout int
} // }
tests := []struct { // tests := []struct {
name string // name string
args args // args args
wantContent []byte // wantContent []byte
wantErr bool // wantErr bool
}{ // }{
// TODO: Add test cases. // // TODO: Add test cases.
} // }
for _, tt := range tests { // for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) { // t.Run(tt.name, func(t *testing.T) {
gotContent, err := Post(tt.args.ctx, tt.args.url, tt.args.header, tt.args.data, tt.args.timeout) // gotContent, err := Post(tt.args.ctx, tt.args.url, tt.args.header, tt.args.data, tt.args.timeout)
if (err != nil) != tt.wantErr { // if (err != nil) != tt.wantErr {
t.Errorf("Post() error = %v, wantErr %v", err, tt.wantErr) // t.Errorf("Post() error = %v, wantErr %v", err, tt.wantErr)
return // return
} // }
if !reflect.DeepEqual(gotContent, tt.wantContent) { // if !reflect.DeepEqual(gotContent, tt.wantContent) {
t.Errorf("Post() = %v, want %v", gotContent, tt.wantContent) // t.Errorf("Post() = %v, want %v", gotContent, tt.wantContent)
} // }
}) // })
} // }
} //}
func TestPostReturn(t *testing.T) { //func TestPostReturn(t *testing.T) {
type args struct { // type args struct {
ctx context.Context // ctx context.Context
url string // url string
header map[string]string // header map[string]string
input any // input any
output any // output any
timeOutSecond int // timeOutSecond int
} // }
tests := []struct { // tests := []struct {
name string // name string
args args // args args
wantErr bool // wantErr bool
}{ // }{
// TODO: Add test cases. // // TODO: Add test cases.
} // }
for _, tt := range tests { // for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) { // t.Run(tt.name, func(t *testing.T) {
if err := PostReturn(tt.args.ctx, tt.args.url, tt.args.header, tt.args.input, tt.args.output, tt.args.timeOutSecond); (err != nil) != tt.wantErr { // if err := PostReturn(tt.args.ctx, tt.args.url, tt.args.header, tt.args.input, tt.args.output, tt.args.timeOutSecond); (err != nil) != tt.wantErr {
t.Errorf("PostReturn() error = %v, wantErr %v", err, tt.wantErr) // t.Errorf("PostReturn() error = %v, wantErr %v", err, tt.wantErr)
} // }
}) // })
} // }
} //}
func Test_callBackPostReturn(t *testing.T) { func Test_callBackPostReturn(t *testing.T) {
type args struct { type args struct {

Write Preview
Loading…
Cancel
Save