@@ -39,6 +39,8 @@ import (
3939 pbConversation "github.com/openimsdk/protocol/conversation"
4040)
4141
42+ const MaxRecursionDepth = 3
43+
4244func (c * Conversation ) setConversation (ctx context.Context , apiReq * pbConversation.SetConversationsReq , localConversation * model_struct.LocalConversation ) error {
4345 apiReq .Conversation .ConversationID = localConversation .ConversationID
4446 apiReq .Conversation .ConversationType = localConversation .ConversationType
@@ -77,19 +79,12 @@ func (c *Conversation) getAdvancedHistoryMessageList(ctx context.Context, req sd
7779 log .ZDebug (ctx , "pull message" , "pull cost time" , time .Since (t ))
7880 t = time .Now ()
7981
80- var thisEndSeq int64
81- thisEndSeq , messageList = c .LocalChatLog2MsgStruct (list , isReverse )
82+ messageList = c .LocalChatLog2MsgStruct (list )
8283 log .ZDebug (ctx , "message convert and unmarshal" , "unmarshal cost time" , time .Since (t ))
8384 t = time .Now ()
8485 if ! isReverse {
8586 sort .Sort (messageList )
86- if thisEndSeq != 0 {
87- c .messagePullForwardEndSeqMap .Store (conversationID , thisEndSeq )
88- }
89- } else {
90- if thisEndSeq != 0 {
91- c .messagePullReverseEndSeqMap .Store (conversationID , thisEndSeq )
92- }
87+
9388 }
9489 log .ZDebug (ctx , "sort" , "sort cost time" , time .Since (t ))
9590 messageListCallback .MessageList = messageList
@@ -104,20 +99,60 @@ func (c *Conversation) fetchMessagesWithGapCheck(ctx context.Context, conversati
10499
105100 // Get the number of invalid messages in this batch to recursive fetching from earlier points.
106101 shouldFetchMoreMessagesNum := func (messages []* model_struct.LocalChatLog ) int {
107- if len (messages ) == 0 {
108- return count
109- }
110-
102+ var thisEndSeq int64
111103 // Represents the number of valid messages in the batch
112104 validateMessageNum := 0
113105 for _ , msg := range messages {
114- if msg .Status < constant .MsgStatusHasDeleted {
115- validateMessageNum ++
116- validMessages = append (validMessages , msg )
106+ if msg .Seq != 0 && thisEndSeq == 0 {
107+ thisEndSeq = msg .Seq
108+ }
109+ if isReverse {
110+ if msg .Seq > thisEndSeq && thisEndSeq != 0 {
111+ thisEndSeq = msg .Seq
112+ }
113+
117114 } else {
115+ if msg .Seq < thisEndSeq && msg .Seq != 0 {
116+ thisEndSeq = msg .Seq
117+ }
118+ }
119+ if msg .Status >= constant .MsgStatusHasDeleted {
118120 log .ZDebug (ctx , "this message has been deleted or exception message" , "msg" , msg )
121+ continue
122+ }
123+
124+ validateMessageNum ++
125+ validMessages = append (validMessages , msg )
126+
127+ }
128+ if ! isReverse {
129+ if thisEndSeq != 0 {
130+ c .messagePullForwardEndSeqMap .StoreWithFunc (conversationID , thisEndSeq , func (key string , value int64 ) bool {
131+ lastEndSeq , _ := c .messagePullForwardEndSeqMap .Load (key )
132+ if value < lastEndSeq || lastEndSeq == 0 {
133+ log .ZDebug (ctx , "update the end sequence of the message" , "lastEndSeq" , lastEndSeq , "thisEndSeq" , value )
134+ return true
135+ }
136+ log .ZWarn (ctx , "The end sequence number of the message is more than the last end sequence number" ,
137+ nil , "conversationID" , key , "value" , value , "lastEndSeq" , lastEndSeq )
138+ return false
139+ })
140+ }
141+ } else {
142+ if thisEndSeq != 0 {
143+ c .messagePullReverseEndSeqMap .StoreWithFunc (conversationID , thisEndSeq , func (key string , value int64 ) bool {
144+ lastEndSeq , _ := c .messagePullReverseEndSeqMap .Load (key )
145+ if value > lastEndSeq || lastEndSeq == 0 {
146+ log .ZDebug (ctx , "update the end sequence of the message" , "lastEndSeq" , lastEndSeq , "thisEndSeq" , value )
147+ return true
148+ }
149+ log .ZWarn (ctx , "The end sequence number of the message is less than the last end sequence number" ,
150+ nil , "conversationID" , key , "value" , value , "lastEndSeq" , lastEndSeq )
151+ return false
152+ })
119153 }
120154 }
155+
121156 return count - validateMessageNum
122157 }
123158 getNewStartTime := func (messages []* model_struct.LocalChatLog ) int64 {
@@ -139,11 +174,11 @@ func (c *Conversation) fetchMessagesWithGapCheck(ctx context.Context, conversati
139174 t = time .Now ()
140175 thisStartSeq := c .validateAndFillInternalGaps (ctx , conversationID , isReverse ,
141176 count , startTime , & list , messageListCallback )
142- log .ZDebug (ctx , "internal continuity check" , "cost time" , time .Since (t ))
177+ log .ZDebug (ctx , "internal continuity check" , "cost time" , time .Since (t ), "thisStartSeq" , thisStartSeq )
143178 t = time .Now ()
144179 c .validateAndFillInterBlockGaps (ctx , thisStartSeq , conversationID ,
145180 isReverse , count , startTime , & list , messageListCallback )
146- log .ZDebug (ctx , "between continuity check" , "cost time" , time .Since (t ))
181+ log .ZDebug (ctx , "between continuity check" , "cost time" , time .Since (t ), "thisStartSeq" , thisStartSeq )
147182 t = time .Now ()
148183 c .validateAndFillEndBlockContinuity (ctx , conversationID , isReverse ,
149184 count , startTime , & list , messageListCallback )
@@ -152,8 +187,10 @@ func (c *Conversation) fetchMessagesWithGapCheck(ctx context.Context, conversati
152187 // continue fetching recursively until the valid messages are sufficient or all messages have been fetched.
153188 missingCount := shouldFetchMoreMessagesNum (list )
154189 if missingCount > 0 && ! messageListCallback .IsEnd {
155- log .ZDebug (ctx , "fetch more messages" , "missingCount" , missingCount , "conversationID" , conversationID )
156- missingMessages , err := c .fetchMessagesWithGapCheck (ctx , conversationID , missingCount , getNewStartTime (list ), isReverse , messageListCallback )
190+ newStartTime := getNewStartTime (list )
191+ log .ZDebug (ctx , "fetch more messages" , "missingCount" , missingCount , "conversationID" ,
192+ conversationID , "newStartTime" , newStartTime )
193+ missingMessages , err := c .fetchMessagesWithGapCheck (ctx , conversationID , missingCount , newStartTime , isReverse , messageListCallback )
157194 if err != nil {
158195 return nil , err
159196 }
@@ -164,31 +201,17 @@ func (c *Conversation) fetchMessagesWithGapCheck(ctx context.Context, conversati
164201 return validMessages , nil
165202}
166203
167- func (c * Conversation ) LocalChatLog2MsgStruct (list []* model_struct.LocalChatLog , isReverse bool ) ( int64 , []* sdk_struct.MsgStruct ) {
204+ func (c * Conversation ) LocalChatLog2MsgStruct (list []* model_struct.LocalChatLog ) []* sdk_struct.MsgStruct {
168205 messageList := make ([]* sdk_struct.MsgStruct , 0 , len (list ))
169- var thisEndSeq int64
170206 for _ , v := range list {
171- if v .Seq != 0 && thisEndSeq == 0 {
172- thisEndSeq = v .Seq
173- }
174- if isReverse {
175- if v .Seq > thisEndSeq && thisEndSeq != 0 {
176- thisEndSeq = v .Seq
177- }
178-
179- } else {
180- if v .Seq < thisEndSeq && v .Seq != 0 {
181- thisEndSeq = v .Seq
182- }
183- }
184207 temp := LocalChatLogToMsgStruct (v )
185208
186209 if temp .AttachedInfoElem .IsPrivateChat && temp .SendTime + int64 (temp .AttachedInfoElem .BurnDuration ) < time .Now ().Unix () {
187210 continue
188211 }
189212 messageList = append (messageList , temp )
190213 }
191- return thisEndSeq , messageList
214+ return messageList
192215}
193216
194217func (c * Conversation ) typingStatusUpdate (ctx context.Context , recvID , msgTip string ) error {
0 commit comments