ayo/gocommon
8
0
Fork 1
Code Issues Pull Requests Projects Releases Wiki Activity

코드 정리 및 websocket도 http와 비슷하게 api handler로 통일

Browse Source
This commit is contained in:
mountain
2023-09-08 11:35:57 +09:00
parent eb54fa2e3a
commit 6cbf32c386
10 changed files with 216 additions and 130 deletions
Download Patch File Download Diff File Expand all files Collapse all files

112
wshandler/wshandler.go
View File

@ -90,30 +90,11 @@ const (
CloseMessage = WebSocketMessageType(websocket.CloseMessage)
PingMessage = WebSocketMessageType(websocket.PingMessage)
PongMessage = WebSocketMessageType(websocket.PongMessage)
Connected = WebSocketMessageType(100)
Disconnected = WebSocketMessageType(101)
)
type Sender struct {
Accid primitive.ObjectID
Alias string
disconnectedCallbacks map[string]func()
}
func (s *Sender) RegistDisconnectedCallback(name string, f func()) (old func()) {
if s.disconnectedCallbacks == nil {
s.disconnectedCallbacks = make(map[string]func())
}
old = s.disconnectedCallbacks[name]
s.disconnectedCallbacks[name] = f
return
}
func (s *Sender) PopDisconnectedCallback(name string) func() {
old := s.disconnectedCallbacks[name]
delete(s.disconnectedCallbacks, name)
return old
Accid primitive.ObjectID
Alias string
}
type EventReceiver interface {
@ -136,10 +117,10 @@ type WebsocketHandler struct {
deliveryChan chan any
localDeliveryChan chan any
sendMsgChan chan send_msg_queue_elem
callReceiver EventReceiver
connWaitGroup sync.WaitGroup
receiverChain []EventReceiver
sessionConsumer session.Consumer
wsApiBroker WebsocketApiBroker
connWaitGroup sync.WaitGroup
sessionConsumer session.Consumer
}
type wsConfig struct {
@ -156,13 +137,13 @@ func init() {
gob.Register([]any{})
}
func NewWebsocketHandler(consumer session.Consumer) (*WebsocketHandler, error) {
func NewWebsocketHandler(consumer session.Consumer, redisUrl string) (*WebsocketHandler, error) {
var config wsConfig
if err := gocommon.LoadConfig(&config); err != nil {
return nil, err
}
redisSync, err := gocommon.NewRedisClient(config.Redis["wshandler"])
redisSync, err := gocommon.NewRedisClient(redisUrl)
if err != nil {
return nil, err
}
@ -196,49 +177,11 @@ func NewWebsocketHandler(consumer session.Consumer) (*WebsocketHandler, error) {
}, nil
}
func (ws *WebsocketHandler) RegisterReceiver(receiver EventReceiver) {
ws.receiverChain = append(ws.receiverChain, receiver)
}
type nilReceiver struct{}
func (r *nilReceiver) OnClientMessageReceived(sender *Sender, messageType WebSocketMessageType, body io.Reader) {
}
func (r *nilReceiver) OnRoomCreated(name string) {}
func (r *nilReceiver) OnRoomDestroyed(name string) {}
type chainReceiver struct {
chain []EventReceiver
}
func (r *chainReceiver) OnClientMessageReceived(sender *Sender, messageType WebSocketMessageType, body io.Reader) {
for _, cr := range r.chain {
cr.OnClientMessageReceived(sender, messageType, body)
}
}
func (r *chainReceiver) OnRoomCreated(name string) {
for _, cr := range r.chain {
cr.OnRoomCreated(name)
}
}
func (r *chainReceiver) OnRoomDestroyed(name string) {
for _, cr := range r.chain {
cr.OnRoomDestroyed(name)
}
func (ws *WebsocketHandler) RegisterApiHandler(handler WebsocketApiHandler) {
ws.wsApiBroker.AddHandler(handler)
}
func (ws *WebsocketHandler) Start(ctx context.Context) {
chain := ws.receiverChain
if len(chain) == 0 {
ws.callReceiver = &nilReceiver{}
} else if len(chain) == 1 {
ws.callReceiver = chain[0]
} else {
ws.callReceiver = &chainReceiver{chain: ws.receiverChain}
}
ws.connWaitGroup.Add(1)
go ws.mainLoop(ctx)
}
@ -334,19 +277,14 @@ func (ws *WebsocketHandler) mainLoop(ctx context.Context) {
room = makeRoom(name, roomDestroyChan, ws.sendMsgChan)
rooms[name] = room
room.start(ctx)
go ws.callReceiver.OnRoomCreated(name)
//go ws.callReceiver.OnRoomCreated(name)
}
return room
}
defer func() {
for _, conn := range entireConns {
var roomnames []string
for _, room := range conn.joinedRooms {
roomnames = append(roomnames, room.name)
}
bt, _ := json.Marshal(roomnames)
ws.callReceiver.OnClientMessageReceived(conn.sender, Disconnected, bytes.NewBuffer(bt))
ws.wsApiBroker.Call(conn.sender, ClientDisconnected, nil)
conn.Close()
}
}()
@ -442,7 +380,7 @@ func (ws *WebsocketHandler) mainLoop(ctx context.Context) {
case destroyedRoom := <-roomDestroyChan:
delete(rooms, destroyedRoom)
go ws.callReceiver.OnRoomDestroyed(destroyedRoom)
//go ws.callReceiver.OnRoomDestroyed(destroyedRoom)
case usermsg := <-ws.localDeliveryChan:
// 로컬에 connection이 있는지 먼저 확인해 보기 위한 채널
@ -509,18 +447,14 @@ func (ws *WebsocketHandler) mainLoop(ctx context.Context) {
case c := <-ws.connInOutChan:
if c.Conn == nil {
delete(entireConns, c.sender.Accid.Hex())
var roomnames []string
for _, room := range c.joinedRooms {
roomnames = append(roomnames, room.name)
room.out(c)
}
c.joinedRooms = nil
bt, _ := json.Marshal(roomnames)
go ws.callReceiver.OnClientMessageReceived(c.sender, Disconnected, bytes.NewBuffer(bt))
go ws.wsApiBroker.Call(c.sender, ClientDisconnected, nil)
} else {
entireConns[c.sender.Accid.Hex()] = c
go ws.callReceiver.OnClientMessageReceived(c.sender, Connected, nil)
go ws.wsApiBroker.Call(c.sender, ClientConnected, nil)
}
}
}
@ -549,16 +483,12 @@ func upgrade_core(ws *WebsocketHandler, conn *websocket.Conn, accid primitive.Ob
break
}
if messageType == websocket.TextMessage {
// 유저가 직접 보낸 메시지
ws.callReceiver.OnClientMessageReceived(c.sender, TextMessage, r)
} else if messageType == websocket.BinaryMessage {
ws.callReceiver.OnClientMessageReceived(c.sender, BinaryMessage, r)
}
}
if c.sender.disconnectedCallbacks != nil {
for _, f := range c.sender.disconnectedCallbacks {
f()
if messageType == websocket.BinaryMessage {
var size [1]byte
r.Read(size[:])
cmd := make([]byte, size[0])
r.Read(cmd)
ws.wsApiBroker.Call(newconn.sender, string(cmd), r)
}
}
ws.connWaitGroup.Done()