fix: 修复P2P主题别名订阅的兼容性问题
1. 订阅重复检查新增Original与RcptTo匹配判断,避免别名订阅误判已订阅 2. 重构订阅get查询处理逻辑,提取为独立函数 3. 针对P2P别名订阅场景返回200响应而非304,帮助客户端完成别名到标准主题名的重命名,避免发布失败 4. 优化群摘要接口的权限校验和异常处理逻辑
This commit is contained in:
+25
-2
@@ -5944,17 +5944,40 @@ func serveIMGroupSummary(wrt http.ResponseWriter, req *http.Request, topic strin
|
|||||||
writeIMError(wrt, req, http.StatusMethodNotAllowed, 30002, imReasonMethodNotAllowed, "method not allowed", nil)
|
writeIMError(wrt, req, http.StatusMethodNotAllowed, 30002, imReasonMethodNotAllowed, "method not allowed", nil)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
principal, selfSub, ok := requireIMGroupMember(wrt, req, topic, true)
|
principal, ok := authenticateIMHTTPBearer(wrt, req)
|
||||||
if !ok {
|
if !ok {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
if types.GetTopicCat(topic) != types.TopicCatGrp {
|
||||||
|
writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "topic"})
|
||||||
|
return
|
||||||
|
}
|
||||||
|
// 已解散的群按文档约定对客户端表现为不存在(RESOURCE_NOT_FOUND),不暴露解散状态。
|
||||||
|
if settings, err := store.GroupBusiness.GetSettings(principal.TenantID, topic); err == nil &&
|
||||||
|
settings != nil && settings.GroupStatus == types.GroupStatusDissolved {
|
||||||
|
writeIMError(wrt, req, http.StatusNotFound, 30004, imReasonResourceNotFound, "group not found", nil)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
selfSub, err := store.Subs.Get(principal.TenantID, topic, principal.Uid, false)
|
||||||
|
if err != nil {
|
||||||
|
writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
access := types.ModeNone
|
||||||
|
if selfSub != nil {
|
||||||
|
access = selfSub.ModeGiven & selfSub.ModeWant
|
||||||
|
}
|
||||||
|
if selfSub == nil || !access.IsJoiner() || !access.IsReader() {
|
||||||
|
writeIMError(wrt, req, http.StatusForbidden, 30005, imReasonPermissionDenied, "topic permission required", nil)
|
||||||
|
return
|
||||||
|
}
|
||||||
requests, _, _ := store.GroupBusiness.ListJoinRequests(principal.TenantID, topic, types.GroupJoinRequestStatusPending, "", 100)
|
requests, _, _ := store.GroupBusiness.ListJoinRequests(principal.TenantID, topic, types.GroupJoinRequestStatusPending, "", 100)
|
||||||
announcement, _ := store.GroupBusiness.LatestAnnouncement(principal.TenantID, topic)
|
announcement, _ := store.GroupBusiness.LatestAnnouncement(principal.TenantID, topic)
|
||||||
writeIMSuccess(wrt, req, map[string]any{
|
writeIMSuccess(wrt, req, map[string]any{
|
||||||
"topic": topic,
|
"topic": topic,
|
||||||
"pending_join_request_count": len(requests),
|
"pending_join_request_count": len(requests),
|
||||||
"announcement": groupAnnouncementPayload(announcement),
|
"announcement": groupAnnouncementPayload(announcement),
|
||||||
"permissions": groupPermissions(selfSub.ModeGiven & selfSub.ModeWant),
|
"permissions": groupPermissions(access),
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
+8
-1
@@ -749,7 +749,14 @@ func (s *Session) subscribe(msg *ClientComMessage) {
|
|||||||
|
|
||||||
s.inflightReqs.Add(1)
|
s.inflightReqs.Add(1)
|
||||||
// Session can subscribe to topic on behalf of a single user at a time.
|
// Session can subscribe to topic on behalf of a single user at a time.
|
||||||
if sub := s.getSub(msg.RcptTo); sub != nil {
|
// Fast-path the duplicate attach only when the client used the canonical
|
||||||
|
// topic name itself. A p2p topic may also be addressed by an alias (the
|
||||||
|
// peer's user id, e.g. "usrXXX") while the session is already attached to
|
||||||
|
// the canonical topic ("p2pXXX"): forwarding it to the topic lets the
|
||||||
|
// client complete the alias->canonical rename with a 200 reply instead of
|
||||||
|
// a 304 that would leave the client-side alias topic object unattached
|
||||||
|
// (publishes on it then fail with "Cannot publish on inactive topic").
|
||||||
|
if sub := s.getSub(msg.RcptTo); sub != nil && msg.Original == msg.RcptTo {
|
||||||
s.queueOut(InfoAlreadySubscribed(msg.Id, msg.Original, msg.Timestamp))
|
s.queueOut(InfoAlreadySubscribed(msg.Id, msg.Original, msg.Timestamp))
|
||||||
s.inflightReqs.Done()
|
s.inflightReqs.Done()
|
||||||
} else {
|
} else {
|
||||||
|
|||||||
+46
-12
@@ -353,7 +353,30 @@ func (t *Topic) registerSession(msg *ClientComMessage) {
|
|||||||
// but there is a gap between topic creation/un-pausing and processing the
|
// but there is a gap between topic creation/un-pausing and processing the
|
||||||
// first subscription request, before the topic is linked to session: a client
|
// first subscription request, before the topic is linked to session: a client
|
||||||
// may send several subscription requests in that gap.
|
// may send several subscription requests in that gap.
|
||||||
|
//
|
||||||
|
// A p2p topic may be addressed by an alias (the peer's user id, e.g. "usrXXX")
|
||||||
|
// while the session is already attached to the same topic under its canonical
|
||||||
|
// name ("p2pXXX"). Replying 304 here leaves the client-side topic object created
|
||||||
|
// for the alias name unattached: the SDK treats any ctrl code >= 300 as failure
|
||||||
|
// and skips the alias->canonical rename, so publishes on it fail with
|
||||||
|
// "Cannot publish on inactive topic". Reply 200 with the canonical name and the
|
||||||
|
// effective access mode so the client can complete the rename instead.
|
||||||
|
asUid := types.ParseUserId(msg.AsUser)
|
||||||
|
if pud, ok := t.perUser[asUid]; ok && t.cat == types.TopicCatP2P && msg.Original != t.name {
|
||||||
|
msg.sess.queueOut(NoErrParams(msg.Id, t.name, types.TimeNow(), map[string]any{
|
||||||
|
"acs": &MsgAccessMode{
|
||||||
|
Want: pud.modeWant.String(),
|
||||||
|
Given: pud.modeGiven.String(),
|
||||||
|
Mode: (pud.modeGiven & pud.modeWant).String(),
|
||||||
|
},
|
||||||
|
}))
|
||||||
|
// The client-side topic object created for the alias name is fresh:
|
||||||
|
// serve the {get} query it sent along with the subscription, otherwise
|
||||||
|
// the reopened conversation would come up without desc/subs/history.
|
||||||
|
t.processSubGetQuery(msg, asUid, false, auth.Level(msg.AuthLvl))
|
||||||
|
} else {
|
||||||
msg.sess.queueOut(InfoAlreadySubscribed(msg.Id, msg.Original, msg.Timestamp))
|
msg.sess.queueOut(InfoAlreadySubscribed(msg.Id, msg.Original, msg.Timestamp))
|
||||||
|
}
|
||||||
} else {
|
} else {
|
||||||
// The topic is alive, so stop the kill timer, if it's ticking. We don't want the topic to die
|
// The topic is alive, so stop the kill timer, if it's ticking. We don't want the topic to die
|
||||||
// while processing the call.
|
// while processing the call.
|
||||||
@@ -879,60 +902,71 @@ func (t *Topic) handleSubscription(msg *ClientComMessage) error {
|
|||||||
}
|
}
|
||||||
|
|
||||||
msgsub := msg.Sub
|
msgsub := msg.Sub
|
||||||
getWhat := 0
|
|
||||||
if msgsub.Get != nil {
|
if msgsub.Get != nil {
|
||||||
getWhat = parseMsgClientMeta(msgsub.Get.What)
|
// Deliver the data requested by the {get} query attached to the subscription.
|
||||||
|
t.processSubGetQuery(msg, asUid, asChan, authLevel)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// processSubGetQuery responds to the {get} query attached to a subscription
|
||||||
|
// request (desc/sub/tags/cred/aux/data/del) by sending the answers as separate
|
||||||
|
// {meta}/{data} packets.
|
||||||
|
func (t *Topic) processSubGetQuery(msg *ClientComMessage, asUid types.Uid, asChan bool, authLevel auth.Level) {
|
||||||
|
msgsub := msg.Sub
|
||||||
|
if msgsub == nil || msgsub.Get == nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
getWhat := parseMsgClientMeta(msgsub.Get.What)
|
||||||
if getWhat&constMsgMetaDesc != 0 {
|
if getWhat&constMsgMetaDesc != 0 {
|
||||||
// Send get.desc as a {meta} packet.
|
// Send get.desc as a {meta} packet.
|
||||||
if err := t.replyGetDesc(msg.sess, asUid, asChan, msgsub.Get.Desc, msg); err != nil {
|
if err := t.replyGetDesc(msg.sess, asUid, asChan, msgsub.Get.Desc, msg); err != nil {
|
||||||
logs.Warn.Printf("topic[%s] handleSubscription Get.Desc failed: %v sid=%s", t.name, err, msg.sess.sid)
|
logs.Warn.Printf("topic[%s] Get.Desc failed: %v sid=%s", t.name, err, msg.sess.sid)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
if getWhat&constMsgMetaSub != 0 {
|
if getWhat&constMsgMetaSub != 0 {
|
||||||
// Send get.sub response as a separate {meta} packet
|
// Send get.sub response as a separate {meta} packet
|
||||||
if err := t.replyGetSub(msg.sess, asUid, authLevel, asChan, msg); err != nil {
|
if err := t.replyGetSub(msg.sess, asUid, authLevel, asChan, msg); err != nil {
|
||||||
logs.Warn.Printf("topic[%s] handleSubscription Get.Sub failed: %v sid=%s", t.name, err, msg.sess.sid)
|
logs.Warn.Printf("topic[%s] Get.Sub failed: %v sid=%s", t.name, err, msg.sess.sid)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
if getWhat&constMsgMetaTags != 0 {
|
if getWhat&constMsgMetaTags != 0 {
|
||||||
// Send get.tags response as a separate {meta} packet
|
// Send get.tags response as a separate {meta} packet
|
||||||
if err := t.replyGetTags(msg.sess, asUid, msg); err != nil {
|
if err := t.replyGetTags(msg.sess, asUid, msg); err != nil {
|
||||||
logs.Warn.Printf("topic[%s] handleSubscription Get.Tags failed: %v sid=%s", t.name, err, msg.sess.sid)
|
logs.Warn.Printf("topic[%s] Get.Tags failed: %v sid=%s", t.name, err, msg.sess.sid)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
if getWhat&constMsgMetaCred != 0 {
|
if getWhat&constMsgMetaCred != 0 {
|
||||||
// Send get.tags response as a separate {meta} packet
|
// Send get.cred response as a separate {meta} packet
|
||||||
if err := t.replyGetCreds(msg.sess, asUid, msg); err != nil {
|
if err := t.replyGetCreds(msg.sess, asUid, msg); err != nil {
|
||||||
logs.Warn.Printf("topic[%s] handleSubscription Get.Cred failed: %v sid=%s", t.name, err, msg.sess.sid)
|
logs.Warn.Printf("topic[%s] Get.Cred failed: %v sid=%s", t.name, err, msg.sess.sid)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
if getWhat&constMsgMetaAux != 0 {
|
if getWhat&constMsgMetaAux != 0 {
|
||||||
// Send get.aux response as a separate {meta} packet
|
// Send get.aux response as a separate {meta} packet
|
||||||
if err := t.replyGetAux(msg.sess, asUid, msg); err != nil {
|
if err := t.replyGetAux(msg.sess, asUid, msg); err != nil {
|
||||||
logs.Warn.Printf("topic[%s] handleSubscription Get.Aux failed: %v sid=%s", t.name, err, msg.sess.sid)
|
logs.Warn.Printf("topic[%s] Get.Aux failed: %v sid=%s", t.name, err, msg.sess.sid)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
if getWhat&constMsgMetaData != 0 {
|
if getWhat&constMsgMetaData != 0 {
|
||||||
// Send get.data response as {data} packets
|
// Send get.data response as {data} packets
|
||||||
if err := t.replyGetData(msg.sess, asUid, asChan, msgsub.Get.Data, msg); err != nil {
|
if err := t.replyGetData(msg.sess, asUid, asChan, msgsub.Get.Data, msg); err != nil {
|
||||||
logs.Warn.Printf("topic[%s] handleSubscription Get.Data failed: %v sid=%s", t.name, err, msg.sess.sid)
|
logs.Warn.Printf("topic[%s] Get.Data failed: %v sid=%s", t.name, err, msg.sess.sid)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
if getWhat&constMsgMetaDel != 0 {
|
if getWhat&constMsgMetaDel != 0 {
|
||||||
// Send get.del response as a separate {meta} packet
|
// Send get.del response as a separate {meta} packet
|
||||||
if err := t.replyGetDel(msg.sess, asUid, msgsub.Get.Del, msg); err != nil {
|
if err := t.replyGetDel(msg.sess, asUid, msgsub.Get.Del, msg); err != nil {
|
||||||
logs.Warn.Printf("topic[%s] handleSubscription Get.Del failed: %v sid=%s", t.name, err, msg.sess.sid)
|
logs.Warn.Printf("topic[%s] Get.Del failed: %v sid=%s", t.name, err, msg.sess.sid)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
return nil
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// handleLeaveRequest processes a session leave request.
|
// handleLeaveRequest processes a session leave request.
|
||||||
|
|||||||
Reference in New Issue
Block a user