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

feat: fix err

Browse Source
This commit is contained in:
icey-yu 2024-09-06 14:43:00 +08:00
parent 834964a3f2
commit 842be3a71d
2 changed files with 59 additions and 45 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
pkg/localcache/lru/lru_lazy.go
View File

@ -156,6 +156,11 @@ func (x *LayLRU[K, V]) GetBatch(keys []K, fetchBatch func([]K) (map[K]V, error))
} }
func (x *LayLRU[K, V]) GetBatchs(keys []K, fetch func(keys []K) (map[K]V, error)) ([]V, error) { func (x *LayLRU[K, V]) GetBatchs(keys []K, fetch func(keys []K) (map[K]V, error)) ([]V, error) {
var (
err error
once sync.Once
)
x.lock.Lock() x.lock.Lock()
res := make([]V, 0) res := make([]V, 0)
queries := make([]K, 0) queries := make([]K, 0)
@ -165,18 +170,28 @@ func (x *LayLRU[K, V]) GetBatchs(keys []K, fetch func(keys []K) (map[K]V, error)
if ok { if ok {
x.lock.Unlock() x.lock.Unlock()
v.lock.Lock() v.lock.Lock()
expires, value, _ := v.expires, v.value, v.err expires, value, err1 := v.expires, v.value, v.err
if expires != 0 && expires > time.Now().UnixMilli() { if expires != 0 && expires > time.Now().UnixMilli() {
v.lock.Unlock() v.lock.Unlock()
x.target.IncrGetHit() x.target.IncrGetHit()
res = append(res, value) res = append(res, value)
if err1 != nil {
once.Do(func() {
err = err1
})
}
continue continue
} }
} }
queries = append(queries, key) queries = append(queries, key)
x.lock.Unlock() x.lock.Unlock()
} }
values, err := fetch(queries) values, err1 := fetch(queries)
if err1 != nil {
once.Do(func() {
err = err1
})
}
for key, val := range values { for key, val := range values {
v := &layLruItem[V]{} v := &layLruItem[V]{}
v.value = val v.value = val
@ -198,7 +213,7 @@ func (x *LayLRU[K, V]) GetBatchs(keys []K, fetch func(keys []K) (map[K]V, error)
return res, err return res, err
} }
func (x *LayLRU[K, V]) SetBatch(data map[K]V) bool { func (x *LayLRU[K, V]) SetBatch(data map[K]V) {
x.lock.Lock() x.lock.Lock()
defer x.lock.Unlock() defer x.lock.Unlock()

83
pkg/rpccache/online.go
View File

@ -155,51 +155,50 @@ func (o *OnlineCache) GetUserOnline(ctx context.Context, userID string) (bool, e
return len(platformIDs) > 0, nil return len(platformIDs) > 0, nil
} }
func (o *OnlineCache) getUserOnlinePlatformBatch(ctx context.Context, userIDs []string) (map[string][]int32, error) { //func (o *OnlineCache) getUserOnlinePlatformBatch(ctx context.Context, userIDs []string) (map[string][]int32, error) {
platformIDsMap, err := o.lruCache.GetBatch(userIDs, func(missingUsers []string) (map[string][]int32, error) { // platformIDsMap, err := o.lruCache.GetBatch(userIDs, func(missingUsers []string) (map[string][]int32, error) {
platformIDsMap := make(map[string][]int32) // platformIDsMap := make(map[string][]int32)
//
// usersStatus, err := o.user.GetUsersOnlinePlatform(ctx, missingUsers)
// if err != nil {
// return nil, err
// }
//
// for _, user := range usersStatus {
// platformIDsMap[user.UserID] = user.PlatformIDs
// }
//
// return platformIDsMap, nil
// })
// if err != nil {
// log.ZError(ctx, "OnlineCache GetUserOnlinePlatform", err, "userID", userIDs)
// return nil, err
// }
//
// //log.ZDebug(ctx, "OnlineCache GetUserOnlinePlatform", "userID", userID, "platformIDs", platformIDs)
// return platformIDsMap, nil
//}
usersStatus, err := o.user.GetUsersOnlinePlatform(ctx, missingUsers) func (o *OnlineCache) GetUsersOnline(ctx context.Context, userIDs []string) ([]string, []string, error) {
if err != nil {
return nil, err
}
for _, user := range usersStatus {
platformIDsMap[user.UserID] = user.PlatformIDs
}
return platformIDsMap, nil
})
if err != nil {
log.ZError(ctx, "OnlineCache GetUserOnlinePlatform", err, "userID", userIDs)
return nil, err
}
//log.ZDebug(ctx, "OnlineCache GetUserOnlinePlatform", "userID", userID, "platformIDs", platformIDs)
return platformIDsMap, nil
}
func (o *OnlineCache) GetUsersOnline(ctx context.Context, usersID []string) ([]string, []string, error) {
t := time.Now() t := time.Now()
var ( var (
onlineUserIDs []string onlineUserIDs = make([]string, 0, len(userIDs))
offlineUserIDs []string offlineUserIDs = make([]string, 0, len(userIDs))
) )
//
userOnlineMap, err := o.getUserOnlinePlatformBatch(ctx, userIDs) //userOnlineMap, err := o.getUserOnlinePlatformBatch(ctx, userIDs)
if err != nil { //if err != nil {
return nil, nil, err // return nil, nil, err
} //}
//
for key, value := range userOnlineMap { //for key, value := range userOnlineMap {
if len(value) > 0 { // if len(value) > 0 {
onlineUserIDs = append(onlineUserIDs, key) // onlineUserIDs = append(onlineUserIDs, key)
} else { // } else {
offlineUserIDs = append(offlineUserIDs, key) // offlineUserIDs = append(offlineUserIDs, key)
} // }
} //}
switch o.fullUserCache { switch o.fullUserCache {
case true: case true:
@ -213,8 +212,8 @@ func (o *OnlineCache) GetUsersOnline(ctx context.Context, usersID []string) ([]s
case false: case false:
} }
log.ZWarn(ctx, "get users online", nil, "online users length", len(onlineUserIDS), "offline users length", len(offlineUserIDs), "cost", time.Since(t)) log.ZWarn(ctx, "get users online", nil, "online users length", len(userIDs), "offline users length", len(offlineUserIDs), "cost", time.Since(t))
return onlineUserIDS, offlineUserIDs, nil return userIDs, offlineUserIDs, nil
} }
//func (o *OnlineCache) GetUsersOnline(ctx context.Context, userIDs []string) ([]string, error) { //func (o *OnlineCache) GetUsersOnline(ctx context.Context, userIDs []string) ([]string, error) {