HyperNet/Messaging
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

✨ Disconnect websocket auto cleaning up subscribed channels

Browse Source
This commit is contained in:
LittleSheep 2025-03-01 18:02:55 +08:00
parent e3a4988ccf
commit eea9b85745
3 changed files with 31 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
pkg/internal/grpc/services.go
View File

@ -6,12 +6,25 @@ import (
"git.solsynth.dev/hypernet/nexus/pkg/nex"
"git.solsynth.dev/hypernet/nexus/pkg/proto"
jsoniter "github.com/json-iterator/go"
"github.com/rs/zerolog/log"
"git.solsynth.dev/hypernet/messaging/pkg/internal/database"
"git.solsynth.dev/hypernet/messaging/pkg/internal/services"
)
func (v *Server) BroadcastEvent(ctx context.Context, in *proto.EventInfo) (*proto.EventResponse, error) {
log.Debug().Str("event", in.GetEvent()).
Msg("Got a broadcasting event...")
switch in.GetEvent() {
// Clear the subscribed channel
case "ws.client.unregister":
// Update user last seen at
data := nex.DecodeMap(in.GetData())
id := data["id"].(string)
services.UnsubscribeAllWithClient(id)
log.Info().Str("client", id).Msg("Client unregistered, cleaning up subscribed channels...")
// Account recycle
case "deletion":
data := nex.DecodeMap(in.GetData())
resType, ok := data["type"].(string)

2
pkg/internal/grpc/stream.go
View File

@ -76,7 +76,7 @@ func (v *Server) PushStream(_ context.Context, request *proto.PushStreamRequest)
action := strings.Split(in.Action, ".")[1]
switch action {
case "subscribe":
services.SubscribeChannel(uint(request.GetUserId()), data.ChannelID)
services.SubscribeChannel(uint(request.GetUserId()), data.ChannelID, request.GetClientId())
case "unsubscribe":
services.UnsubscribeChannel(uint(request.GetUserId()), data.ChannelID)
case "unsubscribeAll":

22
pkg/internal/services/subscribe.go
View File

@ -2,8 +2,8 @@ package services
import "sync"
// ChannelID -> UserID
var subscribeInfo = make(map[uint]map[uint]bool)
// ChannelID -> UserID -> Client ID
var subscribeInfo = make(map[uint]map[uint]string)
var subscribeLock sync.Mutex
// If user subscribed to a channel
@ -19,13 +19,13 @@ func CheckSubscribed(UserID uint, ChannelID uint) bool {
return false
}
func SubscribeChannel(userId uint, channelId uint) {
func SubscribeChannel(userId uint, channelId uint, clientId string) {
subscribeLock.Lock()
defer subscribeLock.Unlock()
if _, ok := subscribeInfo[channelId]; !ok {
subscribeInfo[channelId] = make(map[uint]bool)
subscribeInfo[channelId] = make(map[uint]string)
}
subscribeInfo[channelId][userId] = true
subscribeInfo[channelId][userId] = clientId
}
func UnsubscribeChannel(userId uint, channelId uint) {
@ -49,3 +49,15 @@ func UnsubscribeAllWithChannels(channelId uint) {
defer subscribeLock.Unlock()
delete(subscribeInfo, channelId)
}
func UnsubscribeAllWithClient(clientId string) {
subscribeLock.Lock()
defer subscribeLock.Unlock()
for _, v := range subscribeInfo {
for k, item := range v {
if item == clientId {
delete(v, k)
}
}
}
}