fix client & server experimental fix

internal/hub.go

@@ -11,7 +11,7 @@ import (
 const (
 	writeWait  = 10 * time.Second
 	pongWait   = 60 * time.Second
-	pingPeriod = 55 * time.Second
+	pingPeriod = 25 * time.Second
 )
 
 type Client struct {
@@ -95,7 +95,7 @@ func WritePump(c *Client, h *Hub) {
 			}
 		case <-pingTicker.C:
 			c.Conn.SetWriteDeadline(time.Now().Add(writeWait))
-			if err := c.Conn.WriteMessage(websocket.PingMessage, nil); err != nil {
+			if err := c.Conn.WriteMessage(websocket.PingMessage, []byte{}); err != nil {
 				return
 			}
 		}

v1/client/client.go

@@ -59,8 +59,6 @@ func (b *SafeWebsocketClientBuilder) Build() (*SafeWebsocketClient, error) {
 		return nil, err
 	}
 
-	ctx, cancel := context.WithCancel(context.Background())
-
 	var useTLS bool
 	if b.useTLS != nil {
 		useTLS = *b.useTLS
@@ -72,8 +70,6 @@ func (b *SafeWebsocketClientBuilder) Build() (*SafeWebsocketClient, error) {
 		useTLS:      useTLS,
 		path:        b.path,
 		rawQuery:    b.rawQuery,
-		ctx:         ctx,
-		cancel:      cancel,
 		dataChannel: make(chan []byte, 1),
 		mu:          custom_rwmutex.NewCustomRwMutex(),
 		reconnectCh: make(chan struct{}, 1),
@@ -81,7 +77,6 @@ func (b *SafeWebsocketClientBuilder) Build() (*SafeWebsocketClient, error) {
 	}
 
 	if err := wsClient.connect(); err != nil {
-		cancel()
 		return nil, fmt.Errorf("failed to establish initial connection: %v", err)
 	}
 
@@ -129,21 +124,27 @@ func (wsClient *SafeWebsocketClient) connect() error {
 	}
 
 	conn.SetPingHandler(func(pingData string) error {
-		if err := conn.WriteMessage(websocket.PongMessage, []byte(pingData)); err != nil {
-			if err == websocket.ErrCloseSent {
-				return nil
+		return wsClient.mu.WriteHandler(func() 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
 			}
-			if netErr, ok := err.(interface{ Timeout() bool }); ok && netErr.Timeout() {
-				return nil
-			}
-			return err
-		}
-		return nil
+			return nil
+		})
 	})
 
 	wsClient.mu.WriteHandler(func() error {
+		ctx, cancel := context.WithCancel(context.Background())
+		wsClient.ctx = ctx
+		wsClient.cancel = cancel
 		wsClient.conn = conn
 		wsClient.isConnected = true
+
 		return nil
 	})
 
@@ -177,27 +178,30 @@ func (wsClient *SafeWebsocketClient) startPingTicker() {
 func (wsClient *SafeWebsocketClient) startReceiveHandler() {
 	for {
 		select {
-		case <-wsClient.reconnectCh:
 		case <-wsClient.ctx.Done():
 			log.Println("Reconnect handler stopped")
 			return
 		default:
-			if err := wsClient.mu.ReadHandler(func() error {
-				conn := wsClient.conn
+			// if err := wsClient.mu.ReadHandler(func() error {
+			conn := wsClient.conn
 
-				if conn == nil {
-					return fmt.Errorf("no active connection, waiting for reconnect")
-				}
-				_, message, err := conn.ReadMessage()
-				if err != nil {
-					return err
-				}
-				wsClient.dataChannel <- message
-				return nil
-			}); err != nil {
+			if conn == nil {
 				wsClient.triggerReconnect()
 				return
+				// return fmt.Errorf("no active connection, waiting for reconnect")
 			}
+			_, message, err := conn.ReadMessage()
+			if err != nil {
+				wsClient.triggerReconnect()
+				return
+				// return err
+			}
+			wsClient.dataChannel <- message
+			// return nil
+			// }); err != nil {
+			// 	wsClient.triggerReconnect()
+			// 	return
+			// }
 		}
 	}
 }
@@ -210,8 +214,14 @@ func (wsClient *SafeWebsocketClient) triggerReconnect() {
 }
 
 func (wsClient *SafeWebsocketClient) reconnectHandler() {
-	for range wsClient.reconnectCh {
-		wsClient.connect()
+	for {
+		select {
+		case <-wsClient.reconnectCh:
+			wsClient.cancel()
+			wsClient.connect()
+		case <-wsClient.ctx.Done():
+			return
+		}
 	}
 }