fix: sync topic sequence after server data
This commit is contained in:
@@ -680,6 +680,12 @@ func (t *Topic) handleServerMsg(msg *ServerComMessage) {
|
|||||||
if msg.Pres != nil {
|
if msg.Pres != nil {
|
||||||
t.handlePresence(msg)
|
t.handlePresence(msg)
|
||||||
} else if msg.Info != nil || msg.Data != nil {
|
} else if msg.Info != nil || msg.Data != nil {
|
||||||
|
if msg.Data != nil && msg.Data.SeqId > t.lastID {
|
||||||
|
t.lastID = msg.Data.SeqId
|
||||||
|
if !msg.Data.Timestamp.IsZero() {
|
||||||
|
t.touched = msg.Data.Timestamp
|
||||||
|
}
|
||||||
|
}
|
||||||
if msg.Info != nil && msg.Info.What == "group" {
|
if msg.Info != nil && msg.Info.What == "group" {
|
||||||
// Group management event published by the HTTP or Kafka admin
|
// Group management event published by the HTTP or Kafka admin
|
||||||
// interface: update in-memory topic state (cached permissions,
|
// interface: update in-memory topic state (cached permissions,
|
||||||
|
|||||||
@@ -1866,6 +1866,45 @@ func TestHandleBroadcastPresMe(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestHandleServerDataAdvancesLastID(t *testing.T) {
|
||||||
|
topicName := "p2pZAAAAAAAAABAAAAAAAAAAA"
|
||||||
|
helper := TopicTestHelper{}
|
||||||
|
helper.setUp(t, 2, types.TopicCatP2P, topicName, true)
|
||||||
|
defer helper.tearDown()
|
||||||
|
|
||||||
|
timestamp := time.Now().UTC().Round(time.Millisecond)
|
||||||
|
helper.topic.handleServerMsg(&ServerComMessage{
|
||||||
|
TenantID: types.TenantID(1),
|
||||||
|
RcptTo: topicName,
|
||||||
|
PreserveTopicName: true,
|
||||||
|
Data: &MsgServerData{
|
||||||
|
Topic: topicName,
|
||||||
|
From: types.ZeroUid.UserId(),
|
||||||
|
Timestamp: timestamp,
|
||||||
|
SeqId: 1,
|
||||||
|
Head: map[string]any{"x-im-type": "system_notice"},
|
||||||
|
Content: map[string]any{"text": "你们已成为好友,现在可以开始聊天了。"},
|
||||||
|
},
|
||||||
|
})
|
||||||
|
helper.finish()
|
||||||
|
|
||||||
|
if helper.topic.lastID != 1 {
|
||||||
|
t.Fatalf("Topic.lastID: expected server data to advance to 1, got %d", helper.topic.lastID)
|
||||||
|
}
|
||||||
|
if !helper.topic.touched.Equal(timestamp) {
|
||||||
|
t.Fatalf("Topic.touched: expected %s, got %s", timestamp, helper.topic.touched)
|
||||||
|
}
|
||||||
|
for i, result := range helper.results {
|
||||||
|
if len(result.messages) != 1 {
|
||||||
|
t.Fatalf("session %d expected one server data message, got %d", i, len(result.messages))
|
||||||
|
}
|
||||||
|
msg := result.messages[0].(*ServerComMessage)
|
||||||
|
if msg.Data == nil || msg.Data.SeqId != 1 {
|
||||||
|
t.Fatalf("session %d received unexpected message: %#v", i, msg)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func TestHandleBroadcastPresInactiveTopic(t *testing.T) {
|
func TestHandleBroadcastPresInactiveTopic(t *testing.T) {
|
||||||
topicName := "usrMe"
|
topicName := "usrMe"
|
||||||
numUsers := 1
|
numUsers := 1
|
||||||
|
|||||||
Reference in New Issue
Block a user