diff --git a/wshandler/wshandler.go b/wshandler/wshandler.go index 6836b77..ae73ddf 100644 --- a/wshandler/wshandler.go +++ b/wshandler/wshandler.go @@ -8,7 +8,6 @@ import ( "io" "net/http" "strings" - "sync" "go.mongodb.org/mongo-driver/bson/primitive" "repositories.action2quare.com/ayo/gocommon" @@ -52,12 +51,11 @@ type CommandMessage struct { } type subhandler struct { + name string authCache *gocommon.AuthCollection redisMsgChanName string redisCmdChanName string redisSync *redis.Client - connsLock sync.Mutex - connectedAlias map[string]bool connInOutChan chan *wsconn deliveryChan chan any @@ -101,11 +99,11 @@ func NewWebsocketHandler[T any](authglobal gocommon.AuthCollectionGlobal, receiv authCaches := make(map[string]*subhandler) for _, region := range authglobal.Regions() { sh := &subhandler{ + name: region, authCache: authglobal.Get(region), redisMsgChanName: fmt.Sprintf("_wsh_msg_%s", region), redisCmdChanName: fmt.Sprintf("_wsh_cmd_%s", region), redisSync: redisSync, - connectedAlias: make(map[string]bool), connInOutChan: make(chan *wsconn), deliveryChan: make(chan any, 1000), callReceiver: func(accid primitive.ObjectID, alias string, r io.Reader) { @@ -125,13 +123,6 @@ func NewWebsocketHandler[T any](authglobal gocommon.AuthCollectionGlobal, receiv func (ws *WebsocketHandler) Destructor() { } -func (ws *WebsocketHandler) IsConnected(region string, alias string) bool { - if sh := ws.authCaches[region]; sh != nil { - return sh.connected(alias) - } - return false -} - func (ws *WebsocketHandler) RegisterHandlers(ctx context.Context, serveMux *http.ServeMux, prefix string) error { for region, sh := range ws.authCaches { if region == "default" { @@ -150,23 +141,12 @@ func (ws *WebsocketHandler) RegisterHandlers(ctx context.Context, serveMux *http return nil } -func (sh *subhandler) setConnected(alias string, connected bool) { - sh.connsLock.Lock() - defer sh.connsLock.Unlock() - - if connected { - sh.connectedAlias[alias] = true - } else { - delete(sh.connectedAlias, alias) +func (ws *WebsocketHandler) GetState(region string, accid primitive.ObjectID) string { + state, err := ws.RedisSync.HGet(context.Background(), region, accid.Hex()).Result() + if err == redis.Nil { + return "" } -} - -func (sh *subhandler) connected(alias string) bool { - sh.connsLock.Lock() - defer sh.connsLock.Unlock() - - _, ok := sh.connectedAlias[alias] - return ok + return state } func (sh *subhandler) mainLoop(ctx context.Context) { @@ -271,13 +251,11 @@ func (sh *subhandler) mainLoop(ctx context.Context) { case c := <-sh.connInOutChan: if c.Conn == nil { delete(entireConns, c.alias) - sh.setConnected(c.alias, false) for _, room := range rooms { room.out(c) } sh.callReceiver(c.accid, c.alias, nil) } else { - sh.setConnected(c.alias, true) entireConns[c.alias] = c } } @@ -293,6 +271,7 @@ func upgrade_core(sh *subhandler, conn *websocket.Conn, accid primitive.ObjectID sh.connInOutChan <- newconn go func(c *wsconn, accid primitive.ObjectID, deliveryChan chan<- any) { + sh.redisSync.HSet(context.Background(), sh.name, accid.Hex(), "online") for { messageType, r, err := c.NextReader() if err != nil { @@ -309,6 +288,7 @@ func upgrade_core(sh *subhandler, conn *websocket.Conn, accid primitive.ObjectID sh.callReceiver(accid, c.alias, r) } } + sh.redisSync.HDel(context.Background(), sh.name, accid.Hex()) c.Conn = nil sh.connInOutChan <- c