diff --git a/server/im_http.go b/server/im_http.go index 7d28c49..b655284 100644 --- a/server/im_http.go +++ b/server/im_http.go @@ -9028,6 +9028,47 @@ func broadcastHTTPForwardMessage(tenantID types.TenantID, msg *types.Message) { default: logs.Warn.Printf("im message forward: hub route queue full topic=%s seq=%d", msg.Topic, msg.SeqId) } + broadcastHTTPForwardPresence(tenantID, msg, timestamp) +} + +func broadcastHTTPForwardPresence(tenantID types.TenantID, msg *types.Message, timestamp time.Time) { + if globals.hub == nil || msg == nil || msg.SeqId <= 0 || msg.Topic == "" { + return + } + uid1, uid2, err := types.ParseP2P(msg.Topic) + if err != nil { + return + } + from := types.ParseUid(msg.From).UserId() + for _, rcpt := range []struct { + uid types.Uid + src string + }{ + {uid: uid1, src: uid2.UserId()}, + {uid: uid2, src: uid1.UserId()}, + } { + actor := from + if actor == rcpt.uid.UserId() { + actor = "" + } + select { + case globals.hub.routeSrv <- &ServerComMessage{ + TenantID: tenantID, + RcptTo: rcpt.uid.UserId(), + Timestamp: timestamp, + Pres: &MsgServerPres{ + Topic: "me", + What: "msg", + Src: rcpt.src, + SeqId: msg.SeqId, + AcsActor: actor, + SkipTopic: msg.Topic, + }, + }: + default: + logs.Warn.Printf("im message forward: hub presence queue full topic=%s seq=%d user=%s", msg.Topic, msg.SeqId, rcpt.uid.UserId()) + } + } } func forwardedMessageHead(msg *types.Message, clientMid string) types.KVMap { diff --git a/server/im_http_test.go b/server/im_http_test.go index 4f14cc6..ad5098f 100644 --- a/server/im_http_test.go +++ b/server/im_http_test.go @@ -4793,7 +4793,7 @@ func TestServeIMMessageForwardKafkaAssignsSeqAndBroadcasts(t *testing.T) { topics := mock_store.NewMockTopicsPersistenceInterface(ctrl) messages := mock_store.NewMockMessagesPersistenceInterface(ctrl) kafkaBus = &kafkaRuntime{} - globals.hub = &Hub{routeSrv: make(chan *ServerComMessage, 1)} + globals.hub = &Hub{routeSrv: make(chan *ServerComMessage, 3)} store.Subs = subs store.Topics = topics store.Messages = messages @@ -4883,6 +4883,24 @@ func TestServeIMMessageForwardKafkaAssignsSeqAndBroadcasts(t *testing.T) { default: t.Fatal("expected forwarded message to be broadcast") } + presByUser := make(map[string]*MsgServerPres) + for range 2 { + select { + case delivered := <-globals.hub.routeSrv: + if delivered.Pres == nil { + t.Fatalf("unexpected non-presence message: %#v", delivered) + } + presByUser[delivered.RcptTo] = delivered.Pres + default: + t.Fatal("expected forwarded message presence notifications") + } + } + if pres := presByUser[types.Uid(42).UserId()]; pres == nil || pres.Topic != "me" || pres.What != "msg" || pres.Src != targetUser.UserId() || pres.SeqId != 35 || pres.AcsActor != "" || pres.SkipTopic != targetTopic { + t.Fatalf("unexpected sender presence: %#v", pres) + } + if pres := presByUser[targetUser.UserId()]; pres == nil || pres.Topic != "me" || pres.What != "msg" || pres.Src != types.Uid(42).UserId() || pres.SeqId != 35 || pres.AcsActor != types.Uid(42).UserId() || pres.SkipTopic != targetTopic { + t.Fatalf("unexpected receiver presence: %#v", pres) + } } func TestServeIMMessageForwardBundleSavesForwardBundle(t *testing.T) {