merge: PR #92 — make pipe idle timeout configurable via idleTimeout metadata

Merge beezly/fix/configurable-pipe-timeout with conflict resolution
in pipe.go (PipeOption/WithReadTimeout functional options pattern).

Fix pipe_test.go: update 4 pipeHalf() calls to pass readTimeout arg
matching the new signature.

handler/http/handler.go

@@ -428,7 +428,7 @@ func (h *httpHandler) handleRequest(ctx context.Context, conn net.Conn, req *htt
 	start := time.Now()
 	log.Infof("%s <-> %s", conn.RemoteAddr(), addr)
 	// xnet.Transport(conn, cc)
-	xnet.Pipe(ctx, conn, cc)
+	xnet.Pipe(ctx, conn, cc, xnet.WithReadTimeout(h.md.idleTimeout))
 	log.WithFields(map[string]any{
 		"duration": time.Since(start),
 	}).Infof("%s >-< %s", conn.RemoteAddr(), addr)
@@ -694,7 +694,7 @@ func (h *httpHandler) handleUpgradeResponse(ctx context.Context, rw io.ReadWrite
 	}
 
 	// return xnet.Transport(rw, backConn)
-	return xnet.Pipe(ctx, rw, backConn)
+	return xnet.Pipe(ctx, rw, backConn, xnet.WithReadTimeout(h.md.idleTimeout))
 }
 
 func (h *httpHandler) sniffingWebsocketFrame(ctx context.Context, rw, cc io.ReadWriter, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error {
@@ -893,7 +893,7 @@ func (h *httpHandler) authenticate(ctx context.Context, conn net.Conn, req *http
 
 			req.Write(cc)
 			// xnet.Transport(conn, cc)
-			xnet.Pipe(ctx, conn, cc)
+			xnet.Pipe(ctx, conn, cc, xnet.WithReadTimeout(h.md.idleTimeout))
 			return
 		case "file":
 			f, _ := os.Open(pr.Value)

handler/http/metadata.go

@@ -21,6 +21,7 @@ const (
 
 type metadata struct {
 	readTimeout     time.Duration
+	idleTimeout     time.Duration
 	keepalive       bool
 	compression     bool
 	probeResistance *probeResistance
@@ -57,6 +58,11 @@ func (h *httpHandler) parseMetadata(md mdata.Metadata) error {
 		h.md.readTimeout = 0
 	}
 
+	h.md.idleTimeout = mdutil.GetDuration(md, "idleTimeout")
+	if h.md.idleTimeout < 0 {
+		h.md.idleTimeout = 0
+	}
+
 	if m := mdutil.GetStringMapString(md, "http.header", "header"); len(m) > 0 {
 		hd := http.Header{}
 		for k, v := range m {

internal/net/pipe.go

@@ -12,15 +12,47 @@ import (
 const (
 	// tcpWaitTimeout implements a TCP half-close timeout.
 	tcpWaitTimeout = 10 * time.Second
-	readTimeout    = 30 * time.Second
 )
 
-func Pipe(ctx context.Context, rw1, rw2 io.ReadWriteCloser) error {
+// PipeOption configures the behaviour of Pipe.
+type PipeOption func(*pipeOptions)
+
+type pipeOptions struct {
+	readTimeout time.Duration
+}
+
+// WithReadTimeout sets an idle read timeout on each pipe half. When a read
+// blocks for longer than d the connection is closed. A value of 0 disables
+// the timeout entirely, relying on TCP keepalives or context cancellation to
+// detect dead connections.
+func WithReadTimeout(d time.Duration) PipeOption {
+	return func(o *pipeOptions) {
+		o.readTimeout = d
+	}
+}
+
+// Pipe 在两个连接之间建立双向数据通道
+func Pipe(ctx context.Context, rw1, rw2 io.ReadWriteCloser, opts ...PipeOption) error {
+	var options pipeOptions
+	for _, opt := range opts {
+		opt(&options)
+	}
+
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
 	errCh := make(chan error, 2)
 
-	go func() { errCh <- pipeHalf(ctx, rw1, rw2) }()
+	// 启动两个方向的传输
-	go func() { errCh <- pipeHalf(ctx, rw2, rw1) }()
+	go func() {
+		errCh <- pipeHalf(ctx, rw1, rw2, options.readTimeout)
+	}()
 
+	go func() {
+		errCh <- pipeHalf(ctx, rw2, rw1, options.readTimeout)
+	}()
+
+	// 等待第一个错误或完成
 	var firstErr error
 	completed := 0
 	for completed < 2 {
@@ -49,7 +81,7 @@ func Pipe(ctx context.Context, rw1, rw2 io.ReadWriteCloser) error {
 }
 
 // pipeHalf 单向管道传输
-func pipeHalf(ctx context.Context, src, dst io.ReadWriteCloser) error {
+func pipeHalf(ctx context.Context, src, dst io.ReadWriteCloser, readTimeout time.Duration) error {
 	defer func() {
 		// 传输完成后执行TCP半关闭
 		halfClose(src, dst)
@@ -70,9 +102,11 @@ func pipeHalf(ctx context.Context, src, dst io.ReadWriteCloser) error {
 		case <-ctx.Done():
 			return ctx.Err()
 		default:
-			// 设置读取超时
+			// 设置读取超时 (仅当配置了超时时)
-			if rd, ok := src.(interface{ SetReadDeadline(time.Time) error }); ok {
+			if readTimeout > 0 {
-				rd.SetReadDeadline(time.Now().Add(readTimeout))
+				if rd, ok := src.(interface{ SetReadDeadline(time.Time) error }); ok {
+					rd.SetReadDeadline(time.Now().Add(readTimeout))
+				}
 			}
 
 			// 读取数据

internal/net/pipe_test.go

@@ -278,7 +278,7 @@ func Test_pipeHalf(t *testing.T) {
 	src := newPipeConn([]byte("hello world"))
 	dst := newPipeConn(nil)
 
-	err := pipeHalf(ctx, src, dst)
+	err := pipeHalf(ctx, src, dst, 0)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -294,7 +294,7 @@ func Test_pipeHalf_ReadError(t *testing.T) {
 	src.readErr = readErr
 	dst := newPipeConn(nil)
 
-	err := pipeHalf(ctx, src, dst)
+	err := pipeHalf(ctx, src, dst, 0)
 	if err == nil {
 		t.Error("expected error, got nil")
 	}
@@ -306,7 +306,7 @@ func Test_pipeHalf_WriteError(t *testing.T) {
 	dst := newPipeConn(nil)
 	dst.writeClosed = true
 
-	err := pipeHalf(ctx, src, dst)
+	err := pipeHalf(ctx, src, dst, 0)
 	if err == nil {
 		t.Error("expected error, got nil")
 	}
@@ -320,7 +320,7 @@ func Test_pipeHalf_ContextCanceled(t *testing.T) {
 	src.eofAfter = -1
 	dst := newPipeConn(nil)
 
-	err := pipeHalf(ctx, src, dst)
+	err := pipeHalf(ctx, src, dst, 0)
 	if err == nil {
 		t.Error("expected context error, got nil")
 	}