package websocket import ( "encoding/json" "log" "stocksearch/models" "time" "github.com/gorilla/websocket" ) const ( writeWait = 10 * time.Second // 쓰기 타임아웃 pongWait = 60 * time.Second // Pong 대기 시간 pingPeriod = (pongWait * 9) / 10 // Ping 전송 주기 maxMsgSize = 512 // 최대 메시지 크기 (바이트) ) // Client WebSocket 개별 클라이언트 type Client struct { hub *Hub conn *websocket.Conn send chan []byte // 쓰기 버퍼 채널 (슬로우 클라이언트 방어) } // readPump 클라이언트로부터 메시지 수신 (고루틴으로 실행) // 읽기와 쓰기를 분리해 gorilla/websocket 동시 호출 방지 func (c *Client) readPump() { defer func() { c.hub.unregister <- c c.conn.Close() }() c.conn.SetReadLimit(maxMsgSize) c.conn.SetReadDeadline(time.Now().Add(pongWait)) c.conn.SetPongHandler(func(string) error { c.conn.SetReadDeadline(time.Now().Add(pongWait)) return nil }) for { _, message, err := c.conn.ReadMessage() if err != nil { if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) { log.Printf("WebSocket 읽기 오류: %v", err) } break } // 클라이언트로부터 구독/해제 메시지 처리 var msg models.WSMessage if err := json.Unmarshal(message, &msg); err != nil { log.Printf("메시지 파싱 실패: %v", err) continue } switch msg.Type { case "subscribe": c.hub.subscribe <- &SubscribeMsg{Client: c, Code: msg.Code} case "unsubscribe": c.hub.unsubscribeCode <- &SubscribeMsg{Client: c, Code: msg.Code} } } } // writePump 클라이언트에게 메시지 전송 (고루틴으로 실행) func (c *Client) writePump() { ticker := time.NewTicker(pingPeriod) defer func() { ticker.Stop() c.conn.Close() }() for { select { case message, ok := <-c.send: c.conn.SetWriteDeadline(time.Now().Add(writeWait)) if !ok { // Hub가 채널을 닫음 c.conn.WriteMessage(websocket.CloseMessage, []byte{}) return } if err := c.conn.WriteMessage(websocket.TextMessage, message); err != nil { return } case <-ticker.C: // Ping 전송으로 연결 유지 c.conn.SetWriteDeadline(time.Now().Add(writeWait)) if err := c.conn.WriteMessage(websocket.PingMessage, nil); err != nil { return } } } }