From db21de831affd8a4f6b1cdc4ad6672b8b9b4e056 Mon Sep 17 00:00:00 2001 From: ginuerzh Date: Fri, 1 Aug 2025 23:00:50 +0800 Subject: [PATCH] fix xnet.Pipe --- admission/wrapper/conn.go | 4 ++-- connector/relay/bind.go | 4 ++-- handler/http/handler.go | 6 +++--- handler/http2/handler.go | 2 +- handler/relay/bind.go | 4 ++-- handler/tunnel/entrypoint.go | 8 ++++---- handler/unix/handler.go | 4 ++-- handler/unix/sniffing.go | 6 +++--- internal/io/io.go | 23 +++++++++++++++++++++-- internal/net/net.go | 4 ++-- internal/net/pipe.go | 11 ++++++++--- internal/net/proxyproto/conn.go | 29 +++++++++++++++++++++++++++++ internal/net/proxyproto/listener.go | 20 +++++++++++++++++--- internal/util/forwarder/sniffer.go | 8 ++++---- internal/util/sniffing/sniffer.go | 8 ++++---- internal/util/ss/ss.go | 4 ++-- limiter/conn/wrapper/conn.go | 4 ++-- limiter/traffic/wrapper/conn.go | 6 +++--- metrics/wrapper/conn.go | 4 ++-- observer/stats/wrapper/conn.go | 4 ++-- recorder/http.go | 2 +- service/service.go | 8 ++++++-- 22 files changed, 122 insertions(+), 51 deletions(-) diff --git a/admission/wrapper/conn.go b/admission/wrapper/conn.go index 9f8fb84c..d143cfe7 100644 --- a/admission/wrapper/conn.go +++ b/admission/wrapper/conn.go @@ -62,14 +62,14 @@ func (c *serverConn) CloseRead() error { if sc, ok := c.Conn.(xio.CloseRead); ok { return sc.CloseRead() } - return nil + return xio.ErrUnsupported } func (c *serverConn) CloseWrite() error { if sc, ok := c.Conn.(xio.CloseWrite); ok { return sc.CloseWrite() } - return nil + return xio.ErrUnsupported } type packetConn struct { diff --git a/connector/relay/bind.go b/connector/relay/bind.go index 248a0ab7..cc0f994c 100644 --- a/connector/relay/bind.go +++ b/connector/relay/bind.go @@ -44,7 +44,7 @@ func (c *relayConnector) bindTCP(ctx context.Context, conn net.Conn, network, ad if err != nil { return nil, err } - log.Debugf("bind on %s/%s OK", laddr, laddr.Network()) + log.Infof("bind on %s/%s OK", laddr, laddr.Network()) session, err := mux.ServerSession(conn, c.md.muxCfg) if err != nil { @@ -63,7 +63,7 @@ func (c *relayConnector) bindUDP(ctx context.Context, conn net.Conn, network, ad if err != nil { return nil, err } - log.Debugf("bind on %s/%s OK", laddr, laddr.Network()) + log.Infof("bind on %s/%s OK", laddr, laddr.Network()) ln := udp.NewListener( relay_util.UDPTunClientPacketConn(conn), diff --git a/handler/http/handler.go b/handler/http/handler.go index 62cabd7a..ddcd4519 100644 --- a/handler/http/handler.go +++ b/handler/http/handler.go @@ -462,13 +462,13 @@ func (h *httpHandler) handleProxy(ctx context.Context, conn net.Conn, req *http. log.Trace(string(dump)) } - if close, err := h.proxyRoundTrip(ctx, xio.NewReadWriter(br, conn), req, ro, &pStats, log); err != nil || close { + if close, err := h.proxyRoundTrip(ctx, xio.NewReadWriteCloser(br, conn, conn), req, ro, &pStats, log); err != nil || close { return err } } } -func (h *httpHandler) proxyRoundTrip(ctx context.Context, rw io.ReadWriter, req *http.Request, ro *xrecorder.HandlerRecorderObject, pStats stats.Stats, log logger.Logger) (close bool, err error) { +func (h *httpHandler) proxyRoundTrip(ctx context.Context, rw io.ReadWriteCloser, req *http.Request, ro *xrecorder.HandlerRecorderObject, pStats stats.Stats, log logger.Logger) (close bool, err error) { close = true ro2 := &xrecorder.HandlerRecorderObject{} @@ -682,7 +682,7 @@ func upgradeType(h http.Header) string { return h.Get("Upgrade") } -func (h *httpHandler) handleUpgradeResponse(ctx context.Context, rw io.ReadWriter, req *http.Request, res *http.Response, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error { +func (h *httpHandler) handleUpgradeResponse(ctx context.Context, rw io.ReadWriteCloser, req *http.Request, res *http.Response, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error { reqUpType := upgradeType(req.Header) resUpType := upgradeType(res.Header) if !strings.EqualFold(reqUpType, resUpType) { diff --git a/handler/http2/handler.go b/handler/http2/handler.go index 670aad68..a80c3e83 100644 --- a/handler/http2/handler.go +++ b/handler/http2/handler.go @@ -316,7 +316,7 @@ func (h *http2Handler) roundTrip(ctx context.Context, w http.ResponseWriter, req start := time.Now() log.Infof("%s <-> %s", req.RemoteAddr, host) // xnet.Transport(rw, cc) - xnet.Pipe(ctx, rw, cc) + xnet.Pipe(ctx, xio.NewReadWriteCloser(rw, rw, req.Body), cc) log.WithFields(map[string]any{ "duration": time.Since(start), }).Infof("%s >-< %s", req.RemoteAddr, host) diff --git a/handler/relay/bind.go b/handler/relay/bind.go index fab19d34..3d88edc7 100644 --- a/handler/relay/bind.go +++ b/handler/relay/bind.go @@ -143,7 +143,7 @@ func (h *relayHandler) bindTCP(ctx context.Context, conn net.Conn, network, addr ) log = log.WithFields(map[string]any{}) - log.Debugf("bind on %s/%s OK", ln.Addr(), ln.Addr().Network()) + log.Infof("bind on %s/%s OK", ln.Addr(), ln.Addr().Network()) go func() { defer srv.Close() @@ -201,7 +201,7 @@ func (h *relayHandler) bindUDP(ctx context.Context, conn net.Conn, network, addr return err } - log.Debugf("bind on %s OK", pc.LocalAddr()) + log.Infof("bind on %s OK", pc.LocalAddr()) r := udp.NewRelay(relay_util.UDPTunServerConn(conn), pc). WithBypass(h.options.Bypass). diff --git a/handler/tunnel/entrypoint.go b/handler/tunnel/entrypoint.go index 88e87a94..7a2564fd 100644 --- a/handler/tunnel/entrypoint.go +++ b/handler/tunnel/entrypoint.go @@ -224,7 +224,7 @@ func (ep *entrypoint) handleHTTP(ctx context.Context, conn net.Conn, ro *xrecord ro.Time = time.Time{} - if err := ep.httpRoundTrip(ctx, xio.NewReadWriter(br, conn), req, ro, &pStats, log); err != nil { + if err := ep.httpRoundTrip(ctx, xio.NewReadWriteCloser(br, conn, conn), req, ro, &pStats, log); err != nil { log.Error(err) return err } @@ -245,13 +245,13 @@ func (ep *entrypoint) handleHTTP(ctx context.Context, conn net.Conn, ro *xrecord log.Trace(string(dump)) } - if err := ep.httpRoundTrip(ctx, xio.NewReadWriter(br, conn), req, ro, &pStats, log); err != nil { + if err := ep.httpRoundTrip(ctx, xio.NewReadWriteCloser(br, conn, conn), req, ro, &pStats, log); err != nil { return err } } } -func (ep *entrypoint) httpRoundTrip(ctx context.Context, rw io.ReadWriter, req *http.Request, ro *xrecorder.HandlerRecorderObject, pStats stats.Stats, log logger.Logger) (err error) { +func (ep *entrypoint) httpRoundTrip(ctx context.Context, rw io.ReadWriteCloser, req *http.Request, ro *xrecorder.HandlerRecorderObject, pStats stats.Stats, log logger.Logger) (err error) { ro2 := &xrecorder.HandlerRecorderObject{} *ro2 = *ro ro = ro2 @@ -413,7 +413,7 @@ func upgradeType(h http.Header) string { return h.Get("Upgrade") } -func (ep *entrypoint) handleUpgradeResponse(ctx context.Context, rw io.ReadWriter, req *http.Request, res *http.Response, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error { +func (ep *entrypoint) handleUpgradeResponse(ctx context.Context, rw io.ReadWriteCloser, req *http.Request, res *http.Response, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error { reqUpType := upgradeType(req.Header) resUpType := upgradeType(res.Header) if !strings.EqualFold(reqUpType, resUpType) { diff --git a/handler/unix/handler.go b/handler/unix/handler.go index 18d50bd8..af48a36f 100644 --- a/handler/unix/handler.go +++ b/handler/unix/handler.go @@ -218,7 +218,7 @@ func (h *unixHandler) forwardUnix(ctx context.Context, conn net.Conn, target *ch } defer cc.Close() - var rw io.ReadWriter = conn + var rw io.ReadWriteCloser = conn if h.md.sniffing { if h.md.sniffingTimeout > 0 { conn.SetReadDeadline(time.Now().Add(h.md.sniffingTimeout)) @@ -232,7 +232,7 @@ func (h *unixHandler) forwardUnix(ctx context.Context, conn net.Conn, target *ch conn.SetReadDeadline(time.Time{}) } - rw = xio.NewReadWriter(br, conn) + rw = xio.NewReadWriteCloser(br, conn, conn) switch proto { case sniffing.ProtoHTTP: ro2 := &xrecorder.HandlerRecorderObject{} diff --git a/handler/unix/sniffing.go b/handler/unix/sniffing.go index 3bafa5b4..696b9e7e 100644 --- a/handler/unix/sniffing.go +++ b/handler/unix/sniffing.go @@ -28,7 +28,7 @@ const ( defaultBodySize = 1024 * 1024 // 1MB ) -func (h *unixHandler) handleHTTP(ctx context.Context, rw, cc io.ReadWriter, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error { +func (h *unixHandler) handleHTTP(ctx context.Context, rw, cc io.ReadWriteCloser, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error { br := bufio.NewReader(rw) req, err := http.ReadRequest(br) if err != nil { @@ -81,7 +81,7 @@ func (h *unixHandler) handleHTTP(ctx context.Context, rw, cc io.ReadWriter, ro * } } -func (h *unixHandler) httpRoundTrip(ctx context.Context, rw, cc io.ReadWriter, req *http.Request, ro *xrecorder.HandlerRecorderObject, log logger.Logger) (close bool, err error) { +func (h *unixHandler) httpRoundTrip(ctx context.Context, rw, cc io.ReadWriteCloser, req *http.Request, ro *xrecorder.HandlerRecorderObject, log logger.Logger) (close bool, err error) { close = true if req == nil { @@ -206,7 +206,7 @@ func (h *unixHandler) httpRoundTrip(ctx context.Context, rw, cc io.ReadWriter, r return resp.Close, nil } -func (h *unixHandler) handleTLS(ctx context.Context, rw, cc io.ReadWriter, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error { +func (h *unixHandler) handleTLS(ctx context.Context, rw, cc io.ReadWriteCloser, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error { buf := new(bytes.Buffer) clientHello, err := dissector.ParseClientHello(io.TeeReader(rw, buf)) diff --git a/internal/io/io.go b/internal/io/io.go index af6593e3..a2208e4c 100644 --- a/internal/io/io.go +++ b/internal/io/io.go @@ -1,10 +1,15 @@ package io import ( + "errors" "io" "time" ) +var ( + ErrUnsupported = errors.New("unsupported") +) + type CloseRead interface { CloseRead() error } @@ -29,14 +34,14 @@ func (rw *readWriter) CloseRead() error { if sc, ok := rw.Writer.(CloseRead); ok { return sc.CloseRead() } - return nil + return ErrUnsupported } func (rw *readWriter) CloseWrite() error { if sc, ok := rw.Writer.(CloseWrite); ok { return sc.CloseWrite() } - return nil + return ErrUnsupported } type readWriteCloser struct { @@ -53,6 +58,20 @@ func NewReadWriteCloser(r io.Reader, w io.Writer, c io.Closer) io.ReadWriteClose } } +func (rwc *readWriteCloser) CloseRead() error { + if sc, ok := rwc.Writer.(CloseRead); ok { + return sc.CloseRead() + } + return ErrUnsupported +} + +func (rwc *readWriteCloser) CloseWrite() error { + if sc, ok := rwc.Writer.(CloseWrite); ok { + return sc.CloseWrite() + } + return ErrUnsupported +} + type setReadDeadline interface { SetReadDeadline(t time.Time) error } diff --git a/internal/net/net.go b/internal/net/net.go index d5354589..41f882be 100644 --- a/internal/net/net.go +++ b/internal/net/net.go @@ -129,12 +129,12 @@ func (c *readWriteConn) CloseRead() error { if sc, ok := c.Conn.(xio.CloseRead); ok { return sc.CloseRead() } - return nil + return xio.ErrUnsupported } func (c *readWriteConn) CloseWrite() error { if sc, ok := c.Conn.(xio.CloseWrite); ok { return sc.CloseWrite() } - return nil + return xio.ErrUnsupported } diff --git a/internal/net/pipe.go b/internal/net/pipe.go index 890cc6fe..fedd0006 100644 --- a/internal/net/pipe.go +++ b/internal/net/pipe.go @@ -15,7 +15,7 @@ const ( tcpWaitTimeout = 10 * time.Second ) -func Pipe(ctx context.Context, rw1, rw2 io.ReadWriter) error { +func Pipe(ctx context.Context, rw1, rw2 io.ReadWriteCloser) error { wg := sync.WaitGroup{} wg.Add(2) @@ -55,7 +55,7 @@ func Pipe(ctx context.Context, rw1, rw2 io.ReadWriter) error { return nil } -func pipeBuffer(dst io.ReadWriter, src io.ReadWriter, bufferSize int) error { +func pipeBuffer(dst io.ReadWriteCloser, src io.ReadWriteCloser, bufferSize int) error { buf := bufpool.Get(bufferSize) defer bufpool.Put(buf) @@ -65,8 +65,13 @@ func pipeBuffer(dst io.ReadWriter, src io.ReadWriter, bufferSize int) error { if cr, ok := src.(xio.CloseRead); ok { cr.CloseRead() } + if cw, ok := dst.(xio.CloseWrite); ok { - cw.CloseWrite() + if e := cw.CloseWrite(); e == xio.ErrUnsupported { + dst.Close() + } + } else { + dst.Close() } // Set TCP half-close timeout. diff --git a/internal/net/proxyproto/conn.go b/internal/net/proxyproto/conn.go index dbf88d10..acb525a7 100644 --- a/internal/net/proxyproto/conn.go +++ b/internal/net/proxyproto/conn.go @@ -3,9 +3,38 @@ package proxyproto import ( "net" + xio "github.com/go-gost/x/internal/io" proxyproto "github.com/pires/go-proxyproto" ) +type serverConn struct { + net.Conn +} + +func (c *serverConn) CloseRead() error { + if sc, ok := c.Conn.(xio.CloseRead); ok { + return sc.CloseRead() + } + if conn, ok := c.Conn.(*proxyproto.Conn); ok { + if tcpConn, ok := conn.TCPConn(); ok { + return tcpConn.CloseRead() + } + } + return xio.ErrUnsupported +} + +func (c *serverConn) CloseWrite() error { + if sc, ok := c.Conn.(xio.CloseWrite); ok { + return sc.CloseWrite() + } + if conn, ok := c.Conn.(*proxyproto.Conn); ok { + if tcpConn, ok := conn.TCPConn(); ok { + return tcpConn.CloseWrite() + } + } + return xio.ErrUnsupported +} + func WrapClientConn(ppv int, src, dst net.Addr, c net.Conn) net.Conn { if ppv <= 0 { return c diff --git a/internal/net/proxyproto/listener.go b/internal/net/proxyproto/listener.go index 18d45175..cd043175 100644 --- a/internal/net/proxyproto/listener.go +++ b/internal/net/proxyproto/listener.go @@ -7,13 +7,27 @@ import ( proxyproto "github.com/pires/go-proxyproto" ) +type listener struct { + net.Listener +} + +func (ln *listener) Accept() (net.Conn, error) { + conn, err := ln.Listener.Accept() + if err != nil { + return nil, err + } + return &serverConn{Conn: conn}, nil +} + func WrapListener(ppv int, ln net.Listener, readHeaderTimeout time.Duration) net.Listener { if ppv <= 0 { return ln } - return &proxyproto.Listener{ - Listener: ln, - ReadHeaderTimeout: readHeaderTimeout, + return &listener{ + Listener: &proxyproto.Listener{ + Listener: ln, + ReadHeaderTimeout: readHeaderTimeout, + }, } } diff --git a/internal/util/forwarder/sniffer.go b/internal/util/forwarder/sniffer.go index 1398748e..86f529d3 100644 --- a/internal/util/forwarder/sniffer.go +++ b/internal/util/forwarder/sniffer.go @@ -191,7 +191,7 @@ func (h *Sniffer) HandleHTTP(ctx context.Context, conn net.Conn, opts ...HandleO ro.Dst = cc.RemoteAddr().String() ro.Time = time.Time{} - shouldClose, err := h.httpRoundTrip(ctx, xio.NewReadWriter(br, conn), cc, node, req, &pStats, &ho) + shouldClose, err := h.httpRoundTrip(ctx, xio.NewReadWriteCloser(br, conn, conn), cc, node, req, &pStats, &ho) if err != nil || shouldClose { return err } @@ -212,7 +212,7 @@ func (h *Sniffer) HandleHTTP(ctx context.Context, conn net.Conn, opts ...HandleO log.Trace(string(dump)) } - if shouldClose, err := h.httpRoundTrip(ctx, xio.NewReadWriter(br, conn), cc, node, req, &pStats, &ho); err != nil || shouldClose { + if shouldClose, err := h.httpRoundTrip(ctx, xio.NewReadWriteCloser(br, conn, conn), cc, node, req, &pStats, &ho); err != nil || shouldClose { return err } } @@ -365,7 +365,7 @@ func (h *Sniffer) serveH2(ctx context.Context, conn net.Conn, ho *HandleOptions) return nil } -func (h *Sniffer) httpRoundTrip(ctx context.Context, rw, cc io.ReadWriter, node *chain.Node, req *http.Request, pStats stats.Stats, ho *HandleOptions) (close bool, err error) { +func (h *Sniffer) httpRoundTrip(ctx context.Context, rw, cc io.ReadWriteCloser, node *chain.Node, req *http.Request, pStats stats.Stats, ho *HandleOptions) (close bool, err error) { close = true log := ho.Log @@ -586,7 +586,7 @@ func upgradeType(h http.Header) string { return h.Get("Upgrade") } -func (h *Sniffer) handleUpgradeResponse(ctx context.Context, rw io.ReadWriter, cc io.ReadWriter, req *http.Request, res *http.Response, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error { +func (h *Sniffer) handleUpgradeResponse(ctx context.Context, rw, cc io.ReadWriteCloser, req *http.Request, res *http.Response, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error { reqUpType := upgradeType(req.Header) resUpType := upgradeType(res.Header) if !strings.EqualFold(reqUpType, resUpType) { diff --git a/internal/util/sniffing/sniffer.go b/internal/util/sniffing/sniffer.go index 9980bd83..b769127d 100644 --- a/internal/util/sniffing/sniffer.go +++ b/internal/util/sniffing/sniffer.go @@ -198,7 +198,7 @@ func (h *Sniffer) HandleHTTP(ctx context.Context, conn net.Conn, opts ...HandleO ro.Dst = cc.RemoteAddr().String() ro.Time = time.Time{} - shouldClose, err := h.httpRoundTrip(ctx, xio.NewReadWriter(br, conn), cc, req, ro, &pStats, log) + shouldClose, err := h.httpRoundTrip(ctx, xio.NewReadWriteCloser(br, conn, conn), cc, req, ro, &pStats, log) if err != nil || shouldClose { return err } @@ -219,7 +219,7 @@ func (h *Sniffer) HandleHTTP(ctx context.Context, conn net.Conn, opts ...HandleO log.Trace(string(dump)) } - if shouldClose, err := h.httpRoundTrip(ctx, xio.NewReadWriter(br, conn), cc, req, ro, &pStats, log); err != nil || shouldClose { + if shouldClose, err := h.httpRoundTrip(ctx, xio.NewReadWriteCloser(br, conn, conn), cc, req, ro, &pStats, log); err != nil || shouldClose { return err } } @@ -277,7 +277,7 @@ func (h *Sniffer) serveH2(ctx context.Context, conn net.Conn, ho *HandleOptions) return nil } -func (h *Sniffer) httpRoundTrip(ctx context.Context, rw, cc io.ReadWriter, req *http.Request, ro *xrecorder.HandlerRecorderObject, pStats stats.Stats, log logger.Logger) (close bool, err error) { +func (h *Sniffer) httpRoundTrip(ctx context.Context, rw, cc io.ReadWriteCloser, req *http.Request, ro *xrecorder.HandlerRecorderObject, pStats stats.Stats, log logger.Logger) (close bool, err error) { close = true ro2 := &xrecorder.HandlerRecorderObject{} @@ -433,7 +433,7 @@ func upgradeType(h http.Header) string { return h.Get("Upgrade") } -func (h *Sniffer) handleUpgradeResponse(ctx context.Context, rw io.ReadWriter, cc io.ReadWriter, req *http.Request, res *http.Response, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error { +func (h *Sniffer) handleUpgradeResponse(ctx context.Context, rw, cc io.ReadWriteCloser, req *http.Request, res *http.Response, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error { reqUpType := upgradeType(req.Header) resUpType := upgradeType(res.Header) if !strings.EqualFold(reqUpType, resUpType) { diff --git a/internal/util/ss/ss.go b/internal/util/ss/ss.go index 9ad2e22e..dd4ceec7 100644 --- a/internal/util/ss/ss.go +++ b/internal/util/ss/ss.go @@ -46,12 +46,12 @@ func (c *shadowConn) CloseRead() error { if sc, ok := c.Conn.(xio.CloseRead); ok { return sc.CloseRead() } - return nil + return xio.ErrUnsupported } func (c *shadowConn) CloseWrite() error { if sc, ok := c.Conn.(xio.CloseWrite); ok { return sc.CloseWrite() } - return nil + return xio.ErrUnsupported } diff --git a/limiter/conn/wrapper/conn.go b/limiter/conn/wrapper/conn.go index d86243b7..5f6ed359 100644 --- a/limiter/conn/wrapper/conn.go +++ b/limiter/conn/wrapper/conn.go @@ -55,12 +55,12 @@ func (c *serverConn) CloseRead() error { if sc, ok := c.Conn.(xio.CloseRead); ok { return sc.CloseRead() } - return nil + return xio.ErrUnsupported } func (c *serverConn) CloseWrite() error { if sc, ok := c.Conn.(xio.CloseWrite); ok { return sc.CloseWrite() } - return nil + return xio.ErrUnsupported } diff --git a/limiter/traffic/wrapper/conn.go b/limiter/traffic/wrapper/conn.go index e57e7471..30ce905e 100644 --- a/limiter/traffic/wrapper/conn.go +++ b/limiter/traffic/wrapper/conn.go @@ -11,9 +11,9 @@ import ( "github.com/go-gost/core/limiter" "github.com/go-gost/core/limiter/traffic" "github.com/go-gost/core/metadata" + xio "github.com/go-gost/x/internal/io" xnet "github.com/go-gost/x/internal/net" "github.com/go-gost/x/internal/net/udp" - xio "github.com/go-gost/x/internal/io" ) var ( @@ -118,14 +118,14 @@ func (c *limitConn) CloseRead() error { if sc, ok := c.Conn.(xio.CloseRead); ok { return sc.CloseRead() } - return nil + return xio.ErrUnsupported } func (c *limitConn) CloseWrite() error { if sc, ok := c.Conn.(xio.CloseWrite); ok { return sc.CloseWrite() } - return nil + return xio.ErrUnsupported } type packetConn struct { diff --git a/metrics/wrapper/conn.go b/metrics/wrapper/conn.go index 307fa56d..3df0e45d 100644 --- a/metrics/wrapper/conn.go +++ b/metrics/wrapper/conn.go @@ -92,14 +92,14 @@ func (c *serverConn) CloseRead() error { if sc, ok := c.Conn.(xio.CloseRead); ok { return sc.CloseRead() } - return nil + return xio.ErrUnsupported } func (c *serverConn) CloseWrite() error { if sc, ok := c.Conn.(xio.CloseWrite); ok { return sc.CloseWrite() } - return nil + return xio.ErrUnsupported } type packetConn struct { diff --git a/observer/stats/wrapper/conn.go b/observer/stats/wrapper/conn.go index 9ad45abe..cd9a42b3 100644 --- a/observer/stats/wrapper/conn.go +++ b/observer/stats/wrapper/conn.go @@ -94,14 +94,14 @@ func (c *conn) CloseRead() error { if sc, ok := c.Conn.(xio.CloseRead); ok { return sc.CloseRead() } - return nil + return xio.ErrUnsupported } func (c *conn) CloseWrite() error { if sc, ok := c.Conn.(xio.CloseWrite); ok { return sc.CloseWrite() } - return nil + return xio.ErrUnsupported } type packetConn struct { diff --git a/recorder/http.go b/recorder/http.go index 4af224c5..b6cf60fe 100644 --- a/recorder/http.go +++ b/recorder/http.go @@ -78,7 +78,7 @@ func (r *httpRecorder) Record(ctx context.Context, b []byte, opts ...recorder.Re } if r.header != nil { - req.Header = r.header + req.Header = r.header.Clone() } if req.Header.Get("Content-Type") == "" { diff --git a/service/service.go b/service/service.go index 2451fab3..94fdb6ac 100644 --- a/service/service.go +++ b/service/service.go @@ -160,6 +160,8 @@ func (s *defaultService) Serve() error { defer v.Dec() } + log := s.options.logger + var wg sync.WaitGroup defer wg.Wait() @@ -184,16 +186,18 @@ func (s *defaultService) Serve() error { s.setState(StateFailed) - s.options.logger.Warnf("accept: %v, retrying in %v", e, tempDelay) + log.Warnf("accept: %v, retrying in %v", e, tempDelay) time.Sleep(tempDelay) continue } s.setState(StateClosed) if !errors.Is(e, net.ErrClosed) { - s.options.logger.Errorf("accept: %v", e) + log.Errorf("accept: %v", e) } + log.Debugf("service %s exited!", s.name) + return e }