fix(io): revise websocket connection (#3728)

Signed-off-by: Song Gao <disxiaofei@163.com>

Author: Yisaer

internal/io/http/httpserver/websocket_server.go

@@ -18,7 +18,6 @@ import (
 	"context"
 	"fmt"
 	"net/http"
-	"strings"
 	"sync"
 
 	"github.com/gorilla/websocket"
@@ -92,9 +91,8 @@ func sendProcess(ctx api.StreamContext, topic, sourceID string, c *websocket.Con
 		case d := <-ch:
 			data := d.([]byte)
 			if err := c.WriteMessage(websocket.TextMessage, data); err != nil {
-				if websocket.IsCloseError(err) || strings.Contains(err.Error(), "close") {
-					return
-				}
+				conf.Log.Errorf("write websocket msg err:%v, topic:%v", err, topic)
+				return
 			}
 		}
 	}
@@ -114,10 +112,8 @@ func recvProcess(ctx api.StreamContext, topic string, c *websocket.Conn, cancel
 		}
 		msgType, data, err := c.ReadMessage()
 		if err != nil {
-			if websocket.IsCloseError(err) || strings.Contains(err.Error(), "close") {
-				return
-			}
-			continue
+			conf.Log.Errorf("read websocket msg err:%v, topic:%v", err, topic)
+			return
 		}
 		switch msgType {
 		case websocket.TextMessage:

internal/io/websocket/websocket_sink.go

@@ -60,6 +60,9 @@ func (w *WebsocketSink) Connect(ctx api.StreamContext, sch api.StatusChangeHandl
 		return err
 	}
 	conn, err := w.cw.Wait(ctx)
+	if err != nil {
+		return err
+	}
 	if conn == nil {
 		return fmt.Errorf("websocket endpoint not ready: %v", err)
 	}

internal/io/websocket/websocket_source.go

@@ -64,6 +64,9 @@ func (w *WebsocketSource) Connect(ctx api.StreamContext, sc api.StatusChangeHand
 		return err
 	}
 	conn, err := cw.Wait(ctx)
+	if err != nil {
+		return err
+	}
 	if conn == nil {
 		return fmt.Errorf("websocket endpoint not ready: %v", err)
 	}