diff --git a/server/im_http.go b/server/im_http.go index 21598de..f2aa80c 100644 --- a/server/im_http.go +++ b/server/im_http.go @@ -11626,7 +11626,9 @@ func broadcastIMConversationSystemMessage(tenantID types.TenantID, msg *types.Me logs.Warn.Printf("im system message: hub route queue full topic=%s seq=%d", msg.Topic, msg.SeqId) } deliverIMConversationSystemMessageToP2PSessions(tenantID, msg, out) - publishRealtimeTopicMessage(msg.Topic, out) + if !publishRealtimeP2PConversationSystemMessage(tenantID, msg.Topic, out) { + publishRealtimeTopicMessage(msg.Topic, out) + } broadcastIMConversationSystemPresence(tenantID, msg, timestamp) } @@ -11672,9 +11674,10 @@ func deliverIMConversationSystemMessageToP2PSessions(tenantID types.TenantID, ms if session.getSub(msg.Topic) != nil { return true } - msgCopy := out.copy() - msgCopy.RcptTo = session.uid.UserId() - msgCopy.Data.Topic = msg.Topic + msgCopy, ok := p2pConversationSystemMessageForUser(out, msg.Topic, session.uid) + if !ok { + return true + } if !session.queueOut(msgCopy) { logs.Warn.Printf("im system message: session queue full topic=%s seq=%d user=%s", msg.Topic, msg.SeqId, session.uid.UserId()) } @@ -11682,6 +11685,43 @@ func deliverIMConversationSystemMessageToP2PSessions(tenantID types.TenantID, ms }) } +func publishRealtimeP2PConversationSystemMessage(tenantID types.TenantID, topic string, out *ServerComMessage) bool { + uid1, uid2, err := types.ParseP2P(topic) + if err != nil { + return false + } + for _, uid := range []types.Uid{uid1, uid2} { + msgCopy, ok := p2pConversationSystemMessageForUser(out, topic, uid) + if !ok { + continue + } + publishRealtimeUserMessage(uid, msgCopy) + } + return true +} + +func p2pConversationSystemMessageForUser(out *ServerComMessage, topic string, uid types.Uid) (*ServerComMessage, bool) { + if out == nil || uid.IsZero() { + return nil, false + } + viewTopic, err := types.P2PNameForUser(uid, topic) + if err != nil || viewTopic == "" { + return nil, false + } + msgCopy := out.copy() + msgCopy.RcptTo = uid.UserId() + if msgCopy.Data != nil { + msgCopy.Data.Topic = viewTopic + } + if msgCopy.Info != nil { + msgCopy.Info.Topic = viewTopic + } + if msgCopy.Pres != nil { + msgCopy.Pres.Topic = viewTopic + } + return msgCopy, true +} + func broadcastIMConversationSystemPresence(tenantID types.TenantID, msg *types.Message, timestamp time.Time) { if globals.hub == nil || msg == nil || msg.SeqId <= 0 || msg.Topic == "" { return diff --git a/server/im_http_test.go b/server/im_http_test.go index 646c4a6..76c2f24 100644 --- a/server/im_http_test.go +++ b/server/im_http_test.go @@ -8349,7 +8349,11 @@ func TestPublishIMConversationSystemNoticeDeliversToOnlineP2PSessions(t *testing if !ok || msg.Data == nil { t.Fatalf("unexpected outbound for %s: %#v", uid.UserId(), raw) } - if msg.Data.Topic != topic || msg.Data.SeqId != 1 { + wantTopic, err := types.P2PNameForUser(uid, topic) + if err != nil { + t.Fatal(err) + } + if msg.Data.Topic != wantTopic || msg.Data.SeqId != 1 { t.Fatalf("unexpected p2p data for %s: %#v", uid.UserId(), msg.Data) } if msg.Data.Head["x-im-template"] != "friend_added" || msg.Data.Head["x-im-type"] != "system_notice" {