conversation

internal/rpc/msg/server.go

@@ -24,6 +24,7 @@ type MessageInterceptorChain []MessageInterceptorFunc
 type msgServer struct {
 	RegisterCenter       discoveryregistry.SvcDiscoveryRegistry
 	MsgDatabase          controller.MsgDatabase
+	notificationDatabase controller.NotificationDatabase
 	ExtendMsgDatabase    controller.ExtendMsgDatabase
 	Group                *rpcclient.GroupClient
 	User                 *rpcclient.UserClient
@@ -161,6 +162,7 @@ func (m *msgServer) GetMaxAndMinSeq(ctx context.Context, req *sdkws.GetMaxAndMin
 func (m *msgServer) PullMessageBySeqs(ctx context.Context, req *sdkws.PullMessageBySeqsReq) (*sdkws.PullMessageBySeqsResp, error) {
 	resp := &sdkws.PullMessageBySeqsResp{}
 	for _, seq := range req.SeqRanges {
+		if !seq.IsNotification {
 			msgs, err := m.MsgDatabase.GetMsgBySeqsRange(ctx, seq.ConversationID, seq.Begin, seq.End, seq.Num)
 			if err != nil {
 				return nil, err
@@ -169,6 +171,22 @@ func (m *msgServer) PullMessageBySeqs(ctx context.Context, req *sdkws.PullMessag
 				ConversationID: seq.ConversationID,
 				Msgs:           msgs,
 			})
+		} else {
+			var seqs []int64
+			for i := seq.Begin; i <= seq.End; i++ {
+				seqs = append(seqs, i)
+			}
+			msgs, err := m.notificationDatabase.GetMsgBySeqs(ctx, seq.ConversationID, seqs)
+			if err != nil {
+				return nil, err
+			}
+			resp.Msgs = append(resp.Msgs, &sdkws.PullMsgs{
+				ConversationID: seq.ConversationID,
+				Msgs:           msgs,
+				IsNotification: true,
+			})
+		}
+
 	}
 	return resp, nil
 }