Make pipe read timeout configurable via idleTimeout metadata
The 30s hardcoded readTimeout in Pipe() caused all CONNECT tunnel connections to be hard-closed after 30s of inactivity, breaking WebSocket and long-polling connections through the proxy. Pipe() now accepts a WithReadTimeout(d) option. When d is 0 (the default) no read deadline is set, relying on TCP keepalives or context cancellation to detect dead connections instead. The HTTP handler exposes this as the idleTimeout metadata key. Fixes: https://github.com/go-gost/x/issues/91
This commit is contained in:
@@ -428,7 +428,7 @@ func (h *httpHandler) handleRequest(ctx context.Context, conn net.Conn, req *htt
|
|||||||
start := time.Now()
|
start := time.Now()
|
||||||
log.Infof("%s <-> %s", conn.RemoteAddr(), addr)
|
log.Infof("%s <-> %s", conn.RemoteAddr(), addr)
|
||||||
// xnet.Transport(conn, cc)
|
// xnet.Transport(conn, cc)
|
||||||
xnet.Pipe(ctx, conn, cc)
|
xnet.Pipe(ctx, conn, cc, xnet.WithReadTimeout(h.md.idleTimeout))
|
||||||
log.WithFields(map[string]any{
|
log.WithFields(map[string]any{
|
||||||
"duration": time.Since(start),
|
"duration": time.Since(start),
|
||||||
}).Infof("%s >-< %s", conn.RemoteAddr(), addr)
|
}).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.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 {
|
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)
|
req.Write(cc)
|
||||||
// xnet.Transport(conn, cc)
|
// xnet.Transport(conn, cc)
|
||||||
xnet.Pipe(ctx, conn, cc)
|
xnet.Pipe(ctx, conn, cc, xnet.WithReadTimeout(h.md.idleTimeout))
|
||||||
return
|
return
|
||||||
case "file":
|
case "file":
|
||||||
f, _ := os.Open(pr.Value)
|
f, _ := os.Open(pr.Value)
|
||||||
|
|||||||
@@ -21,6 +21,7 @@ const (
|
|||||||
|
|
||||||
type metadata struct {
|
type metadata struct {
|
||||||
readTimeout time.Duration
|
readTimeout time.Duration
|
||||||
|
idleTimeout time.Duration
|
||||||
keepalive bool
|
keepalive bool
|
||||||
compression bool
|
compression bool
|
||||||
probeResistance *probeResistance
|
probeResistance *probeResistance
|
||||||
@@ -57,6 +58,11 @@ func (h *httpHandler) parseMetadata(md mdata.Metadata) error {
|
|||||||
h.md.readTimeout = 0
|
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 {
|
if m := mdutil.GetStringMapString(md, "http.header", "header"); len(m) > 0 {
|
||||||
hd := http.Header{}
|
hd := http.Header{}
|
||||||
for k, v := range m {
|
for k, v := range m {
|
||||||
|
|||||||
+29
-7
@@ -12,11 +12,32 @@ import (
|
|||||||
const (
|
const (
|
||||||
// tcpWaitTimeout implements a TCP half-close timeout.
|
// tcpWaitTimeout implements a TCP half-close timeout.
|
||||||
tcpWaitTimeout = 10 * time.Second
|
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 在两个连接之间建立双向数据通道
|
// 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)
|
ctx, cancel := context.WithCancel(ctx)
|
||||||
defer cancel()
|
defer cancel()
|
||||||
|
|
||||||
@@ -24,11 +45,11 @@ func Pipe(ctx context.Context, rw1, rw2 io.ReadWriteCloser) error {
|
|||||||
|
|
||||||
// 启动两个方向的传输
|
// 启动两个方向的传输
|
||||||
go func() {
|
go func() {
|
||||||
errCh <- pipeHalf(ctx, rw1, rw2)
|
errCh <- pipeHalf(ctx, rw1, rw2, options.readTimeout)
|
||||||
}()
|
}()
|
||||||
|
|
||||||
go func() {
|
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 单向管道传输
|
// 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() {
|
defer func() {
|
||||||
// 传输完成后执行TCP半关闭
|
// 传输完成后执行TCP半关闭
|
||||||
halfClose(src, dst)
|
halfClose(src, dst)
|
||||||
@@ -60,7 +81,6 @@ func pipeHalf(ctx context.Context, src, dst io.ReadWriteCloser) error {
|
|||||||
buf := bufpool.Get(bufferSize / 2)
|
buf := bufpool.Get(bufferSize / 2)
|
||||||
defer bufpool.Put(buf)
|
defer bufpool.Put(buf)
|
||||||
|
|
||||||
|
|
||||||
// 创建带超时的读取器
|
// 创建带超时的读取器
|
||||||
reader := &readDeadliner{
|
reader := &readDeadliner{
|
||||||
Reader: src,
|
Reader: src,
|
||||||
@@ -73,10 +93,12 @@ func pipeHalf(ctx context.Context, src, dst io.ReadWriteCloser) error {
|
|||||||
case <-ctx.Done():
|
case <-ctx.Done():
|
||||||
return ctx.Err()
|
return ctx.Err()
|
||||||
default:
|
default:
|
||||||
// 设置读取超时
|
// 设置读取超时 (仅当配置了超时时)
|
||||||
|
if readTimeout > 0 {
|
||||||
if rd, ok := src.(interface{ SetReadDeadline(time.Time) error }); ok {
|
if rd, ok := src.(interface{ SetReadDeadline(time.Time) error }); ok {
|
||||||
rd.SetReadDeadline(time.Now().Add(readTimeout))
|
rd.SetReadDeadline(time.Now().Add(readTimeout))
|
||||||
}
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// 读取数据
|
// 读取数据
|
||||||
nr, err := reader.Read(buf)
|
nr, err := reader.Read(buf)
|
||||||
|
|||||||
Reference in New Issue
Block a user