Compare commits

...

5 Commits

Author SHA1 Message Date
88ae539f54 fix: fixing memory leak from unbuffered channel 2025-10-07 09:16:51 +07:00
78a5b21531 fix: fixing websocket error message log 2025-10-03 10:40:46 +07:00
e011931436 fix: messageType typo 2025-10-01 15:09:06 +07:00
70d9a37a4e fix: pong data race 2025-10-01 14:41:23 +07:00
2ab2e07b9b fix: pong not sent 2025-10-01 14:00:22 +07:00
3 changed files with 25 additions and 24 deletions

2
go.mod
View File

@@ -4,4 +4,4 @@ go 1.24.5
require github.com/gorilla/websocket v1.5.3 require github.com/gorilla/websocket v1.5.3
require git.neurocipta.com/rogerferdinan/custom-rwmutex v1.0.0 // indirect require git.neurocipta.com/rogerferdinan/custom-rwmutex v1.0.0

View File

@@ -23,9 +23,9 @@ type Client struct {
func NewClient(conn *websocket.Conn, subscribedPath string) *Client { func NewClient(conn *websocket.Conn, subscribedPath string) *Client {
return &Client{ return &Client{
Conn: conn, Conn: conn,
Send: make(chan []byte, 64), Send: make(chan []byte, 2),
SubscribedPath: subscribedPath, SubscribedPath: subscribedPath,
done: make(chan struct{}), done: make(chan struct{}, 1),
} }
} }
@@ -38,10 +38,10 @@ type Hub struct {
func NewHub() *Hub { func NewHub() *Hub {
return &Hub{ return &Hub{
Broadcast: make(chan []byte), Broadcast: make(chan []byte, 1),
Register: make(chan *Client), Register: make(chan *Client, 1),
Unregister: make(chan *Client), Unregister: make(chan *Client, 1),
Clients: make(map[*Client]bool), Clients: make(map[*Client]bool, 1),
} }
} }
@@ -117,7 +117,7 @@ func ReadPump(c *Client, h *Hub) {
for { for {
messageType, message, err := c.Conn.ReadMessage() messageType, message, err := c.Conn.ReadMessage()
if err != nil { if err != nil {
if !websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) { if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) {
log.Printf("WebSocket error: %v", err) log.Printf("WebSocket error: %v", err)
} }
break break

View File

@@ -16,7 +16,7 @@ import (
const ( const (
pingPeriod = 10 * time.Second pingPeriod = 10 * time.Second
readDeadline = 120 * time.Second readDeadline = 30 * time.Second
) )
type SafeMap[K comparable, V any] struct { type SafeMap[K comparable, V any] struct {
@@ -68,10 +68,10 @@ func (sm *SafeMap[K, V]) Len() int {
type MessageType uint type MessageType uint
const ( const (
MessageTypeText MessageType = iota + 1 MessageTypeText MessageType = websocket.TextMessage
MessageTypePing MessageType = iota MessageTypePing MessageType = websocket.PingMessage
MessageTypePong MessageType = iota MessageTypePong MessageType = websocket.PongMessage
MessageTypeClose MessageType = iota MessageTypeClose MessageType = websocket.CloseMessage
) )
type Message struct { type Message struct {
@@ -155,7 +155,7 @@ func (b *SafeWebsocketClientBuilder) Build(ctx context.Context) (*SafeWebsocketC
reconnectCh: make(chan struct{}, 1), reconnectCh: make(chan struct{}, 1),
isConnected: false, isConnected: false,
doneMap: NewSafeMap[string, chan struct{}](), doneMap: NewSafeMap[string, chan struct{}](),
writeChan: make(chan Message), writeChan: make(chan Message, 1),
} }
if b.authenticateFn != nil { if b.authenticateFn != nil {
@@ -193,6 +193,7 @@ type SafeWebsocketClient struct {
authenticateFn func(*SafeWebsocketClient) error authenticateFn func(*SafeWebsocketClient) error
writeChan chan Message writeChan chan Message
pongChan chan error
} }
func (wsClient *SafeWebsocketClient) connect() error { func (wsClient *SafeWebsocketClient) connect() error {
@@ -267,6 +268,9 @@ func (wsClient *SafeWebsocketClient) connect() error {
if err := c.WriteMessage(int(data.MessageType), data.Data); err != nil { if err := c.WriteMessage(int(data.MessageType), data.Data); err != nil {
log.Printf("error on write message: %v\n", err) log.Printf("error on write message: %v\n", err)
wsClient.triggerReconnect() wsClient.triggerReconnect()
if data.MessageType == MessageTypePong {
wsClient.pongChan <- err
}
return return
} }
} }
@@ -328,15 +332,12 @@ func (wsClient *SafeWebsocketClient) connect() error {
MessageType: MessageTypePong, MessageType: MessageTypePong,
Data: []byte(pingData), Data: []byte(pingData),
} }
// if err := conn.WriteMessage(websocket.PongMessage, []byte(pingData)); err != nil {
// if err == websocket.ErrCloseSent { select {
// return nil case err := <-wsClient.pongChan:
// } return err
// if netErr, ok := err.(interface{ Timeout() bool }); ok && netErr.Timeout() { default:
// return nil }
// }
// return err
// }
return nil return nil
}) })
@@ -424,7 +425,7 @@ func (wsClient *SafeWebsocketClient) reconnectHandler() {
} }
func (wsClient *SafeWebsocketClient) ReconnectChannel() <-chan struct{} { func (wsClient *SafeWebsocketClient) ReconnectChannel() <-chan struct{} {
reconnectCh := make(chan struct{}) reconnectCh := make(chan struct{}, 1)
wsClient.mu.WriteHandler(func() error { wsClient.mu.WriteHandler(func() error {
wsClient.reconnectChans = append(wsClient.reconnectChans, reconnectCh) wsClient.reconnectChans = append(wsClient.reconnectChans, reconnectCh)
return nil return nil