Compare commits

...

5 Commits

Author SHA1 Message Date
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
9a9c65c24c fix: ping not sent 2025-10-01 13:45:48 +07:00
8a11ce0103 fix: extend read deadline wait 2025-09-30 13:13:29 +07:00
cf63683c9c fix: fixing panic on pong write message 2025-09-30 10:43:39 +07:00

View File

@@ -16,7 +16,7 @@ import (
const ( const (
pingPeriod = 10 * time.Second pingPeriod = 10 * time.Second
readDeadline = 30 * time.Second readDeadline = 120 * time.Second
) )
type SafeMap[K comparable, V any] struct { type SafeMap[K comparable, V any] struct {
@@ -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 {
@@ -220,19 +221,6 @@ func (wsClient *SafeWebsocketClient) connect() error {
return fmt.Errorf("failed to connect to %s: %w", wsClient.baseHost, err) return fmt.Errorf("failed to connect to %s: %w", wsClient.baseHost, err)
} }
conn.SetPingHandler(func(pingData string) error {
if err := conn.WriteMessage(websocket.PongMessage, []byte(pingData)); err != nil {
if err == websocket.ErrCloseSent {
return nil
}
if netErr, ok := err.(interface{ Timeout() bool }); ok && netErr.Timeout() {
return nil
}
return err
}
return nil
})
pingCtx, pingCancel := context.WithCancel(context.Background()) pingCtx, pingCancel := context.WithCancel(context.Background())
wsClient.mu.WriteHandler(func() error { wsClient.mu.WriteHandler(func() error {
wsClient.cancelFuncs = append(wsClient.cancelFuncs, pingCancel) wsClient.cancelFuncs = append(wsClient.cancelFuncs, pingCancel)
@@ -280,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
} }
} }
@@ -336,6 +327,20 @@ func (wsClient *SafeWebsocketClient) connect() error {
} }
}() }()
conn.SetPingHandler(func(pingData string) error {
wsClient.writeChan <- Message{
MessageType: MessageTypePong,
Data: []byte(pingData),
}
select {
case err := <-wsClient.pongChan:
return err
default:
}
return nil
})
return nil return nil
} }
@@ -351,7 +356,7 @@ func (wsClient *SafeWebsocketClient) startPingTicker(ctx context.Context) {
case <-ticker.C: case <-ticker.C:
wsClient.writeChan <- Message{ wsClient.writeChan <- Message{
MessageType: websocket.PingMessage, MessageType: websocket.PingMessage,
Data: nil, Data: []byte{},
} }
} }
} }