|
|
|
|
@@ -68,10 +68,10 @@ func (sm *SafeMap[K, V]) Len() int {
|
|
|
|
|
type MessageType uint
|
|
|
|
|
|
|
|
|
|
const (
|
|
|
|
|
MessageTypeText MessageType = iota + 1
|
|
|
|
|
MessageTypePing MessageType = iota
|
|
|
|
|
MessageTypePong MessageType = iota
|
|
|
|
|
MessageTypeClose MessageType = iota
|
|
|
|
|
MessageTypeText MessageType = websocket.TextMessage
|
|
|
|
|
MessageTypePing MessageType = websocket.PingMessage
|
|
|
|
|
MessageTypePong MessageType = websocket.PongMessage
|
|
|
|
|
MessageTypeClose MessageType = websocket.CloseMessage
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
type Message struct {
|
|
|
|
|
@@ -193,6 +193,7 @@ type SafeWebsocketClient struct {
|
|
|
|
|
authenticateFn func(*SafeWebsocketClient) error
|
|
|
|
|
|
|
|
|
|
writeChan chan Message
|
|
|
|
|
pongChan chan 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)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
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())
|
|
|
|
|
wsClient.mu.WriteHandler(func() error {
|
|
|
|
|
wsClient.cancelFuncs = append(wsClient.cancelFuncs, pingCancel)
|
|
|
|
|
@@ -260,20 +248,29 @@ func (wsClient *SafeWebsocketClient) connect() error {
|
|
|
|
|
return nil
|
|
|
|
|
})
|
|
|
|
|
|
|
|
|
|
var c *websocket.Conn
|
|
|
|
|
wsClient.mu.ReadHandler(func() error {
|
|
|
|
|
c = conn
|
|
|
|
|
return nil
|
|
|
|
|
})
|
|
|
|
|
|
|
|
|
|
for {
|
|
|
|
|
select {
|
|
|
|
|
case <-ctx.Done():
|
|
|
|
|
log.Println("Writer stopped due to client shutdown")
|
|
|
|
|
return
|
|
|
|
|
case data := <-wsClient.writeChan:
|
|
|
|
|
if conn == nil {
|
|
|
|
|
if c == nil {
|
|
|
|
|
wsClient.triggerReconnect()
|
|
|
|
|
return
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
if err := conn.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)
|
|
|
|
|
wsClient.triggerReconnect()
|
|
|
|
|
if data.MessageType == MessageTypePong {
|
|
|
|
|
wsClient.pongChan <- err
|
|
|
|
|
}
|
|
|
|
|
return
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
@@ -287,22 +284,28 @@ 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():
|
|
|
|
|
log.Println("Reader stopped due to client shutdown")
|
|
|
|
|
return
|
|
|
|
|
default:
|
|
|
|
|
if conn == nil {
|
|
|
|
|
wsClient.triggerReconnect()
|
|
|
|
|
return
|
|
|
|
|
}
|
|
|
|
|
if err := conn.SetReadDeadline(time.Now().Add(readDeadline)); err != nil {
|
|
|
|
|
if err := c.SetReadDeadline(time.Now().Add(readDeadline)); err != nil {
|
|
|
|
|
log.Printf("error on read deadline: %v\n", err)
|
|
|
|
|
return
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
messageType, data, err := conn.ReadMessage()
|
|
|
|
|
messageType, data, err := c.ReadMessage()
|
|
|
|
|
if err != nil {
|
|
|
|
|
log.Printf("error on read message: %v\n", err)
|
|
|
|
|
wsClient.triggerReconnect()
|
|
|
|
|
@@ -324,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
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
@@ -339,7 +356,7 @@ func (wsClient *SafeWebsocketClient) startPingTicker(ctx context.Context) {
|
|
|
|
|
case <-ticker.C:
|
|
|
|
|
wsClient.writeChan <- Message{
|
|
|
|
|
MessageType: websocket.PingMessage,
|
|
|
|
|
Data: nil,
|
|
|
|
|
Data: []byte{},
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|