fix http handler for tunnel

handler/forward/local/handler.go

@@ -185,6 +185,7 @@ func (h *forwardHandler) handleHTTP(ctx context.Context, rw io.ReadWriter, log l
 		resp := &http.Response{
 			ProtoMajor: 1,
 			ProtoMinor: 1,
+			Header:     http.Header{},
 			StatusCode: http.StatusServiceUnavailable,
 		}
 

handler/forward/remote/handler.go

@@ -11,7 +11,6 @@ import (
 	"net/http"
 	"net/http/httputil"
 	"strconv"
-	"sync"
 	"time"
 
 	"github.com/go-gost/core/chain"
@@ -183,7 +182,6 @@ func (h *forwardHandler) Handle(ctx context.Context, conn net.Conn, opts ...hand
 
 func (h *forwardHandler) handleHTTP(ctx context.Context, rw io.ReadWriter, remoteAddr net.Addr, localAddr net.Addr, log logger.Logger) (err error) {
 	br := bufio.NewReader(rw)
-	var connPool sync.Map
 
 	for {
 		resp := &http.Response{
@@ -233,13 +231,7 @@ func (h *forwardHandler) handleHTTP(ctx context.Context, rw io.ReadWriter, remot
 				ctx = auth_util.ContextWithID(ctx, auth_util.ID(id))
 			}
 
-			var cc net.Conn
+			cc, err := h.router.Dial(ctx, "tcp", target.Addr)
-			if v, ok := connPool.Load(target); ok {
-				cc = v.(net.Conn)
-				log.Debugf("reuse connection to node %s(%s)", target.Name, target.Addr)
-			}
-			if cc == nil {
-				cc, err = h.router.Dial(ctx, "tcp", target.Addr)
 			if err != nil {
 				// TODO: the router itself may be failed due to the failed node in the router,
 				// the dead marker may be a wrong operation.
@@ -252,6 +244,9 @@ func (h *forwardHandler) handleHTTP(ctx context.Context, rw io.ReadWriter, remot
 			if marker := target.Marker(); marker != nil {
 				marker.Reset()
 			}
+			defer cc.Close()
+
+			log.Debugf("new connection to node %s(%s)", target.Name, target.Addr)
 
 			if tlsSettings := target.Options().TLS; tlsSettings != nil {
 				cc = tls.Client(cc, &tls.Config{
@@ -260,22 +255,6 @@ func (h *forwardHandler) handleHTTP(ctx context.Context, rw io.ReadWriter, remot
 				})
 			}
 
-				cc = proxyproto.WrapClientConn(h.md.proxyProtocol, remoteAddr, localAddr, cc)
-
-				connPool.Store(target, cc)
-				log.Debugf("new connection to node %s(%s)", target.Name, target.Addr)
-
-				go func() {
-					defer cc.Close()
-					err := xnet.CopyBuffer(rw, cc, 8192)
-					if err != nil {
-						resp.Write(rw)
-					}
-					log.Debugf("close connection to node %s(%s), reason: %v", target.Name, target.Addr, err)
-					connPool.Delete(target)
-				}()
-			}
-
 			if httpSettings := target.Options().HTTP; httpSettings != nil {
 				if httpSettings.Host != "" {
 					req.Host = httpSettings.Host
@@ -289,35 +268,28 @@ func (h *forwardHandler) handleHTTP(ctx context.Context, rw io.ReadWriter, remot
 				dump, _ := httputil.DumpRequest(req, false)
 				log.Trace(string(dump))
 			}
+
+			cc = proxyproto.WrapClientConn(h.md.proxyProtocol, remoteAddr, localAddr, cc)
+
 			if err := req.Write(cc); err != nil {
-				log.Warnf("send request to node %s(%s) failed: %v", target.Name, target.Addr, err)
+				log.Warnf("send request to node %s(%s): %v", target.Name, target.Addr, err)
 				return resp.Write(rw)
 			}
 
-			if req.Header.Get("Upgrade") == "websocket" {
+			res, err := http.ReadResponse(bufio.NewReader(cc), req)
-				err := xnet.CopyBuffer(cc, br, 8192)
+			if err != nil {
-				if err == nil {
+				log.Warnf("read response from node %s(%s): %v", target.Name, target.Addr, err)
-					err = io.EOF
+				return resp.Write(rw)
-				}
-				return err
 			}
+			defer res.Body.Close()
 
-			// cc.SetReadDeadline(time.Now().Add(10 * time.Second))
+			return res.Write(rw)
-
-			return nil
 		}()
 		if err != nil {
 			break
 		}
 	}
 
-	connPool.Range(func(key, value any) bool {
-		if value != nil {
-			value.(net.Conn).Close()
-		}
-		return true
-	})
-
 	return
 }
 

handler/relay/entrypoint.go

@@ -8,7 +8,6 @@ import (
 	"net"
 	"net/http"
 	"net/http/httputil"
-	"sync"
 	"time"
 
 	"github.com/go-gost/core/handler"
@@ -179,8 +178,7 @@ func (h *tunnelHandler) Handle(ctx context.Context, conn net.Conn, opts ...handl
 	h.options.Logger.Debugf("sniffing: host=%s, protocol=%s", host, protocol)
 
 	if protocol == forward.ProtoHTTP {
-		h.handleHTTP(ctx, conn.RemoteAddr(), rw, log)
+		return h.handleHTTP(ctx, conn.RemoteAddr(), rw, log)
-		return nil
 	}
 
 	var tunnelID relay.TunnelID
@@ -241,12 +239,12 @@ func (h *tunnelHandler) Handle(ctx context.Context, conn net.Conn, opts ...handl
 
 func (h *tunnelHandler) handleHTTP(ctx context.Context, raddr net.Addr, rw io.ReadWriter, log logger.Logger) (err error) {
 	br := bufio.NewReader(rw)
-	var connPool sync.Map
 
 	for {
 		resp := &http.Response{
 			ProtoMajor: 1,
 			ProtoMinor: 1,
+			Header:     http.Header{},
 			StatusCode: http.StatusServiceUnavailable,
 		}
 
@@ -278,20 +276,13 @@ func (h *tunnelHandler) handleHTTP(ctx context.Context, raddr net.Addr, rw io.Re
 				"tunnel": tunnelID.String(),
 			})
 
-			var cc net.Conn
+			cc, cid, err := getTunnelConn("tcp", h.pool, tunnelID, 3, log)
-			if v, ok := connPool.Load(tunnelID); ok {
-				cc = v.(net.Conn)
-				log.Debugf("connection to tunnel %s found in pool", tunnelID)
-			}
-			if cc == nil {
-				var cid relay.ConnectorID
-				cc, cid, err = getTunnelConn("tcp", h.pool, tunnelID, 3, log)
 			if err != nil {
 				log.Error(err)
 				return resp.Write(rw)
 			}
+			defer cc.Close()
 
-				connPool.Store(tunnelID, cc)
 			log.Debugf("new connection to tunnel %s(connector %s)", tunnelID, cid)
 
 			var features []relay.Feature
@@ -314,37 +305,23 @@ func (h *tunnelHandler) handleHTTP(ctx context.Context, raddr net.Addr, rw io.Re
 				Features: features,
 			}).WriteTo(cc)
 
-				go func() {
-					defer cc.Close()
-					err := xnet.CopyBuffer(rw, cc, 8192)
-					if err != nil {
-						resp.Write(rw)
-					}
-					log.Debugf("close connection to tunnel %s(connector %s), reason: %v", tunnelID, cid, err)
-					connPool.Delete(tunnelID)
-				}()
-			}
-
 			if log.IsLevelEnabled(logger.TraceLevel) {
 				dump, _ := httputil.DumpRequest(req, false)
 				log.Trace(string(dump))
 			}
 			if err := req.Write(cc); err != nil {
-				log.Warnf("send request to tunnel %s failed: %v", tunnelID, err)
+				log.Warnf("send request to tunnel %s: %v", tunnelID, err)
 				return resp.Write(rw)
 			}
 
-			if req.Header.Get("Upgrade") == "websocket" {
+			res, err := http.ReadResponse(bufio.NewReader(cc), req)
-				err := xnet.CopyBuffer(cc, br, 8192)
+			if err != nil {
-				if err == nil {
+				log.Warnf("read response from tunnel %s: %v", tunnelID, err)
-					err = io.EOF
+				return resp.Write(rw)
-				}
-				return err
 			}
+			defer res.Body.Close()
 
-			// cc.SetReadDeadline(time.Now().Add(10 * time.Second))
+			return res.Write(rw)
-
-			return nil
 		}()
 		if err != nil {
 			// log.Error(err)
@@ -352,12 +329,5 @@ func (h *tunnelHandler) handleHTTP(ctx context.Context, raddr net.Addr, rw io.Re
 		}
 	}
 
-	connPool.Range(func(key, value any) bool {
-		if value != nil {
-			value.(net.Conn).Close()
-		}
-		return true
-	})
-
 	return
 }