fix: broadcast offline presence as soon as the last session drops
Logout left contacts seeing the user online until the me topic idle timeout fired presUsersOfInterest(off), and even that frame was silently dropped for p2p routes by handlePresence's xoriginal check. MarkUserOffline now returns the remaining session count and cleanUp notifies users of interest immediately once it hits zero; p2p presence is delivered straight to the contact's me topic, mirroring presSingleUserOffline. Zentao #22.
This commit is contained in:
@@ -278,6 +278,7 @@ func (t *Topic) presUsersOfInterest(what, ua string) {
|
||||
|
||||
// Publish user's update to his/her users of interest on their 'me' topic while user's 'me' topic is offline
|
||||
// Case A: user is being deleted, "gone".
|
||||
// Case B: user's last session just disconnected, "off".
|
||||
func presUsersOfInterestOffline(tenantID types.TenantID, uid types.Uid, subs []types.Subscription, what string) {
|
||||
// Push update to subscriptions
|
||||
for i := range subs {
|
||||
@@ -286,6 +287,37 @@ func presUsersOfInterestOffline(tenantID types.TenantID, uid types.Uid, subs []t
|
||||
continue
|
||||
}
|
||||
|
||||
if types.GetTopicCat(subs[i].Topic) == types.TopicCatP2P {
|
||||
// A presence request routed to the p2p topic would be silently
|
||||
// dropped: the topic is loaded under its canonical p2p name while
|
||||
// the request carries Topic="me" (notifyOnOrSkip), so the
|
||||
// xoriginal check in handlePresence rejects it. Deliver to the
|
||||
// contact's 'me' topic directly instead, the same way
|
||||
// presSingleUserOffline does.
|
||||
uid1, uid2, err := types.ParseP2P(subs[i].Topic)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
contact := uid1
|
||||
if contact == uid {
|
||||
contact = uid2
|
||||
}
|
||||
if contact == uid || contact.IsZero() {
|
||||
continue
|
||||
}
|
||||
routeUserServerMessage(contact, &ServerComMessage{
|
||||
TenantID: tenantID,
|
||||
Pres: &MsgServerPres{
|
||||
Topic: notifyOn,
|
||||
What: what,
|
||||
Src: subs[i].Topic,
|
||||
WantReply: false,
|
||||
},
|
||||
RcptTo: contact.UserId(),
|
||||
})
|
||||
continue
|
||||
}
|
||||
|
||||
globals.hub.routeSrv <- &ServerComMessage{
|
||||
TenantID: tenantID,
|
||||
Pres: &MsgServerPres{
|
||||
|
||||
+11
-1
@@ -465,9 +465,19 @@ func (s *Session) cleanUp(expired bool) {
|
||||
if err := store.SecurityAudit.CloseLogin(s.tenantID, s.sid, time.Now().UTC().Round(time.Millisecond)); err != nil {
|
||||
logs.Warn.Println("s.cleanUp: failed to close login audit", s.uid, err, s.sid)
|
||||
}
|
||||
if err := store.MarkUserOffline(s.tenantID, s.uid, s.sid, currentNodeID()); err != nil {
|
||||
remaining, err := store.MarkUserOffline(s.tenantID, s.uid, s.sid, currentNodeID())
|
||||
if err != nil {
|
||||
logs.Warn.Println("s.cleanUp: failed to mark user offline", s.uid, err, s.sid)
|
||||
}
|
||||
// 最后一个会话已断开(如主动登出、断网):立即通知关注者该用户离线,
|
||||
// 不再等待 me topic 闲置超时后的 presUsersOfInterest("off")。
|
||||
if err == nil && remaining == 0 && globals.hub != nil {
|
||||
if uoi, uerr := store.Users.GetSubs(s.tenantID, s.uid); uerr == nil {
|
||||
presUsersOfInterestOffline(s.tenantID, s.uid, uoi, "off")
|
||||
} else {
|
||||
logs.Warn.Println("s.cleanUp: failed to notify users of interest", s.uid, uerr, s.sid)
|
||||
}
|
||||
}
|
||||
}
|
||||
s.purgeChannels()
|
||||
s.inflightReqs.Wait()
|
||||
|
||||
@@ -109,7 +109,7 @@ type groupMessageRuntime interface {
|
||||
InvalidateHotMessages(tenantID types.TenantID, topic string, ranges []types.Range) error
|
||||
MarkUserOnline(tenantID types.TenantID, uid types.Uid, sid, nodeID, device, userAgent string) error
|
||||
RefreshUserPresence(tenantID types.TenantID, uid types.Uid, sid, nodeID string) error
|
||||
MarkUserOffline(tenantID types.TenantID, uid types.Uid, sid, nodeID string) error
|
||||
MarkUserOffline(tenantID types.TenantID, uid types.Uid, sid, nodeID string) (int, error)
|
||||
GetUserPresence(tenantID types.TenantID, uid types.Uid) (*UserPresence, error)
|
||||
MarkTopicMemberOnline(tenantID types.TenantID, topic string, uid types.Uid, nodeID string) error
|
||||
MarkTopicMemberOffline(tenantID types.TenantID, topic string, uid types.Uid, nodeID string) error
|
||||
@@ -300,9 +300,9 @@ func RefreshUserPresence(tenantID types.TenantID, uid types.Uid, sid, nodeID str
|
||||
return groupMsgRuntime.RefreshUserPresence(tenantID, uid, sid, nodeID)
|
||||
}
|
||||
|
||||
func MarkUserOffline(tenantID types.TenantID, uid types.Uid, sid, nodeID string) error {
|
||||
func MarkUserOffline(tenantID types.TenantID, uid types.Uid, sid, nodeID string) (int, error) {
|
||||
if groupMsgRuntime == nil || !tenantID.IsValid() || uid.IsZero() || sid == "" {
|
||||
return nil
|
||||
return 0, nil
|
||||
}
|
||||
return groupMsgRuntime.MarkUserOffline(tenantID, uid, sid, nodeID)
|
||||
}
|
||||
@@ -939,7 +939,7 @@ func (r *redisGroupMessageRuntime) RefreshUserPresence(tenantID types.TenantID,
|
||||
return err
|
||||
}
|
||||
|
||||
func (r *redisGroupMessageRuntime) MarkUserOffline(tenantID types.TenantID, uid types.Uid, sid, nodeID string) error {
|
||||
func (r *redisGroupMessageRuntime) MarkUserOffline(tenantID types.TenantID, uid types.Uid, sid, nodeID string) (int, error) {
|
||||
if nodeID == "" {
|
||||
nodeID = "local"
|
||||
}
|
||||
@@ -956,11 +956,11 @@ func (r *redisGroupMessageRuntime) MarkUserOffline(tenantID types.TenantID, uid
|
||||
pipe.SRem(ctx, sessionsKey, member)
|
||||
pipe.SRem(ctx, nodeKey, uid.UserId())
|
||||
if _, err := pipe.Exec(ctx); err != nil {
|
||||
return err
|
||||
return 0, err
|
||||
}
|
||||
count, err := r.client.SCard(ctx, sessionsKey).Result()
|
||||
if err != nil {
|
||||
return err
|
||||
return 0, err
|
||||
}
|
||||
if count > 0 {
|
||||
pipe = r.client.Pipeline()
|
||||
@@ -968,14 +968,14 @@ func (r *redisGroupMessageRuntime) MarkUserOffline(tenantID types.TenantID, uid
|
||||
pipe.Expire(ctx, userKey, ttl)
|
||||
pipe.Expire(ctx, sessionsKey, ttl)
|
||||
_, err = pipe.Exec(ctx)
|
||||
return err
|
||||
return int(count), err
|
||||
}
|
||||
pipe = r.client.Pipeline()
|
||||
pipe.HSet(ctx, userKey, "online", "0", "session_count", 0, "node_id", nodeID, "last_seen_at", now.Format(time.RFC3339Nano))
|
||||
pipe.Expire(ctx, userKey, ttl)
|
||||
pipe.Del(ctx, sessionsKey)
|
||||
_, err = pipe.Exec(ctx)
|
||||
return err
|
||||
return 0, err
|
||||
}
|
||||
|
||||
func (r *redisGroupMessageRuntime) GetUserPresence(tenantID types.TenantID, uid types.Uid) (*UserPresence, error) {
|
||||
|
||||
@@ -114,8 +114,8 @@ func (r *testGroupMessageRuntime) RefreshUserPresence(types.TenantID, types.Uid,
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *testGroupMessageRuntime) MarkUserOffline(types.TenantID, types.Uid, string, string) error {
|
||||
return nil
|
||||
func (r *testGroupMessageRuntime) MarkUserOffline(types.TenantID, types.Uid, string, string) (int, error) {
|
||||
return 0, nil
|
||||
}
|
||||
|
||||
func (r *testGroupMessageRuntime) GetUserPresence(types.TenantID, types.Uid) (*UserPresence, error) {
|
||||
|
||||
Reference in New Issue
Block a user