@@ -22,11 +22,15 @@ import (
2222 "github.com/openimsdk/open-im-server/v3/pkg/rpcclient"
2323 "github.com/openimsdk/protocol/constant"
2424 "github.com/openimsdk/protocol/msggateway"
25+ "github.com/openimsdk/protocol/sdkws"
2526 "github.com/openimsdk/tools/discovery"
2627 "github.com/openimsdk/tools/errs"
2728 "github.com/openimsdk/tools/log"
2829 "github.com/openimsdk/tools/mcontext"
30+ "github.com/openimsdk/tools/mq/memamq"
31+ "github.com/openimsdk/tools/utils/datautil"
2932 "google.golang.org/grpc"
33+ "sync/atomic"
3034)
3135
3236func (s * Server ) InitServer (ctx context.Context , config * Config , disCov discovery.SvcDiscoveryRegistry , server * grpc.Server ) error {
@@ -57,6 +61,7 @@ type Server struct {
5761 pushTerminal map [int ]struct {}
5862 ready func (srv * Server ) error
5963 userRcp rpcclient.UserRpcClient
64+ queue * memamq.MemoryQueue
6065}
6166
6267func (s * Server ) SetLongConnServer (LongConnServer LongConnServer ) {
@@ -70,6 +75,7 @@ func NewServer(rpcPort int, longConnServer LongConnServer, conf *Config, ready f
7075 pushTerminal : make (map [int ]struct {}),
7176 config : conf ,
7277 ready : ready ,
78+ queue : memamq .NewMemoryQueue (512 , 1024 * 16 ),
7379 }
7480 s .pushTerminal [constant .IOSPlatformID ] = struct {}{}
7581 s .pushTerminal [constant .AndroidPlatformID ] = struct {}{}
@@ -125,55 +131,93 @@ func (s *Server) OnlineBatchPushOneMsg(ctx context.Context, req *msggateway.Onli
125131 return nil , nil
126132}
127133
128- func (s * Server ) SuperGroupOnlineBatchPushOneMsg (ctx context.Context , req * msggateway.OnlineBatchPushOneMsgReq ,
129- ) (* msggateway.OnlineBatchPushOneMsgResp , error ) {
130- var singleUserResults []* msggateway.SingleMsgToUserResults
131- for _ , v := range req .PushToUserIDs {
132- var resp []* msggateway.SingleMsgToUserPlatform
133- results := & msggateway.SingleMsgToUserResults {
134- UserID : v ,
134+ func (s * Server ) pushToUser (ctx context.Context , userID string , msgData * sdkws.MsgData ) * msggateway.SingleMsgToUserResults {
135+ clients , ok := s .LongConnServer .GetUserAllCons (userID )
136+ if ! ok {
137+ log .ZDebug (ctx , "push user not online" , "userID" , userID )
138+ return & msggateway.SingleMsgToUserResults {
139+ UserID : userID ,
135140 }
136- clients , ok := s .LongConnServer .GetUserAllCons (v )
137- if ! ok {
138- log .ZDebug (ctx , "push user not online" , "userID" , v )
139- results .Resp = resp
140- singleUserResults = append (singleUserResults , results )
141+ }
142+ log .ZDebug (ctx , "push user online" , "clients" , clients , "userID" , userID )
143+ result := & msggateway.SingleMsgToUserResults {
144+ UserID : userID ,
145+ Resp : make ([]* msggateway.SingleMsgToUserPlatform , 0 , len (clients )),
146+ }
147+ for _ , client := range clients {
148+ if client == nil {
141149 continue
142150 }
143-
144- log .ZDebug (ctx , "push user online" , "clients" , clients , "userID" , v )
145- for _ , client := range clients {
146- if client == nil {
147- continue
151+ userPlatform := & msggateway.SingleMsgToUserPlatform {
152+ RecvPlatFormID : int32 (client .PlatformID ),
153+ }
154+ if ! client .IsBackground ||
155+ (client .IsBackground && client .PlatformID != constant .IOSPlatformID ) {
156+ err := client .PushMessage (ctx , msgData )
157+ if err != nil {
158+ userPlatform .ResultCode = int64 (servererrs .ErrPushMsgErr .Code ())
159+ } else {
160+ if _ , ok := s .pushTerminal [client .PlatformID ]; ok {
161+ result .OnlinePush = true
162+ }
148163 }
164+ } else {
165+ userPlatform .ResultCode = int64 (servererrs .ErrIOSBackgroundPushErr .Code ())
166+ }
167+ result .Resp = append (result .Resp , userPlatform )
168+ }
169+ return result
170+ }
149171
150- userPlatform := & msggateway.SingleMsgToUserPlatform {
151- RecvPlatFormID : int32 (client .PlatformID ),
172+ func (s * Server ) SuperGroupOnlineBatchPushOneMsg (ctx context.Context , req * msggateway.OnlineBatchPushOneMsgReq ) (* msggateway.OnlineBatchPushOneMsgResp , error ) {
173+ if len (req .PushToUserIDs ) == 0 {
174+ return & msggateway.OnlineBatchPushOneMsgResp {}, nil
175+ }
176+ ch := make (chan * msggateway.SingleMsgToUserResults , len (req .PushToUserIDs ))
177+ var count atomic.Int64
178+ count .Add (int64 (len (req .PushToUserIDs )))
179+ for i := range req .PushToUserIDs {
180+ userID := req .PushToUserIDs [i ]
181+ err := s .queue .PushCtx (ctx , func () {
182+ ch <- s .pushToUser (ctx , userID , req .MsgData )
183+ if count .Add (- 1 ) == 0 {
184+ close (ch )
152185 }
153- if ! client .IsBackground ||
154- (client .IsBackground && client .PlatformID != constant .IOSPlatformID ) {
155- err := client .PushMessage (ctx , req .MsgData )
156- if err != nil {
157- userPlatform .ResultCode = int64 (servererrs .ErrPushMsgErr .Code ())
158- resp = append (resp , userPlatform )
159- } else {
160- if _ , ok := s .pushTerminal [client .PlatformID ]; ok {
161- results .OnlinePush = true
162- resp = append (resp , userPlatform )
163- }
164- }
165- } else {
166- userPlatform .ResultCode = int64 (servererrs .ErrIOSBackgroundPushErr .Code ())
167- resp = append (resp , userPlatform )
186+ })
187+ if err != nil {
188+ if count .Add (- 1 ) == 0 {
189+ close (ch )
190+ }
191+ log .ZError (ctx , "pushToUser MemoryQueue failed" , err , "userID" , userID )
192+ ch <- & msggateway.SingleMsgToUserResults {
193+ UserID : userID ,
168194 }
169195 }
170- results .Resp = resp
171- singleUserResults = append (singleUserResults , results )
172196 }
173-
174- return & msggateway.OnlineBatchPushOneMsgResp {
175- SinglePushResult : singleUserResults ,
176- }, nil
197+ resp := & msggateway.OnlineBatchPushOneMsgResp {
198+ SinglePushResult : make ([]* msggateway.SingleMsgToUserResults , 0 , len (req .PushToUserIDs )),
199+ }
200+ for {
201+ select {
202+ case <- ctx .Done ():
203+ log .ZError (ctx , "SuperGroupOnlineBatchPushOneMsg ctx done" , context .Cause (ctx ))
204+ userIDSet := datautil .SliceSet (req .PushToUserIDs )
205+ for _ , results := range resp .SinglePushResult {
206+ delete (userIDSet , results .UserID )
207+ }
208+ for userID := range userIDSet {
209+ resp .SinglePushResult = append (resp .SinglePushResult , & msggateway.SingleMsgToUserResults {
210+ UserID : userID ,
211+ })
212+ }
213+ return resp , nil
214+ case res , ok := <- ch :
215+ if ! ok {
216+ return resp , nil
217+ }
218+ resp .SinglePushResult = append (resp .SinglePushResult , res )
219+ }
220+ }
177221}
178222
179223func (s * Server ) KickUserOffline (
0 commit comments