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

fix:get max seqs

Browse Source
This commit is contained in:
icey-yu 2024-07-03 12:08:52 +08:00
parent e7b4a24319
commit 0f2acd1569
1 changed files with 10 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

40
pkg/rpccache/conversation.go
View File

@ -23,14 +23,12 @@ import (
pbconversation "github.com/openimsdk/protocol/conversation" pbconversation "github.com/openimsdk/protocol/conversation"
"github.com/openimsdk/tools/errs" "github.com/openimsdk/tools/errs"
"github.com/openimsdk/tools/log" "github.com/openimsdk/tools/log"
"github.com/openimsdk/tools/mq/memamq"
"github.com/redis/go-redis/v9" "github.com/redis/go-redis/v9"
"sync" "golang.org/x/sync/errgroup"
) )
const ( const (
notificationWorkerCount = 20 conversationWorkerCount = 20
notificationBufferSize = 20000
) )
func NewConversationLocalCache(client rpcclient.ConversationRpcClient, localCache *config.LocalCache, cli redis.UniversalClient) *ConversationLocalCache { func NewConversationLocalCache(client rpcclient.ConversationRpcClient, localCache *config.LocalCache, cli redis.UniversalClient) *ConversationLocalCache {
@ -45,7 +43,6 @@ func NewConversationLocalCache(client rpcclient.ConversationRpcClient, localCach
localcache.WithLocalSuccessTTL(lc.Success()), localcache.WithLocalSuccessTTL(lc.Success()),
localcache.WithLocalFailedTTL(lc.Failed()), localcache.WithLocalFailedTTL(lc.Failed()),
), ),
queue: memamq.NewMemoryQueue(notificationWorkerCount, notificationBufferSize),
} }
if lc.Enable() { if lc.Enable() {
go subscriberRedisDeleteCache(context.Background(), cli, lc.Topic, x.local.DelLocal) go subscriberRedisDeleteCache(context.Background(), cli, lc.Topic, x.local.DelLocal)
@ -56,8 +53,6 @@ func NewConversationLocalCache(client rpcclient.ConversationRpcClient, localCach
type ConversationLocalCache struct { type ConversationLocalCache struct {
client rpcclient.ConversationRpcClient client rpcclient.ConversationRpcClient
local localcache.Cache[any] local localcache.Cache[any]
queue *memamq.MemoryQueue
} }
func (c *ConversationLocalCache) GetConversationIDs(ctx context.Context, ownerUserID string) (val []string, err error) { func (c *ConversationLocalCache) GetConversationIDs(ctx context.Context, ownerUserID string) (val []string, err error) {
@ -101,48 +96,33 @@ func (c *ConversationLocalCache) GetSingleConversationRecvMsgOpt(ctx context.Con
func (c *ConversationLocalCache) GetConversations(ctx context.Context, ownerUserID string, conversationIDs []string) ([]*pbconversation.Conversation, error) { func (c *ConversationLocalCache) GetConversations(ctx context.Context, ownerUserID string, conversationIDs []string) ([]*pbconversation.Conversation, error) {
var ( var (
conversations = make([]*pbconversation.Conversation, 0, len(conversationIDs)) conversations = make([]*pbconversation.Conversation, 0, len(conversationIDs))
errChan = make(chan error, 1)
conversationsChan = make(chan *pbconversation.Conversation, len(conversationIDs)) conversationsChan = make(chan *pbconversation.Conversation, len(conversationIDs))
wg sync.WaitGroup
) )
wg.Add(len(conversationIDs)) g, ctx := errgroup.WithContext(ctx)
g.SetLimit(conversationWorkerCount)
for _, conversationID := range conversationIDs { for _, conversationID := range conversationIDs {
conversationID := conversationID conversationID := conversationID
err := c.queue.Push(func() { g.Go(func() error {
defer wg.Done()
conversation, err := c.GetConversation(ctx, ownerUserID, conversationID) conversation, err := c.GetConversation(ctx, ownerUserID, conversationID)
if err != nil { if err != nil {
if errs.ErrRecordNotFound.Is(err) { if errs.ErrRecordNotFound.Is(err) {
return return nil
} }
select { return err
case errChan <- err:
default:
}
return
} }
conversationsChan <- conversation conversationsChan <- conversation
return nil
}) })
if err != nil {
// push err
return nil, errs.Wrap(err)
} }
} if err := g.Wait(); err != nil {
wg.Wait()
close(errChan)
close(conversationsChan)
err, ok := <-errChan
if ok {
return nil, err return nil, err
} }
close(conversationsChan)
for conversation := range conversationsChan { for conversation := range conversationsChan {
conversations = append(conversations, conversation) conversations = append(conversations, conversation)
} }
return conversations, nil return conversations, nil
} }