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

update waitgroup to errgroup.

Browse Source
This commit is contained in:
Monet Lee 2024-08-01 15:05:00 +08:00
parent 3e547958b0
commit 07eb0cb387
1 changed files with 8 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
internal/rpc/msg/clear.go
View File

@ -3,12 +3,10 @@ package msg
import ( import (
"context" "context"
"strings" "strings"
"sync"
"time" "time"
"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/convert" "github.com/openimsdk/open-im-server/v3/pkg/common/convert"
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
"github.com/openimsdk/protocol/conversation" "github.com/openimsdk/protocol/conversation"
pbconversation "github.com/openimsdk/protocol/conversation" pbconversation "github.com/openimsdk/protocol/conversation"
"github.com/openimsdk/protocol/msg" "github.com/openimsdk/protocol/msg"
@ -18,6 +16,7 @@ import (
"github.com/openimsdk/tools/mcontext" "github.com/openimsdk/tools/mcontext"
"github.com/openimsdk/tools/utils/idutil" "github.com/openimsdk/tools/utils/idutil"
"github.com/openimsdk/tools/utils/stringutil" "github.com/openimsdk/tools/utils/stringutil"
"golang.org/x/sync/errgroup"
) )
// hard delete in Database. // hard delete in Database.
@ -94,16 +93,15 @@ func (m *msgServer) DestructMsgs(ctx context.Context, req *msg.DestructMsgsReq)
temp := convert.ConversationsPb2DB(req.Conversations) temp := convert.ConversationsPb2DB(req.Conversations)
batchNum := 100 batchNum := 100
var wg sync.WaitGroup
wg.Add((len(temp) + batchNum - 1) / batchNum) errg, _ := errgroup.WithContext(ctx)
errg.SetLimit(100)
for i := 0; i < len(temp); i += batchNum { for i := 0; i < len(temp); i += batchNum {
batch := temp[i:min(i+batchNum, len(temp))] batch := temp[i:min(i+batchNum, len(temp))]
go func(batch []*model.Conversation) { errg.Go(func() error {
defer wg.Done() for _, conversation := range batch {
for _, conversation := range temp {
handleCtx := mcontext.NewCtx(stringutil.GetSelfFuncName() + "-" + idutil.OperationIDGenerator() + "-" + conversation.ConversationID + "-" + conversation.OwnerUserID) handleCtx := mcontext.NewCtx(stringutil.GetSelfFuncName() + "-" + idutil.OperationIDGenerator() + "-" + conversation.ConversationID + "-" + conversation.OwnerUserID)
log.ZDebug(handleCtx, "User MsgsDestruct", log.ZDebug(handleCtx, "User MsgsDestruct",
"conversationID", conversation.ConversationID, "conversationID", conversation.ConversationID,
@ -131,10 +129,8 @@ func (m *msgServer) DestructMsgs(ctx context.Context, req *msg.DestructMsgsReq)
// m.msgNotificationSender.UserDeleteMsgsNotification(handleCtx, conversation.OwnerUserID, conversation.ConversationID, seqs) // m.msgNotificationSender.UserDeleteMsgsNotification(handleCtx, conversation.OwnerUserID, conversation.ConversationID, seqs)
} }
} }
}(batch) return nil
})
} }
wg.Wait()
return nil, nil return nil, nil
} }