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 205bb401..9a20bac7 100644 --- a/internal/net/pipe.go +++ b/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 { - 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)) + } } // 读取数据 diff --git a/internal/net/pipe_test.go b/internal/net/pipe_test.go index e9abcb60..f8403e84 100644 --- a/internal/net/pipe_test.go +++ b/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") }