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

Merge pull request #3359 from withchao/pre-release-v3.8.4

Browse Source 
fix: standalone mode cannot be used
pull/3369/head
chao 4 months ago committed by GitHub
parent 3526672316 9616e920ed
commit ab32167b19
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
10 changed files with 126 additions and 106 deletions
Show Stats Download Patch File Download Diff File

2
go.mod
Unescape View File

@ -13,7 +13,7 @@ require (
github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0
github.com/mitchellh/mapstructure v1.5.0
github.com/openimsdk/protocol v0.0.73-alpha.6
github.com/openimsdk/tools v0.0.50-alpha.81
github.com/openimsdk/tools v0.0.50-alpha.83
github.com/pkg/errors v0.9.1 // indirect
github.com/prometheus/client_golang v1.18.0
github.com/stretchr/testify v1.9.0

4
go.sum
Unescape View File

@ -349,8 +349,8 @@ github.com/openimsdk/gomake v0.0.15-alpha.5 h1:eEZCEHm+NsmcO3onXZPIUbGFCYPYbsX5b
github.com/openimsdk/gomake v0.0.15-alpha.5/go.mod h1:PndCozNc2IsQIciyn9mvEblYWZwJmAI+06z94EY+csI=
github.com/openimsdk/protocol v0.0.73-alpha.6 h1:sna9coWG7HN1zObBPtvG0Ki/vzqHXiB4qKbA5P3w7kc=
github.com/openimsdk/protocol v0.0.73-alpha.6/go.mod h1:WF7EuE55vQvpyUAzDXcqg+B+446xQyEba0X35lTINmw=
github.com/openimsdk/tools v0.0.50-alpha.81 h1:VbuJKtigNXLkCKB/Q6f2UHsqoSaTOAwS8F51c1nhOCA=
github.com/openimsdk/tools v0.0.50-alpha.81/go.mod h1:n2poR3asX1e1XZce4O+MOWAp+X02QJRFvhcLCXZdzRo=
github.com/openimsdk/tools v0.0.50-alpha.83 h1:7c1D40YGqIWUmGfCII5pduETGC/8c2DyS9SQ4LvoplU=
github.com/openimsdk/tools v0.0.50-alpha.83/go.mod h1:n2poR3asX1e1XZce4O+MOWAp+X02QJRFvhcLCXZdzRo=
github.com/pelletier/go-toml/v2 v2.2.2 h1:aYUidT7k73Pcl9nb2gScu7NSrKCSHIDE89b3+6Wq+LM=
github.com/pelletier/go-toml/v2 v2.2.2/go.mod h1:1t835xjRzz80PqgE6HHgN2JOsmgYu/h4qDAS4n929Rs=
github.com/pierrec/lz4/v4 v4.1.21 h1:yOVMLb6qSIDP67pl/5F7RepeKYu/VmTyEXvuMI5d9mQ=

9
internal/api/router.go
Unescape View File

@ -9,11 +9,8 @@ import (
"github.com/gin-gonic/gin"
"github.com/gin-gonic/gin/binding"
"github.com/go-playground/validator/v10"
"github.com/openimsdk/open-im-server/v3/pkg/authverify"
"github.com/openimsdk/tools/mcontext"
"github.com/openimsdk/tools/utils/datautil"
"github.com/openimsdk/open-im-server/v3/internal/api/jssdk"
"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/prommetrics"
"github.com/openimsdk/open-im-server/v3/pkg/common/servererrs"
@ -357,9 +354,7 @@ func GinParseToken(authClient *rpcli.AuthClient) gin.HandlerFunc {
func setGinIsAdmin(imAdminUserID []string) gin.HandlerFunc {
return func(c *gin.Context) {
opUserID := mcontext.GetOpUserID(c)
admin := datautil.Contain(opUserID, imAdminUserID...)
c.Set(authverify.CtxIsAdminKey, admin)
c.Set(authverify.CtxAdminUserIDsKey, imAdminUserID)
}
}

2
internal/msggateway/ws_server.go
Unescape View File

@ -49,7 +49,7 @@ type WsServer struct {
unregisterChan chan *Client
kickHandlerChan chan *kickHandler
clients UserMap
online *rpccache.OnlineCache
online rpccache.OnlineCache
subscription *Subscription
clientPool sync.Pool
onlineUserNum atomic.Int64

10
internal/msgtransfer/online_msg_to_mongo_handler.go
Unescape View File

@ -48,15 +48,7 @@ func (mc *OnlineHistoryMongoConsumerHandler) HandleChatWs2Mongo(ctx context.Cont
log.ZDebug(ctx, "mongo consumer recv msg", "msgs", msgFromMQ.String())
err = mc.msgTransferDatabase.BatchInsertChat2DB(ctx, msgFromMQ.ConversationID, msgFromMQ.MsgData, msgFromMQ.LastSeq)
if err != nil {
log.ZError(
ctx,
"single data insert to mongo err",
err,
"msg",
msgFromMQ.MsgData,
"conversationID",
msgFromMQ.ConversationID,
)
log.ZError(ctx, "single data insert to mongo err", err, "msg", msgFromMQ.MsgData, "conversationID", msgFromMQ.ConversationID)
prommetrics.MsgInsertMongoFailedCounter.Inc()
} else {
prommetrics.MsgInsertMongoSuccessCounter.Inc()

8
internal/push/push_handler.go
Unescape View File

@ -33,7 +33,7 @@ type ConsumerHandler struct {
offlinePusher offlinepush.OfflinePusher
onlinePusher OnlinePusher
pushDatabase controller.PushDatabase
onlineCache *rpccache.OnlineCache
onlineCache rpccache.OnlineCache
groupLocalCache *rpccache.GroupLocalCache
conversationLocalCache *rpccache.ConversationLocalCache
webhookClient *webhook.Client
@ -120,11 +120,7 @@ func (c *ConsumerHandler) HandleMs2PsChat(ctx context.Context, msg []byte) {
}
func (c *ConsumerHandler) WaitCache() {
c.onlineCache.Lock.Lock()
for c.onlineCache.CurrentPhase.Load() < rpccache.DoSubscribeOver {
c.onlineCache.Cond.Wait()
}
c.onlineCache.Lock.Unlock()
c.onlineCache.WaitCache()
}
// Push2User Suitable for two types of conversations, one is SingleChatType and the other is NotificationChatType.

6
pkg/authverify/token.go
Unescape View File

@ -51,15 +51,15 @@ func CheckSystemAccount(ctx context.Context, level int32) bool {
}
const (
CtxIsAdminKey = "CtxIsAdminKey"
CtxAdminUserIDsKey = "CtxAdminUserIDsKey"
)
func WithIMAdminUserIDs(ctx context.Context, imAdminUserID []string) context.Context {
return context.WithValue(ctx, CtxIsAdminKey, imAdminUserID)
return context.WithValue(ctx, CtxAdminUserIDsKey, imAdminUserID)
}
func GetIMAdminUserIDs(ctx context.Context) []string {
imAdminUserID, _ := ctx.Value(CtxIsAdminKey).([]string)
imAdminUserID, _ := ctx.Value(CtxAdminUserIDsKey).([]string)
return imAdminUserID
}

60
pkg/common/startrpc/start.go
Unescape View File

@ -47,64 +47,6 @@ func init() {
prommetrics.RegistryAll()
}
func getConfigRpcMaxRequestBody(value reflect.Value) *conf.MaxRequestBody {
for value.Kind() == reflect.Pointer {
value = value.Elem()
}
if value.Kind() == reflect.Struct {
num := value.NumField()
for i := 0; i < num; i++ {
field := value.Field(i)
if !field.CanInterface() {
continue
}
for field.Kind() == reflect.Pointer {
field = field.Elem()
}
switch elem := field.Interface().(type) {
case conf.Share:
return &elem.RPCMaxBodySize
case conf.MaxRequestBody:
return &elem
}
if field.Kind() == reflect.Struct {
if elem := getConfigRpcMaxRequestBody(field); elem != nil {
return elem
}
}
}
}
return nil
}
func getConfigShare(value reflect.Value) *conf.Share {
for value.Kind() == reflect.Pointer {
value = value.Elem()
}
if value.Kind() == reflect.Struct {
num := value.NumField()
for i := 0; i < num; i++ {
field := value.Field(i)
if !field.CanInterface() {
continue
}
for field.Kind() == reflect.Pointer {
field = field.Elem()
}
switch elem := field.Interface().(type) {
case conf.Share:
return &elem
}
if field.Kind() == reflect.Struct {
if elem := getConfigShare(field); elem != nil {
return elem
}
}
}
}
return nil
}
func Start[T any](ctx context.Context, disc *conf.Discovery, prometheusConfig *conf.Prometheus, listenIP,
registerIP string, autoSetPorts bool, rpcPorts []int, index int, rpcRegisterName string, notification *conf.Notification, config T,
watchConfigNames []string, watchServiceNames []string,
@ -122,8 +64,8 @@ func Start[T any](ctx context.Context, disc *conf.Discovery, prometheusConfig *c
options = append(options,
grpcsrv.GrpcServerMetadataContext(),
grpcsrv.GrpcServerLogger(),
grpcsrv.GrpcServerErrorConvert(),
grpcsrv.GrpcServerLogger(),
grpcsrv.GrpcServerRequestValidate(),
grpcsrv.GrpcServerPanicCapture(),
)

39
pkg/common/startrpc/tools.go
Unescape View File

@ -0,0 +1,39 @@
package startrpc
import (
"reflect"
conf "github.com/openimsdk/open-im-server/v3/pkg/common/config"
)
func getConfig[T any](value reflect.Value) *T {
for value.Kind() == reflect.Pointer {
value = value.Elem()
}
if value.Kind() == reflect.Struct {
num := value.NumField()
for i := 0; i < num; i++ {
field := value.Field(i)
for field.Kind() == reflect.Pointer {
field = field.Elem()
}
if field.Kind() == reflect.Struct {
if elem, ok := field.Interface().(T); ok {
return &elem
}
if elem := getConfig[T](field); elem != nil {
return elem
}
}
}
}
return nil
}
func getConfigRpcMaxRequestBody(value reflect.Value) *conf.MaxRequestBody {
return getConfig[conf.MaxRequestBody](value)
}
func getConfigShare(value reflect.Value) *conf.Share {
return getConfig[conf.Share](value)
}

92
pkg/rpccache/online.go
Unescape View File

@ -9,6 +9,7 @@ import (
"sync/atomic"
"time"
"github.com/openimsdk/open-im-server/v3/pkg/common/config"
"github.com/openimsdk/open-im-server/v3/pkg/rpcli"
"github.com/openimsdk/protocol/constant"
"github.com/openimsdk/protocol/user"
@ -23,9 +24,25 @@ import (
"github.com/redis/go-redis/v9"
)
func NewOnlineCache(client *rpcli.UserClient, group *GroupLocalCache, rdb redis.UniversalClient, fullUserCache bool, fn func(ctx context.Context, userID string, platformIDs []int32)) (*OnlineCache, error) {
const (
Begin uint32 = iota
DoOnlineStatusOver
DoSubscribeOver
)
type OnlineCache interface {
GetUserOnlinePlatform(ctx context.Context, userID string) ([]int32, error)
GetUserOnline(ctx context.Context, userID string) (bool, error)
GetUsersOnline(ctx context.Context, userIDs []string) ([]string, []string, error)
WaitCache()
}
func NewOnlineCache(client *rpcli.UserClient, group *GroupLocalCache, rdb redis.UniversalClient, fullUserCache bool, fn func(ctx context.Context, userID string, platformIDs []int32)) (OnlineCache, error) {
if config.Standalone() {
return disableOnlineCache{client: client}, nil
}
l := &sync.Mutex{}
x := &OnlineCache{
x := &defaultOnlineCache{
client: client,
group: group,
fullUserCache: fullUserCache,
@ -60,13 +77,7 @@ func NewOnlineCache(client *rpcli.UserClient, group *GroupLocalCache, rdb redis.
return x, nil
}
const (
Begin uint32 = iota
DoOnlineStatusOver
DoSubscribeOver
)
type OnlineCache struct {
type defaultOnlineCache struct {
client *rpcli.UserClient
group *GroupLocalCache
@ -82,7 +93,7 @@ type OnlineCache struct {
CurrentPhase atomic.Uint32
}
func (o *OnlineCache) initUsersOnlineStatus(ctx context.Context) (err error) {
func (o *defaultOnlineCache) initUsersOnlineStatus(ctx context.Context) (err error) {
log.ZDebug(ctx, "init users online status begin")
var (
@ -135,7 +146,7 @@ func (o *OnlineCache) initUsersOnlineStatus(ctx context.Context) (err error) {
return nil
}
func (o *OnlineCache) doSubscribe(ctx context.Context, rdb redis.UniversalClient, fn func(ctx context.Context, userID string, platformIDs []int32)) {
func (o *defaultOnlineCache) doSubscribe(ctx context.Context, rdb redis.UniversalClient, fn func(ctx context.Context, userID string, platformIDs []int32)) {
o.Lock.Lock()
ch := rdb.Subscribe(ctx, cachekey.OnlineChannel).Channel()
for o.CurrentPhase.Load() < DoOnlineStatusOver {
@ -186,7 +197,7 @@ func (o *OnlineCache) doSubscribe(ctx context.Context, rdb redis.UniversalClient
}
}
func (o *OnlineCache) getUserOnlinePlatform(ctx context.Context, userID string) ([]int32, error) {
func (o *defaultOnlineCache) getUserOnlinePlatform(ctx context.Context, userID string) ([]int32, error) {
platformIDs, err := o.lruCache.Get(userID, func() ([]int32, error) {
return o.client.GetUserOnlinePlatform(ctx, userID)
})
@ -198,7 +209,7 @@ func (o *OnlineCache) getUserOnlinePlatform(ctx context.Context, userID string)
return platformIDs, nil
}
func (o *OnlineCache) GetUserOnlinePlatform(ctx context.Context, userID string) ([]int32, error) {
func (o *defaultOnlineCache) GetUserOnlinePlatform(ctx context.Context, userID string) ([]int32, error) {
platformIDs, err := o.getUserOnlinePlatform(ctx, userID)
if err != nil {
return nil, err
@ -208,7 +219,7 @@ func (o *OnlineCache) GetUserOnlinePlatform(ctx context.Context, userID string)
return platformIDs, nil
}
func (o *OnlineCache) GetUserOnline(ctx context.Context, userID string) (bool, error) {
func (o *defaultOnlineCache) GetUserOnline(ctx context.Context, userID string) (bool, error) {
platformIDs, err := o.getUserOnlinePlatform(ctx, userID)
if err != nil {
return false, err
@ -216,7 +227,7 @@ func (o *OnlineCache) GetUserOnline(ctx context.Context, userID string) (bool, e
return len(platformIDs) > 0, nil
}
func (o *OnlineCache) getUserOnlinePlatformBatch(ctx context.Context, userIDs []string) (map[string][]int32, error) {
func (o *defaultOnlineCache) getUserOnlinePlatformBatch(ctx context.Context, userIDs []string) (map[string][]int32, error) {
if len(userIDs) == 0 {
return nil, nil
}
@ -240,7 +251,7 @@ func (o *OnlineCache) getUserOnlinePlatformBatch(ctx context.Context, userIDs []
return platformIDsMap, nil
}
func (o *OnlineCache) GetUsersOnline(ctx context.Context, userIDs []string) ([]string, []string, error) {
func (o *defaultOnlineCache) GetUsersOnline(ctx context.Context, userIDs []string) ([]string, []string, error) {
t := time.Now()
var (
@ -276,7 +287,7 @@ func (o *OnlineCache) GetUsersOnline(ctx context.Context, userIDs []string) ([]s
return onlineUserIDs, offlineUserIDs, nil
}
func (o *OnlineCache) setUserOnline(userID string, platformIDs []int32) {
func (o *defaultOnlineCache) setUserOnline(userID string, platformIDs []int32) {
switch o.fullUserCache {
case true:
o.mapCache.Store(userID, platformIDs)
@ -285,6 +296,51 @@ func (o *OnlineCache) setUserOnline(userID string, platformIDs []int32) {
}
}
func (o *OnlineCache) setHasUserOnline(userID string, platformIDs []int32) bool {
func (o *defaultOnlineCache) setHasUserOnline(userID string, platformIDs []int32) bool {
return o.lruCache.SetHas(userID, platformIDs)
}
func (o *defaultOnlineCache) WaitCache() {
o.Lock.Lock()
for o.CurrentPhase.Load() < DoSubscribeOver {
o.Cond.Wait()
}
o.Lock.Unlock()
}
type disableOnlineCache struct {
client *rpcli.UserClient
}
func (o disableOnlineCache) GetUserOnlinePlatform(ctx context.Context, userID string) ([]int32, error) {
return o.client.GetUserOnlinePlatform(ctx, userID)
}
func (o disableOnlineCache) GetUserOnline(ctx context.Context, userID string) (bool, error) {
onlinePlatform, err := o.client.GetUserOnlinePlatform(ctx, userID)
if err != nil {
return false, err
}
return len(onlinePlatform) > 0, err
}
func (o disableOnlineCache) GetUsersOnline(ctx context.Context, userIDs []string) ([]string, []string, error) {
var (
onlineUserIDs = make([]string, 0, len(userIDs))
offlineUserIDs = make([]string, 0, len(userIDs))
)
for _, userID := range userIDs {
online, err := o.GetUserOnline(ctx, userID)
if err != nil {
return nil, nil, err
}
if online {
onlineUserIDs = append(onlineUserIDs, userID)
} else {
offlineUserIDs = append(offlineUserIDs, userID)
}
}
return onlineUserIDs, offlineUserIDs, nil
}
func (o disableOnlineCache) WaitCache() {}

Write Preview
Loading…
Cancel
Save