|
|
|
|
@@ -3,7 +3,6 @@ package client
|
|
|
|
|
import (
|
|
|
|
|
"context"
|
|
|
|
|
"fmt"
|
|
|
|
|
"io"
|
|
|
|
|
"log"
|
|
|
|
|
"net/url"
|
|
|
|
|
"strings"
|
|
|
|
|
@@ -221,26 +220,14 @@ func (wsClient *SafeWebsocketClient) connect() error {
|
|
|
|
|
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())
|
|
|
|
|
go wsClient.startPingTicker(pingCtx)
|
|
|
|
|
wsClient.mu.WriteHandler(func() error {
|
|
|
|
|
wsClient.cancelFuncs = append(wsClient.cancelFuncs, pingCancel)
|
|
|
|
|
return nil
|
|
|
|
|
})
|
|
|
|
|
|
|
|
|
|
go wsClient.startPingTicker(pingCtx)
|
|
|
|
|
|
|
|
|
|
if wsClient.conn != nil {
|
|
|
|
|
wsClient.conn.Close()
|
|
|
|
|
}
|
|
|
|
|
@@ -260,22 +247,25 @@ func (wsClient *SafeWebsocketClient) connect() error {
|
|
|
|
|
return nil
|
|
|
|
|
})
|
|
|
|
|
|
|
|
|
|
writer, err := conn.NextWriter(websocket.TextMessage)
|
|
|
|
|
if err != nil {
|
|
|
|
|
return
|
|
|
|
|
}
|
|
|
|
|
var c *websocket.Conn
|
|
|
|
|
wsClient.mu.ReadHandler(func() error {
|
|
|
|
|
c = conn
|
|
|
|
|
return nil
|
|
|
|
|
})
|
|
|
|
|
|
|
|
|
|
for {
|
|
|
|
|
select {
|
|
|
|
|
case <-ctx.Done():
|
|
|
|
|
fmt.Println("Writer stopped due to client shutdown")
|
|
|
|
|
log.Println("Writer stopped due to client shutdown")
|
|
|
|
|
return
|
|
|
|
|
case data := <-wsClient.writeChan:
|
|
|
|
|
if conn == nil {
|
|
|
|
|
if c == nil {
|
|
|
|
|
wsClient.triggerReconnect()
|
|
|
|
|
return
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
if _, err := writer.Write(data.Data); err != nil {
|
|
|
|
|
if err := c.WriteMessage(int(data.MessageType), data.Data); err != nil {
|
|
|
|
|
log.Printf("error on write message: %v\n", err)
|
|
|
|
|
wsClient.triggerReconnect()
|
|
|
|
|
return
|
|
|
|
|
}
|
|
|
|
|
@@ -290,50 +280,66 @@ func (wsClient *SafeWebsocketClient) connect() error {
|
|
|
|
|
return nil
|
|
|
|
|
})
|
|
|
|
|
|
|
|
|
|
var c *websocket.Conn
|
|
|
|
|
|
|
|
|
|
wsClient.mu.ReadHandler(func() error {
|
|
|
|
|
c = conn
|
|
|
|
|
return nil
|
|
|
|
|
})
|
|
|
|
|
if c == nil {
|
|
|
|
|
wsClient.triggerReconnect()
|
|
|
|
|
return
|
|
|
|
|
}
|
|
|
|
|
for {
|
|
|
|
|
select {
|
|
|
|
|
case <-ctx.Done():
|
|
|
|
|
fmt.Println("Reader stopped due to client shutdown")
|
|
|
|
|
log.Println("Reader stopped due to client shutdown")
|
|
|
|
|
return
|
|
|
|
|
default:
|
|
|
|
|
if conn == nil {
|
|
|
|
|
return
|
|
|
|
|
}
|
|
|
|
|
if err := conn.SetReadDeadline(time.Now().Add(readDeadline)); err != nil {
|
|
|
|
|
wsClient.triggerReconnect()
|
|
|
|
|
fmt.Printf("error on read deadline: %v\n", err)
|
|
|
|
|
if err := c.SetReadDeadline(time.Now().Add(readDeadline)); err != nil {
|
|
|
|
|
log.Printf("error on read deadline: %v\n", err)
|
|
|
|
|
return
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
mt, reader, err := conn.NextReader()
|
|
|
|
|
messageType, data, err := c.ReadMessage()
|
|
|
|
|
if err != nil {
|
|
|
|
|
log.Printf("error on read message: %v\n", err)
|
|
|
|
|
wsClient.triggerReconnect()
|
|
|
|
|
return
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
if mt != websocket.TextMessage {
|
|
|
|
|
if messageType != websocket.TextMessage {
|
|
|
|
|
continue
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
readerBytes, err := io.ReadAll(reader)
|
|
|
|
|
if err != nil {
|
|
|
|
|
fmt.Printf("io reader failed: %v\n", err)
|
|
|
|
|
wsClient.triggerReconnect()
|
|
|
|
|
return
|
|
|
|
|
}
|
|
|
|
|
select {
|
|
|
|
|
case wsClient.dataChannel <- readerBytes:
|
|
|
|
|
case wsClient.dataChannel <- data:
|
|
|
|
|
case <-ctx.Done():
|
|
|
|
|
return
|
|
|
|
|
default:
|
|
|
|
|
fmt.Println("Data channel full, dropping message")
|
|
|
|
|
log.Println("Data channel full, dropping message")
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
}
|
|
|
|
|
}()
|
|
|
|
|
|
|
|
|
|
conn.SetPingHandler(func(pingData string) error {
|
|
|
|
|
wsClient.writeChan <- Message{
|
|
|
|
|
MessageType: MessageTypePong,
|
|
|
|
|
Data: []byte(pingData),
|
|
|
|
|
}
|
|
|
|
|
// 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
|
|
|
|
|
})
|
|
|
|
|
|
|
|
|
|
return nil
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
@@ -458,7 +464,7 @@ func (wsClient *SafeWebsocketClient) Close() error {
|
|
|
|
|
wsClient.conn.Close()
|
|
|
|
|
}
|
|
|
|
|
wsClient.isConnected = false
|
|
|
|
|
close(wsClient.dataChannel)
|
|
|
|
|
// close(wsClient.dataChannel)
|
|
|
|
|
|
|
|
|
|
return nil
|
|
|
|
|
}
|
|
|
|
|
|