88 lines
2.5 KiB
Go
88 lines
2.5 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
|
|
"github.com/tinode/chat/server/logs"
|
|
"github.com/tinode/chat/server/store"
|
|
"github.com/tinode/chat/server/store/types"
|
|
)
|
|
|
|
func startRedisGroupFanout(ctx context.Context) {
|
|
if err := store.StartGroupFanoutSubscriber(ctx, currentNodeID(), func(raw []byte) {
|
|
var msg ServerComMessage
|
|
if err := json.Unmarshal(raw, &msg); err != nil {
|
|
logs.Warn.Println("redis group fanout: malformed message", err)
|
|
return
|
|
}
|
|
if !msg.TenantID.IsValid() || msg.RcptTo == "" || types.GetTopicCat(msg.RcptTo) != types.TopicCatGrp {
|
|
return
|
|
}
|
|
msg.sess = nil
|
|
deliverRedisGroupFanout(&msg)
|
|
}); err != nil {
|
|
logs.Warn.Println("redis group fanout: subscriber disabled", err)
|
|
}
|
|
}
|
|
|
|
func deliverRedisGroupFanout(msg *ServerComMessage) {
|
|
key := types.TopicKey{TenantID: msg.TenantID, Topic: msg.RcptTo}
|
|
if globals.hub.topicGet(key) != nil {
|
|
select {
|
|
case globals.hub.routeSrv <- msg:
|
|
default:
|
|
logs.Err.Println("redis group fanout: hub routeSrv queue full", msg.RcptTo)
|
|
}
|
|
return
|
|
}
|
|
|
|
delivered := 0
|
|
localMembers := redisGroupFanoutMembers(msg.TenantID, msg.RcptTo)
|
|
globals.sessionStore.Range(func(_ string, sess *Session) bool {
|
|
if sess == nil || sess.isCluster() || sess.isProxy() || sess.tenantID != msg.TenantID || sess.sid == msg.SkipSid {
|
|
return true
|
|
}
|
|
_, listedOnline := localMembers[sess.uid.UserId()]
|
|
if !listedOnline && !sess.hasSub(msg.RcptTo) {
|
|
return true
|
|
}
|
|
|
|
msgCopy := msg.copy()
|
|
prepareRedisGroupFanoutMessage(msgCopy, msg.RcptTo)
|
|
if sess.queueOut(msgCopy) {
|
|
delivered++
|
|
} else {
|
|
logs.Warn.Printf("redis group fanout: session queue full, sid=%s topic=%s", sess.sid, msg.RcptTo)
|
|
}
|
|
return true
|
|
})
|
|
if delivered == 0 {
|
|
logs.Info.Println("redis group fanout: no local subscribers for unloaded topic", msg.RcptTo)
|
|
}
|
|
}
|
|
|
|
func redisGroupFanoutMembers(tenantID types.TenantID, topic string) map[string]struct{} {
|
|
uids, err := store.GetTopicNodeOnlineMembers(tenantID, topic, currentNodeID())
|
|
if err != nil {
|
|
logs.Warn.Println("redis group fanout: failed to read local online members", topic, err)
|
|
return nil
|
|
}
|
|
members := make(map[string]struct{}, len(uids))
|
|
for _, uid := range uids {
|
|
members[uid.UserId()] = struct{}{}
|
|
}
|
|
return members
|
|
}
|
|
|
|
func prepareRedisGroupFanoutMessage(msg *ServerComMessage, topic string) {
|
|
switch {
|
|
case msg.Data != nil && msg.Data.Topic == "":
|
|
msg.Data.Topic = topic
|
|
case msg.Pres != nil && msg.Pres.Topic == "":
|
|
msg.Pres.Topic = topic
|
|
case msg.Info != nil && msg.Info.Topic == "":
|
|
msg.Info.Topic = topic
|
|
}
|
|
}
|