diff --git a/handler/http/handler.go b/handler/http/handler.go index 33905b84..65a6f7ee 100644 --- a/handler/http/handler.go +++ b/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) diff --git a/handler/http/metadata.go b/handler/http/metadata.go index b15a83a8..abd159a0 100644 --- a/handler/http/metadata.go +++ b/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 { diff --git a/internal/net/pipe.go b/internal/net/pipe.go index 94270648..d0db7aa2 100644 --- a/internal/net/pipe.go +++ b/internal/net/pipe.go @@ -12,11 +12,32 @@ import ( const ( // tcpWaitTimeout implements a TCP half-close timeout. tcpWaitTimeout = 10 * time.Second - readTimeout = 30 * time.Second ) +// 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) error { +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() @@ -24,11 +45,11 @@ func Pipe(ctx context.Context, rw1, rw2 io.ReadWriteCloser) error { // 启动两个方向的传输 go func() { - errCh <- pipeHalf(ctx, rw1, rw2) + errCh <- pipeHalf(ctx, rw1, rw2, options.readTimeout) }() go func() { - errCh <- pipeHalf(ctx, rw2, rw1) + errCh <- pipeHalf(ctx, rw2, rw1, options.readTimeout) }() // 等待第一个错误或完成 @@ -51,7 +72,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) @@ -60,7 +81,6 @@ func pipeHalf(ctx context.Context, src, dst io.ReadWriteCloser) error { buf := bufpool.Get(bufferSize / 2) defer bufpool.Put(buf) - // 创建带超时的读取器 reader := &readDeadliner{ Reader: src, @@ -73,9 +93,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 { - rd.SetReadDeadline(time.Now().Add(readTimeout)) + // 设置读取超时 (仅当配置了超时时) + if readTimeout > 0 { + if rd, ok := src.(interface{ SetReadDeadline(time.Time) error }); ok { + rd.SetReadDeadline(time.Now().Add(readTimeout)) + } } // 读取数据