GetState 함수 추가

This commit is contained in:
2023-07-06 00:01:45 +09:00
parent 822681bf74
commit 7ef3e68d16

View File

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