From 92a20e3c9a7a1d71fba2b34e19d0230d78dfdf1a Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E5=88=98=E7=A5=A5=E8=B6=85?= <iwind.liu@gmail.com>
Date: Fri, 30 Sep 2022 16:34:21 +0800
Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8DWebsocket=E6=97=A0=E6=B3=95?=
 =?UTF-8?q?=E6=AD=A3=E5=B8=B8=E4=BA=A4=E4=BA=92=E7=9A=84=E9=97=AE=E9=A2=98?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 internal/nodes/http_request_websocket.go | 56 +++++++++++++++++++++++-
 internal/utils/readers/reader_print.go   | 28 ++++++++++++
 internal/utils/writers/writer_print.go   | 28 ++++++++++++
 3 files changed, 111 insertions(+), 1 deletion(-)
 create mode 100644 internal/utils/readers/reader_print.go
 create mode 100644 internal/utils/writers/writer_print.go

diff --git a/internal/nodes/http_request_websocket.go b/internal/nodes/http_request_websocket.go
index f9141ff..de6dd0f 100644
--- a/internal/nodes/http_request_websocket.go
+++ b/internal/nodes/http_request_websocket.go
@@ -2,6 +2,7 @@ package nodes
 
 import (
 	"bufio"
+	"bytes"
 	"errors"
 	"github.com/TeaOSLab/EdgeNode/internal/utils"
 	"io"
@@ -9,8 +10,36 @@ import (
 	"net/url"
 )
 
+// WebsocketResponseReader Websocket响应Reader
+type WebsocketResponseReader struct {
+	rawReader io.Reader
+	buf       []byte
+}
+
+func NewWebsocketResponseReader(rawReader io.Reader) *WebsocketResponseReader {
+	return &WebsocketResponseReader{
+		rawReader: rawReader,
+	}
+}
+
+func (this *WebsocketResponseReader) Read(p []byte) (n int, err error) {
+	n, err = this.rawReader.Read(p)
+	if n > 0 {
+		if len(this.buf) == 0 {
+			this.buf = make([]byte, n)
+			copy(this.buf, p[:n])
+		} else {
+			this.buf = append(this.buf, p[:n]...)
+		}
+	}
+	return
+}
+
 // 处理Websocket请求
 func (this *HTTPRequest) doWebsocket(requestHost string, isLastRetry bool) (shouldRetry bool) {
+	// 设置不缓存
+	this.web.Cache = nil
+
 	if this.web.WebsocketRef == nil || !this.web.WebsocketRef.IsOn || this.web.Websocket == nil || !this.web.Websocket.IsOn {
 		this.writer.WriteHeader(http.StatusForbidden)
 		this.addError(errors.New("websocket have not been enabled yet"))
@@ -84,14 +113,20 @@ func (this *HTTPRequest) doWebsocket(requestHost string, isLastRetry bool) (shou
 
 	go func() {
 		// 读取第一个响应
-		resp, err := http.ReadResponse(bufio.NewReader(originConn), this.RawReq)
+		var respReader = NewWebsocketResponseReader(originConn)
+		resp, err := http.ReadResponse(bufio.NewReader(respReader), this.RawReq)
 		if err != nil {
+			if resp.Body != nil {
+				_ = resp.Body.Close()
+			}
+
 			_ = clientConn.Close()
 			_ = originConn.Close()
 			return
 		}
 
 		this.processResponseHeaders(resp.Header, resp.StatusCode)
+		this.writer.statusCode = resp.StatusCode
 
 		// 将响应写回客户端
 		err = resp.Write(clientConn)
@@ -105,6 +140,25 @@ func (this *HTTPRequest) doWebsocket(requestHost string, isLastRetry bool) (shou
 			return
 		}
 
+		// 剩余已经从源站读取的内容
+		var headerBytes = respReader.buf
+		var headerIndex = bytes.Index(headerBytes, []byte{'\r', '\n', '\r', '\n'}) // CRLF
+		if headerIndex > 0 {
+			var leftBytes = headerBytes[headerIndex+4:]
+			if len(leftBytes) > 0 {
+				_, err = clientConn.Write(leftBytes)
+				if err != nil {
+					if resp.Body != nil {
+						_ = resp.Body.Close()
+					}
+
+					_ = clientConn.Close()
+					_ = originConn.Close()
+					return
+				}
+			}
+		}
+
 		if resp.Body != nil {
 			_ = resp.Body.Close()
 		}
diff --git a/internal/utils/readers/reader_print.go b/internal/utils/readers/reader_print.go
new file mode 100644
index 0000000..b484774
--- /dev/null
+++ b/internal/utils/readers/reader_print.go
@@ -0,0 +1,28 @@
+// Copyright 2022 Liuxiangchao iwind.liu@gmail.com. All rights reserved. Official site: https://goedge.cn .
+
+package readers
+
+import (
+	"io"
+	"log"
+)
+
+type PrintReader struct {
+	rawReader io.Reader
+	tag       string
+}
+
+func NewPrintReader(rawReader io.Reader, tag string) io.Reader {
+	return &PrintReader{
+		rawReader: rawReader,
+		tag:       tag,
+	}
+}
+
+func (this *PrintReader) Read(p []byte) (n int, err error) {
+	n, err = this.rawReader.Read(p)
+	if n > 0 {
+		log.Println("[" + this.tag + "]" + string(p[:n]))
+	}
+	return
+}
diff --git a/internal/utils/writers/writer_print.go b/internal/utils/writers/writer_print.go
new file mode 100644
index 0000000..7a51605
--- /dev/null
+++ b/internal/utils/writers/writer_print.go
@@ -0,0 +1,28 @@
+// Copyright 2022 Liuxiangchao iwind.liu@gmail.com. All rights reserved. Official site: https://goedge.cn .
+
+package writers
+
+import (
+	"io"
+	"log"
+)
+
+type PrintWriter struct {
+	rawWriter io.Writer
+	tag       string
+}
+
+func NewPrintWriter(rawWriter io.Writer, tag string) io.Writer {
+	return &PrintWriter{
+		rawWriter: rawWriter,
+		tag:       tag,
+	}
+}
+
+func (this *PrintWriter) Write(p []byte) (n int, err error) {
+	n, err = this.rawWriter.Write(p)
+	if n > 0 {
+		log.Println("[" + this.tag + "]" + string(p[:n]))
+	}
+	return
+}