replace xnet.Transport by xnet.Pipe

admission/wrapper/conn.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/go-gost/core/admission"
 	"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"
 )
@@ -57,6 +58,20 @@ func (c *serverConn) Metadata() metadata.Metadata {
 	return nil
 }
 
+func (c *serverConn) CloseRead() error {
+	if sc, ok := c.Conn.(xio.CloseRead); ok {
+		return sc.CloseRead()
+	}
+	return nil
+}
+
+func (c *serverConn) CloseWrite() error {
+	if sc, ok := c.Conn.(xio.CloseWrite); ok {
+		return sc.CloseWrite()
+	}
+	return nil
+}
+
 type packetConn struct {
 	net.PacketConn
 	admission admission.Admission

handler/dns/handler.go

@@ -142,6 +142,7 @@ func (h *dnsHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler.
 		Network:    conn.LocalAddr().Network(),
 		RemoteAddr: conn.RemoteAddr().String(),
 		LocalAddr:  conn.LocalAddr().String(),
+		Host:       conn.LocalAddr().String(),
 		Proto:      "dns",
 		Time:       start,
 		SID:        string(ctxvalue.SidFromContext(ctx)),

handler/forward/local/handler.go

@@ -252,7 +252,8 @@ func (h *forwardHandler) Handle(ctx context.Context, conn net.Conn, opts ...hand
 
 	t := time.Now()
 	log.Infof("%s <-> %s", conn.RemoteAddr(), target.Addr)
-	xnet.Transport(conn, cc)
+	// xnet.Transport(conn, cc)
+	xnet.Pipe(ctx, conn, cc)
 	log.WithFields(map[string]any{
 		"duration": time.Since(t),
 	}).Infof("%s >-< %s", conn.RemoteAddr(), target.Addr)

handler/forward/remote/handler.go

@@ -264,7 +264,8 @@ func (h *forwardHandler) Handle(ctx context.Context, conn net.Conn, opts ...hand
 
 	t := time.Now()
 	log.Infof("%s <-> %s", conn.RemoteAddr(), target.Addr)
-	xnet.Transport(conn, cc)
+	// xnet.Transport(conn, cc)
+	xnet.Pipe(ctx, conn, cc)
 	log.WithFields(map[string]any{
 		"duration": time.Since(t),
 	}).Infof("%s >-< %s", conn.RemoteAddr(), target.Addr)

handler/http/handler.go

@@ -427,7 +427,8 @@ 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.Transport(conn, cc)
+	xnet.Pipe(ctx, conn, cc)
 	log.WithFields(map[string]any{
 		"duration": time.Since(start),
 	}).Infof("%s >-< %s", conn.RemoteAddr(), addr)
@@ -703,7 +704,8 @@ func (h *httpHandler) handleUpgradeResponse(ctx context.Context, rw io.ReadWrite
 		return h.sniffingWebsocketFrame(ctx, rw, backConn, ro, log)
 	}
 
-	return xnet.Transport(rw, backConn)
+	// return xnet.Transport(rw, backConn)
+	return xnet.Pipe(ctx, rw, backConn)
 }
 
 func (h *httpHandler) sniffingWebsocketFrame(ctx context.Context, rw, cc io.ReadWriter, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error {
@@ -901,7 +903,8 @@ func (h *httpHandler) authenticate(ctx context.Context, conn net.Conn, req *http
 			defer cc.Close()
 
 			req.Write(cc)
-			xnet.Transport(conn, cc)
+			// xnet.Transport(conn, cc)
+			xnet.Pipe(ctx, conn, cc)
 			return
 		case "file":
 			f, _ := os.Open(pr.Value)

handler/http2/handler.go

@@ -315,7 +315,8 @@ 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.Transport(rw, cc)
+		xnet.Pipe(ctx, rw, cc)
 		log.WithFields(map[string]any{
 			"duration": time.Since(start),
 		}).Infof("%s >-< %s", req.RemoteAddr, host)

handler/redirect/tcp/handler.go

@@ -242,7 +242,8 @@ func (h *redirectHandler) Handle(ctx context.Context, conn net.Conn, opts ...han
 
 	t := time.Now()
 	log.Infof("%s <-> %s", conn.RemoteAddr(), dstAddr)
-	xnet.Transport(conn, cc)
+	// xnet.Transport(conn, cc)
+	xnet.Pipe(ctx, conn, cc)
 	log.WithFields(map[string]any{
 		"duration": time.Since(t),
 	}).Infof("%s >-< %s", conn.RemoteAddr(), dstAddr)

handler/redirect/udp/handler.go

@@ -137,7 +137,8 @@ func (h *redirectHandler) Handle(ctx context.Context, conn net.Conn, opts ...han
 
 	t := time.Now()
 	log.Infof("%s <-> %s", conn.RemoteAddr(), dstAddr)
-	xnet.Transport(conn, cc)
+	// xnet.Transport(conn, cc)
+	xnet.Pipe(ctx, conn, cc)
 	log.WithFields(map[string]any{
 		"duration": time.Since(t),
 	}).Infof("%s >-< %s", conn.RemoteAddr(), dstAddr)

handler/relay/connect.go

@@ -202,7 +202,8 @@ func (h *relayHandler) handleConnect(ctx context.Context, conn net.Conn, network
 
 	t := time.Now()
 	log.Infof("%s <-> %s", conn.RemoteAddr(), address)
-	xnet.Transport(conn, cc)
+	// xnet.Transport(conn, cc)
+	xnet.Pipe(ctx, conn, cc)
 	log.WithFields(map[string]any{
 		"duration": time.Since(t),
 	}).Infof("%s >-< %s", conn.RemoteAddr(), address)

handler/relay/entrypoint.go

@@ -113,7 +113,8 @@ func (h *tcpHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler.
 
 	t := time.Now()
 	log.Debugf("%s <-> %s", conn.RemoteAddr(), cc.RemoteAddr())
-	xnet.Transport(conn, cc)
+	// xnet.Transport(conn, cc)
+	xnet.Pipe(ctx, conn, cc)
 	log.WithFields(map[string]any{"duration": time.Since(t)}).
 		Debugf("%s >-< %s", conn.RemoteAddr(), cc.RemoteAddr())
 	return nil

handler/relay/forward.go

@@ -122,7 +122,8 @@ func (h *relayHandler) handleForward(ctx context.Context, conn net.Conn, network
 
 	t := time.Now()
 	log.Debugf("%s <-> %s", conn.RemoteAddr(), target.Addr)
-	xnet.Transport(conn, cc)
+	// xnet.Transport(conn, cc)
+	xnet.Pipe(ctx, conn, cc)
 	log.WithFields(map[string]any{
 		"duration": time.Since(t),
 	}).Debugf("%s >-< %s", conn.RemoteAddr(), target.Addr)

handler/serial/handler.go

@@ -101,7 +101,8 @@ func (h *serialHandler) Handle(ctx context.Context, conn net.Conn, opts ...handl
 
 	t := time.Now()
 	log.Infof("%s <-> %s", conn.LocalAddr(), "@")
-	xnet.Transport(conn, cc)
+	// xnet.Transport(conn, cc)
+	xnet.Pipe(ctx, conn, cc)
 	log.WithFields(map[string]any{
 		"duration": time.Since(t),
 	}).Infof("%s >-< %s", conn.LocalAddr(), "@")
@@ -130,7 +131,8 @@ func (h *serialHandler) forwardSerial(ctx context.Context, conn net.Conn, target
 
 	t := time.Now()
 	log.Infof("%s <-> %s", conn.LocalAddr(), target.Addr)
-	xnet.Transport(conn, port)
+	// xnet.Transport(conn, port)
+	xnet.Pipe(ctx, conn, port)
 	log.WithFields(map[string]any{
 		"duration": time.Since(t),
 	}).Infof("%s >-< %s", conn.LocalAddr(), target.Addr)

handler/sni/handler.go

@@ -134,9 +134,7 @@ func (h *sniHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler.
 		return cc, err
 	}
 	dialTLS := func(ctx context.Context, network, address string, cfg *tls.Config) (net.Conn, error) {
-		var buf bytes.Buffer
+		cc, err := dial(ctx, network, address)
-		cc, err := h.options.Router.Dial(ctxvalue.ContextWithBuffer(ctx, &buf), "tcp", address)
-		ro.Route = buf.String()
 		if err != nil {
 			return nil, err
 		}

handler/socks/v4/handler.go

@@ -325,7 +325,8 @@ func (h *socks4Handler) handleConnect(ctx context.Context, conn net.Conn, req *g
 
 	t := time.Now()
 	log.Infof("%s <-> %s", conn.RemoteAddr(), ro.Host)
-	xnet.Transport(conn, cc)
+	// xnet.Transport(conn, cc)
+	xnet.Pipe(ctx, conn, cc)
 	log.WithFields(map[string]any{
 		"duration": time.Since(t),
 	}).Infof("%s >-< %s", conn.RemoteAddr(), ro.Host)

handler/socks/v5/bind.go

@@ -132,7 +132,8 @@ func (h *socks5Handler) serveBind(ctx context.Context, conn net.Conn, ln net.Lis
 			defer close(errc)
 			defer pc1.Close()
 
-			errc <- xnet.Transport(conn, pc1)
+			// errc <- xnet.Transport(conn, pc1)
+			errc <- xnet.Pipe(ctx, conn, pc1)
 		}()
 
 		return errc
@@ -172,7 +173,8 @@ func (h *socks5Handler) serveBind(ctx context.Context, conn net.Conn, ln net.Lis
 
 		start := time.Now()
 		log.Debugf("%s <-> %s", rc.LocalAddr(), rc.RemoteAddr())
-		xnet.Transport(pc2, rc)
+		// xnet.Transport(pc2, rc)
+		xnet.Pipe(ctx, pc2, rc)
 		log.WithFields(map[string]any{"duration": time.Since(start)}).
 			Debugf("%s >-< %s", rc.LocalAddr(), rc.RemoteAddr())
 

handler/socks/v5/connect.go

@@ -140,7 +140,8 @@ func (h *socks5Handler) handleConnect(ctx context.Context, conn net.Conn, networ
 
 	t := time.Now()
 	log.Infof("%s <-> %s", conn.RemoteAddr(), address)
-	xnet.Transport(conn, cc)
+	// xnet.Transport(conn, cc)
+	xnet.Pipe(ctx, conn, cc)
 	log.WithFields(map[string]any{
 		"duration": time.Since(t),
 	}).Infof("%s >-< %s", conn.RemoteAddr(), address)

handler/socks/v5/mbind.go

@@ -162,7 +162,8 @@ func (h *socks5Handler) serveMuxBind(ctx context.Context, conn net.Conn, ln net.
 
 			t := time.Now()
 			log.Debugf("%s <-> %s", c.LocalAddr(), c.RemoteAddr())
-			xnet.Transport(sc, c)
+			// xnet.Transport(sc, c)
+			xnet.Pipe(ctx, sc, c)
 			log.WithFields(map[string]any{"duration": time.Since(t)}).
 				Debugf("%s >-< %s", c.LocalAddr(), c.RemoteAddr())
 		}(rc)

handler/ss/handler.go

@@ -231,7 +231,8 @@ func (h *ssHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler.H
 
 	t := time.Now()
 	log.Infof("%s <-> %s", conn.RemoteAddr(), ro.Host)
-	xnet.Transport(conn, cc)
+	// xnet.Transport(conn, cc)
+	xnet.Pipe(ctx, conn, cc)
 	log.WithFields(map[string]any{
 		"duration": time.Since(t),
 	}).Infof("%s >-< %s", conn.RemoteAddr(), ro.Host)

handler/sshd/handler.go

@@ -219,7 +219,8 @@ func (h *forwardHandler) handleDirectForward(ctx context.Context, conn *sshd_uti
 
 	t := time.Now()
 	log.Infof("%s <-> %s", cc.LocalAddr(), targetAddr)
-	xnet.Transport(conn, cc)
+	// xnet.Transport(conn, cc)
+	xnet.Pipe(ctx, conn, cc)
 	log.WithFields(map[string]any{
 		"duration": time.Since(t),
 	}).Infof("%s >-< %s", cc.LocalAddr(), targetAddr)
@@ -321,7 +322,8 @@ func (h *forwardHandler) handleRemoteForward(ctx context.Context, conn *sshd_uti
 
 				t := time.Now()
 				log.Debugf("%s <-> %s", conn.LocalAddr(), conn.RemoteAddr())
-				xnet.Transport(ch, conn)
+				// xnet.Transport(ch, conn)
+				xnet.Pipe(ctx, ch, conn)
 				log.WithFields(map[string]any{
 					"duration": time.Since(t),
 				}).Debugf("%s >-< %s", conn.LocalAddr(), conn.RemoteAddr())

handler/tungo/handler.go

@@ -20,6 +20,7 @@ import (
 	"github.com/go-gost/x/registry"
 	"github.com/xjasonlyu/tun2socks/v2/core"
 	"github.com/xjasonlyu/tun2socks/v2/core/adapter"
+	"gvisor.dev/gvisor/pkg/tcpip/stack"
 )
 
 func init() {
@@ -33,6 +34,7 @@ type tungoHandler struct {
 	limiter  traffic.TrafficLimiter
 	cancel   context.CancelFunc
 	recorder recorder.RecorderObject
+	stack    *stack.Stack
 }
 
 func NewHandler(opts ...handler.Option) handler.Handler {
@@ -110,9 +112,10 @@ func (h *tungoHandler) Handle(ctx context.Context, conn net.Conn, opts ...handle
 		tcpQueue:   make(chan adapter.TCPConn),
 		udpQueue:   make(chan adapter.UDPConn),
 		udpTimeout: h.md.udpTimeout,
-		procCancel: func() { /* nop */ },
+		procCancel: func() {},
 
 		sniffing:                h.md.sniffing,
+		sniffingUDP:             h.md.sniffingUDP,
 		sniffingTimeout:         h.md.sniffingTimeout,
 		sniffingResponseTimeout: h.md.sniffingResponseTimeout,
 		sniffingFallback:        h.md.sniffingFallback,
@@ -133,9 +136,10 @@ func (h *tungoHandler) Handle(ctx context.Context, conn net.Conn, opts ...handle
 	if err != nil {
 		return err
 	}
-	defer stack.Close()
 
-	<-ctx.Done()
+	h.stack = stack
+
+	stack.Wait()
 
 	return nil
 }
@@ -145,6 +149,9 @@ func (h *tungoHandler) Close() error {
 	if h.cancel != nil {
 		h.cancel()
 	}
+	if h.stack != nil {
+		h.stack.Close()
+	}
 	return nil
 }
 

handler/tungo/metadata.go

@@ -11,6 +11,7 @@ type metadata struct {
 	udpTimeout time.Duration
 
 	sniffing                bool
+	sniffingUDP                bool
 	sniffingTimeout         time.Duration
 	sniffingResponseTimeout time.Duration
 	sniffingFallback        bool
@@ -26,6 +27,7 @@ func (h *tungoHandler) parseMetadata(md mdata.Metadata) (err error) {
 	h.md.udpTimeout = mdutil.GetDuration(md, "udpTimeout")
 
 	h.md.sniffing = mdutil.GetBool(md, "sniffing")
+	h.md.sniffingUDP = mdutil.GetBool(md, "sniffing.udp")
 	h.md.sniffingTimeout = mdutil.GetDuration(md, "sniffing.timeout")
 	h.md.sniffingResponseTimeout = mdutil.GetDuration(md, "sniffing.responseTimeout")
 	h.md.sniffingFallback = mdutil.GetBool(md, "sniffing.fallback")

handler/tungo/stack.go

@@ -23,6 +23,7 @@ import (
 	xstats "github.com/go-gost/x/observer/stats"
 	stats_wrapper "github.com/go-gost/x/observer/stats/wrapper"
 	xrecorder "github.com/go-gost/x/recorder"
+	"github.com/miekg/dns"
 	"github.com/rs/xid"
 	"github.com/xjasonlyu/tun2socks/v2/core/adapter"
 )
@@ -39,13 +40,14 @@ type transportHandler struct {
 	tcpQueue chan adapter.TCPConn
 	udpQueue chan adapter.UDPConn
 
-	// UDP session timeout.
-	udpTimeout time.Duration
-
 	procOnce   sync.Once
 	procCancel context.CancelFunc
 
+	// UDP session timeout.
+	udpTimeout time.Duration
+
 	sniffing                bool
+	sniffingUDP             bool
 	sniffingTimeout         time.Duration
 	sniffingResponseTimeout time.Duration
 	sniffingFallback        bool
@@ -112,6 +114,7 @@ func (h *transportHandler) handleTCPConn(originConn adapter.TCPConn) {
 		Network:    "tcp",
 		RemoteAddr: remoteAddr.String(),
 		Dst:        dstAddr.String(),
+		Host:       dstAddr.String(),
 		ClientIP:   remoteAddr.String(),
 		Time:       start,
 		SID:        sid,
@@ -143,6 +146,9 @@ func (h *transportHandler) handleTCPConn(originConn adapter.TCPConn) {
 		ro.InputBytes = pStats.Get(stats.KindInputBytes)
 		ro.OutputBytes = pStats.Get(stats.KindOutputBytes)
 		ro.Duration = time.Since(start)
+		if err := ro.Record(ctx, h.recorder.Recorder); err != nil {
+			log.Errorf("record: %v", err)
+		}
 
 		log.WithFields(map[string]any{
 			"src":         ro.Src,
@@ -174,7 +180,6 @@ func (h *transportHandler) handleTCPConn(originConn adapter.TCPConn) {
 		}
 
 		dial := func(ctx context.Context, network, address string) (net.Conn, error) {
-			log.Debugf("dial %s/%s", address, network)
 			var cc net.Conn
 			var err error
 			if address != "" {
@@ -257,7 +262,7 @@ func (h *transportHandler) handleTCPConn(originConn adapter.TCPConn) {
 
 	t := time.Now()
 	log.Infof("%s <-> %s", remoteAddr, dstAddr)
-	xnet.Transport(conn, cc)
+	xnet.Pipe(ctx, conn, cc)
 	log.WithFields(map[string]any{
 		"duration": time.Since(t),
 	}).Infof("%s >-< %s", remoteAddr, dstAddr)
@@ -284,9 +289,15 @@ func (h *transportHandler) handleUDPConn(uc adapter.UDPConn) {
 		Network:    "udp",
 		RemoteAddr: remoteAddr.String(),
 		Dst:        dstAddr.String(),
+		Host:       dstAddr.String(),
+		ClientIP:   remoteAddr.String(),
 		Time:       start,
 		SID:        sid,
 	}
+	if h, _, _ := net.SplitHostPort(ro.ClientIP); h != "" {
+		ro.ClientIP = h
+	}
+
 	log := h.opts.Logger.WithFields(map[string]any{
 		"network": ro.Network,
 		"remote":  ro.RemoteAddr,
@@ -309,6 +320,9 @@ func (h *transportHandler) handleUDPConn(uc adapter.UDPConn) {
 		ro.InputBytes = pStats.Get(stats.KindInputBytes)
 		ro.OutputBytes = pStats.Get(stats.KindOutputBytes)
 		ro.Duration = time.Since(start)
+		if err := ro.Record(ctx, h.recorder.Recorder); err != nil {
+			log.Errorf("record: %v", err)
+		}
 
 		log.WithFields(map[string]any{
 			"src":         ro.Src,
@@ -330,13 +344,13 @@ func (h *transportHandler) handleUDPConn(uc adapter.UDPConn) {
 
 	t := time.Now()
 	log.Infof("%s <-> %s", remoteAddr, dstAddr)
-	pipeData(conn, cc, h.udpTimeout)
+	pipePacketData(conn, cc, h.sniffingUDP, ro, h.udpTimeout)
 	log.WithFields(map[string]any{
 		"duration": time.Since(t),
 	}).Infof("%s >-< %s", remoteAddr, dstAddr)
 }
 
-func pipeData(conn1, conn2 net.Conn, timeout time.Duration) {
+func pipePacketData(conn1, conn2 net.Conn, sniffing bool, ro *xrecorder.HandlerRecorderObject, timeout time.Duration) {
 	if timeout <= 0 {
 		timeout = udpSessionTimeout
 	}
@@ -346,17 +360,18 @@ func pipeData(conn1, conn2 net.Conn, timeout time.Duration) {
 
 	go func() {
 		defer wg.Done()
-		copyData(conn1, conn2, timeout)
+		copyPacketData(conn1, conn2, sniffing, false, ro, timeout)
 	}()
 	go func() {
 		defer wg.Done()
-		copyData(conn2, conn1, timeout)
+		copyPacketData(conn2, conn1, sniffing, true, ro, timeout)
 	}()
 	wg.Wait()
 }
 
-func copyData(dst, src net.Conn, timeout time.Duration) error {
+func copyPacketData(dst, src net.Conn, sniffing bool, c2s bool, ro *xrecorder.HandlerRecorderObject, timeout time.Duration) error {
-	buf := make([]byte, math.MaxUint16)
+	buf := make([]byte, math.MaxUint16/2)
+	isDNS := false
 
 	for {
 		src.SetReadDeadline(time.Now().Add(timeout))
@@ -369,9 +384,41 @@ func copyData(dst, src net.Conn, timeout time.Duration) error {
 			return err
 		}
 
+		if n == 0 {
+			return nil
+		}
+
+		if sniffing {
+			// try to sniff DNS msg.
+			{
+				msg := &dns.Msg{}
+				if msg.Unpack(buf[:n]) == nil && len(msg.Question) > 0 {
+					if c2s {
+						ro.Proto = "dns"
+						ro.DNS = &xrecorder.DNSRecorderObject{
+							ID:       int(msg.Id),
+							Name:     msg.Question[0].Name,
+							Class:    dns.Class(msg.Question[0].Qclass).String(),
+							Type:     dns.Type(msg.Question[0].Qtype).String(),
+							Question: msg.String(),
+						}
+					} else {
+						if ro.DNS != nil {
+							ro.DNS.Answer = msg.String()
+						}
+					}
+					isDNS = true
+				}
+			}
+		}
+
 		if _, err = dst.Write(buf[:n]); err != nil {
 			return err
 		}
 		dst.SetReadDeadline(time.Now().Add(timeout))
+
+		if isDNS {
+			return nil
+		}
 	}
 }

handler/tunnel/connect.go

@@ -116,7 +116,8 @@ func (h *tunnelHandler) handleConnect(ctx context.Context, req *relay.Request, c
 
 	t := time.Now()
 	log.Debugf("%s <-> %s", conn.RemoteAddr(), cc.RemoteAddr())
-	xnet.Transport(conn, cc)
+	// xnet.Transport(conn, cc)
+	xnet.Pipe(ctx, conn, cc)
 	log.WithFields(map[string]any{
 		"duration": time.Since(t),
 	}).Debugf("%s >-< %s", conn.RemoteAddr(), cc.RemoteAddr())

handler/tunnel/entrypoint.go

@@ -435,7 +435,8 @@ func (ep *entrypoint) handleUpgradeResponse(ctx context.Context, rw io.ReadWrite
 		return ep.sniffingWebsocketFrame(ctx, rw, backConn, ro, log)
 	}
 
-	return xnet.Transport(rw, backConn)
+	// return xnet.Transport(rw, backConn)
+	return xnet.Pipe(ctx, rw, backConn)
 }
 
 func (ep *entrypoint) sniffingWebsocketFrame(ctx context.Context, rw, cc io.ReadWriter, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error {
@@ -616,7 +617,8 @@ func (ep *entrypoint) HandleTLS(ctx context.Context, conn net.Conn, ro *xrecorde
 		return err
 	}
 
-	xnet.Transport(conn, cc)
+	// xnet.Transport(conn, cc)
+	xnet.Pipe(ctx, conn, cc)
 	return err
 }
 
@@ -698,7 +700,8 @@ func (ep *entrypoint) handleConnect(ctx context.Context, conn net.Conn, ro *xrec
 
 	t := time.Now()
 	log.Debugf("%s <-> %s", conn.RemoteAddr(), cc.RemoteAddr())
-	xnet.Transport(conn, cc)
+	// xnet.Transport(conn, cc)
+	xnet.Pipe(ctx, conn, cc)
 	log.WithFields(map[string]any{
 		"duration": time.Since(t),
 	}).Debugf("%s >-< %s", conn.RemoteAddr(), cc.RemoteAddr())

handler/unix/handler.go

@@ -194,7 +194,8 @@ func (h *unixHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler
 
 	t := time.Now()
 	log.Infof("%s <-> %s", conn.LocalAddr(), "@")
-	xnet.Transport(conn, cc)
+	// xnet.Transport(conn, cc)
+	xnet.Pipe(ctx, conn, cc)
 	log.WithFields(map[string]any{
 		"duration": time.Since(t),
 	}).Infof("%s >-< %s", conn.LocalAddr(), "@")
@@ -245,7 +246,8 @@ func (h *unixHandler) forwardUnix(ctx context.Context, conn net.Conn, target *ch
 
 	t := time.Now()
 	log.Infof("%s <-> %s", conn.LocalAddr(), target.Addr)
-	xnet.Transport(rw, cc)
+	// xnet.Transport(rw, cc)
+	xnet.Pipe(ctx, rw, cc)
 	log.WithFields(map[string]any{
 		"duration": time.Since(t),
 	}).Infof("%s >-< %s", conn.LocalAddr(), target.Addr)

handler/unix/sniffing.go

@@ -18,7 +18,7 @@ import (
 	dissector "github.com/go-gost/tls-dissector"
 	xbypass "github.com/go-gost/x/bypass"
 	xio "github.com/go-gost/x/internal/io"
-	netpkg "github.com/go-gost/x/internal/net"
+	xnet "github.com/go-gost/x/internal/net"
 	xhttp "github.com/go-gost/x/internal/net/http"
 	tls_util "github.com/go-gost/x/internal/util/tls"
 	xrecorder "github.com/go-gost/x/recorder"
@@ -199,13 +199,14 @@ func (h *unixHandler) httpRoundTrip(ctx context.Context, rw, cc io.ReadWriter, r
 	}
 
 	if req.Header.Get("Upgrade") == "websocket" {
-		netpkg.Transport(rw, cc)
+		// xnet.Transport(rw, cc)
+		xnet.Pipe(ctx, rw, cc)
 	}
 
 	return resp.Close, nil
 }
 
-func (h *unixHandler) handleTLS(_ context.Context, rw, cc io.ReadWriter, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error {
+func (h *unixHandler) handleTLS(ctx context.Context, rw, cc io.ReadWriter, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error {
 	buf := new(bytes.Buffer)
 
 	clientHello, err := dissector.ParseClientHello(io.TeeReader(rw, buf))
@@ -250,7 +251,8 @@ func (h *unixHandler) handleTLS(_ context.Context, rw, cc io.ReadWriter, ro *xre
 
 	t := time.Now()
 	log.Infof("%s <-> %s", ro.RemoteAddr, ro.Host)
-	netpkg.Transport(rw, cc)
+	// xnet.Transport(rw, cc)
+	xnet.Pipe(ctx, rw, cc)
 	log.WithFields(map[string]any{
 		"duration": time.Since(t),
 	}).Infof("%s >-< %s", ro.RemoteAddr, ro.Host)

internal/io/io.go

@@ -5,6 +5,14 @@ import (
 	"time"
 )
 
+type CloseRead interface {
+	CloseRead() error
+}
+
+type CloseWrite interface {
+	CloseWrite() error
+}
+
 type readWriter struct {
 	io.Reader
 	io.Writer
@@ -17,6 +25,20 @@ func NewReadWriter(r io.Reader, w io.Writer) io.ReadWriter {
 	}
 }
 
+func (rw *readWriter) CloseRead() error {
+	if sc, ok := rw.Writer.(CloseRead); ok {
+		return sc.CloseRead()
+	}
+	return nil
+}
+
+func (rw *readWriter) CloseWrite() error {
+	if sc, ok := rw.Writer.(CloseWrite); ok {
+		return sc.CloseWrite()
+	}
+	return nil
+}
+
 type readWriteCloser struct {
 	io.Reader
 	io.Writer

internal/net/addr.go

@@ -193,7 +193,3 @@ func (r *IPRange) Parse(s string) error {
 func (r *IPRange) Contains(addr netip.Addr) bool {
 	return !(addr.Less(r.Min) || r.Max.Less(addr))
 }
-
-type ClientAddr interface {
-	ClientAddr() net.Addr
-}

internal/net/dialer/dialer.go

@@ -135,7 +135,7 @@ func (d *Dialer) dialOnce(ctx context.Context, network, addr, ifceName string, i
 			}
 			err = sc.Control(func(fd uintptr) {
 				if ifceName != "" {
-					if err := bindDevice(network, fd, ifceName); err != nil {
+					if err := bindDevice(network, addr, fd, ifceName); err != nil {
 						log.Warnf("bind device: %v", err)
 					}
 				}
@@ -159,13 +159,13 @@ func (d *Dialer) dialOnce(ctx context.Context, network, addr, ifceName string, i
 		Control: func(network, address string, c syscall.RawConn) error {
 			return c.Control(func(fd uintptr) {
 				if ifceName != "" {
-					if err := bindDevice(network, fd, ifceName); err != nil {
+					if err := bindDevice(network, address, fd, ifceName); err != nil {
-						log.Warnf("bind device: %v", err)
+						log.Warnf("%s/%s bind device: %v", address, network, err)
 					}
 				}
 				if d.Mark != 0 {
 					if err := setMark(fd, d.Mark); err != nil {
-						log.Warnf("set mark: %v", err)
+						log.Warnf("%s/%s set mark: %v", address, network, err)
 					}
 				}
 			})

internal/net/dialer/dialer_darwin.go

@@ -7,18 +7,23 @@ import (
 	"golang.org/x/sys/unix"
 )
 
-func bindDevice(network string, fd uintptr, ifceName string) error {
+func bindDevice(network, address string, fd uintptr, ifceName string) error {
 	if ifceName == "" {
 		return nil
 	}
 
+	host, _, _ := net.SplitHostPort(address)
+	if ip := net.ParseIP(host); ip != nil && !ip.IsGlobalUnicast() {
+		return nil
+	}
+
 	ifce, err := net.InterfaceByName(ifceName)
-	if err == nil {
+	if err != nil {
 		return err
 	}
 
 	switch network {
-	case "tcp", "tcp4", "udp4":
+	case "tcp", "tcp4", "udp", "udp4":
 		return unix.SetsockoptInt(int(fd), syscall.IPPROTO_IP, syscall.IP_BOUND_IF, ifce.Index)
 	case "tcp6", "udp6":
 		return unix.SetsockoptInt(int(fd), syscall.IPPROTO_IPV6, syscall.IPV6_BOUND_IF, ifce.Index)

internal/net/dialer/dialer_freebsd.go

@@ -0,0 +1,16 @@
+package dialer
+
+import (
+	"golang.org/x/sys/unix"
+)
+
+func bindDevice(network, address string, fd uintptr, ifceName string) error {
+	return nil
+}
+
+func setMark(fd uintptr, mark int) error {
+	if mark != 0 {
+		return nil
+	}
+	return unix.SetsockoptInt(int(fd), unix.SOL_SOCKET, unix.SO_USER_COOKIE, mark)
+}

internal/net/dialer/dialer_linux.go

@@ -1,13 +1,21 @@
 package dialer
 
 import (
+	"net"
+
 	"golang.org/x/sys/unix"
 )
 
-func bindDevice(network string, fd uintptr, ifceName string) error {
+func bindDevice(network, address string, fd uintptr, ifceName string) error {
 	if ifceName == "" {
 		return nil
 	}
+
+	host, _, _ := net.SplitHostPort(address)
+	if ip := net.ParseIP(host); ip != nil && !ip.IsGlobalUnicast() {
+		return nil
+	}
+
 	return unix.BindToDevice(int(fd), ifceName)
 }
 

internal/net/dialer/dialer_openbsd.go

@@ -0,0 +1,16 @@
+package dialer
+
+import (
+	"golang.org/x/sys/unix"
+)
+
+func bindDevice(network, address string, fd uintptr, ifceName string) error {
+	return nil
+}
+
+func setMark(fd uintptr, mark int) error {
+	if mark != 0 {
+		return nil
+	}
+	return unix.SetsockoptInt(int(fd), unix.SOL_SOCKET, unix.SO_RTABLE, mark)
+}

internal/net/dialer/dialer_other.go

@@ -1,8 +1,8 @@
-//go:build !linux && !windows && !darwin
+//go:build !unix && !windows
 
 package dialer
 
-func bindDevice(network string, fd uintptr, ifceName string) error {
+func bindDevice(network, address string, fd uintptr, ifceName string) error {
 	return nil
 }
 

internal/net/dialer/dialer_windows.go

@@ -13,21 +13,33 @@ const (
 	IPV6_UNICAST_IF = 31
 )
 
-func bindDevice(network string, fd uintptr, ifceName string) error {
+func bindDevice(network, address string, fd uintptr, ifceName string) error {
 	if ifceName == "" {
 		return nil
 	}
 
+	host, _, _ := net.SplitHostPort(address)
+	ip := net.ParseIP(host)
+	if ip != nil && !ip.IsGlobalUnicast() {
+		return nil
+	}
+
 	ifce, err := net.InterfaceByName(ifceName)
-	if err == nil {
+	if err != nil {
 		return err
 	}
 
 	switch network {
-	case "tcp", "tcp4", "udp4":
+	case "tcp", "tcp4", "udp", "udp4":
 		return bindSocketToInterface4(windows.Handle(fd), uint32(ifce.Index))
 	case "tcp6", "udp6":
-		return bindSocketToInterface6(windows.Handle(fd), uint32(ifce.Index))
+		err = bindSocketToInterface6(windows.Handle(fd), uint32(ifce.Index))
+		if network == "udp6" && ip == nil {
+			// The underlying IP net maybe IPv4 even if the `network` param is `udp6`,
+			// so we should bind socket to interface4 at the same time.
+			err = bindSocketToInterface4(windows.Handle(fd), uint32(ifce.Index))
+		}
+		return err
 	}
 
 	return nil

internal/net/net.go

@@ -9,6 +9,7 @@ import (
 	"strings"
 	"syscall"
 
+	xio "github.com/go-gost/x/internal/io"
 	"github.com/vishvananda/netns"
 )
 
@@ -25,6 +26,10 @@ type RemoteAddr interface {
 	RemoteAddr() net.Addr
 }
 
+type ClientAddr interface {
+	ClientAddr() net.Addr
+}
+
 // tcpraw.TCPConn
 type SetDSCP interface {
 	SetDSCP(int) error
@@ -119,3 +124,17 @@ func (c *readWriteConn) Read(p []byte) (int, error) {
 func (c *readWriteConn) Write(p []byte) (int, error) {
 	return c.w.Write(p)
 }
+
+func (c *readWriteConn) CloseRead() error {
+	if sc, ok := c.Conn.(xio.CloseRead); ok {
+		return sc.CloseRead()
+	}
+	return nil
+}
+
+func (c *readWriteConn) CloseWrite() error {
+	if sc, ok := c.Conn.(xio.CloseWrite); ok {
+		return sc.CloseWrite()
+	}
+	return nil
+}

internal/net/pipe.go

@@ -0,0 +1,76 @@
+package net
+
+import (
+	"context"
+	"io"
+	"sync"
+	"time"
+
+	"github.com/go-gost/core/common/bufpool"
+	xio "github.com/go-gost/x/internal/io"
+)
+
+const (
+	// tcpWaitTimeout implements a TCP half-close timeout.
+	tcpWaitTimeout = 30 * time.Second
+)
+
+func Pipe(ctx context.Context, rw1, rw2 io.ReadWriter) error {
+	wg := sync.WaitGroup{}
+	wg.Add(2)
+
+	ch := make(chan error, 2)
+
+	go func() {
+		defer wg.Done()
+		if err := pipeBuffer(rw1, rw2, bufferSize/2); err != nil {
+			ch <- err
+		}
+	}()
+	go func() {
+		defer wg.Done()
+		if err := pipeBuffer(rw2, rw1, bufferSize/2); err != nil {
+			ch <- err
+		}
+	}()
+
+	done := make(chan struct{})
+	go func() {
+		wg.Wait()
+		close(done)
+	}()
+
+	select {
+	case <-done:
+	case <-ctx.Done():
+		return nil
+	}
+
+	select {
+	case err := <-ch:
+		return err
+	default:
+	}
+
+	return nil
+}
+
+func pipeBuffer(dst io.ReadWriter, src io.ReadWriter, bufferSize int) error {
+	buf := bufpool.Get(bufferSize)
+	defer bufpool.Put(buf)
+
+	_, err := io.CopyBuffer(dst, src, buf)
+
+	// Do the upload/download side TCP half-close.
+	if cr, ok := src.(xio.CloseRead); ok {
+		cr.CloseRead()
+	}
+	if cw, ok := dst.(xio.CloseWrite); ok {
+		cw.CloseWrite()
+	}
+
+	// Set TCP half-close timeout.
+	xio.SetReadDeadline(dst, time.Now().Add(tcpWaitTimeout))
+
+	return err
+}

internal/util/forwarder/sniffer.go

@@ -602,7 +602,8 @@ func (h *Sniffer) handleUpgradeResponse(ctx context.Context, rw io.ReadWriter, c
 		return h.sniffingWebsocketFrame(ctx, rw, cc, ro, log)
 	}
 
-	return xnet.Transport(rw, cc)
+	// return xnet.Transport(rw, cc)
+	return xnet.Pipe(ctx, rw, cc)
 }
 
 func (h *Sniffer) sniffingWebsocketFrame(ctx context.Context, rw, cc io.ReadWriter, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error {
@@ -861,7 +862,8 @@ func (h *Sniffer) HandleTLS(ctx context.Context, conn net.Conn, opts ...HandleOp
 	}
 
 	log.Infof("%s <-> %s", ro.RemoteAddr, ro.Host)
-	xnet.Transport(conn, cc)
+	// xnet.Transport(conn, cc)
+	xnet.Pipe(ctx, conn, cc)
 	log.WithFields(map[string]any{
 		"duration": time.Since(ro.Time),
 	}).Infof("%s >-< %s", ro.RemoteAddr, ro.Host)

internal/util/sniffing/sniffer.go

@@ -449,7 +449,8 @@ func (h *Sniffer) handleUpgradeResponse(ctx context.Context, rw io.ReadWriter, c
 		return h.sniffingWebsocketFrame(ctx, rw, cc, ro, log)
 	}
 
-	return xnet.Transport(rw, cc)
+	// return xnet.Transport(rw, cc)
+	return xnet.Pipe(ctx, rw, cc)
 }
 
 func (h *Sniffer) sniffingWebsocketFrame(ctx context.Context, rw, cc io.ReadWriter, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error {
@@ -663,7 +664,8 @@ func (h *Sniffer) HandleTLS(ctx context.Context, conn net.Conn, opts ...HandleOp
 	}
 
 	log.Infof("%s <-> %s", ro.RemoteAddr, ro.Host)
-	xnet.Transport(conn, cc)
+	// xnet.Transport(conn, cc)
+	xnet.Pipe(ctx, conn, cc)
 	log.WithFields(map[string]any{
 		"duration": time.Since(ro.Time),
 	}).Infof("%s >-< %s", ro.RemoteAddr, ro.Host)

internal/util/ss/ss.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"net"
 
+	xio "github.com/go-gost/x/internal/io"
 	"github.com/shadowsocks/go-shadowsocks2/core"
 )
 
@@ -40,3 +41,17 @@ func (c *shadowConn) Write(b []byte) (n int, err error) {
 	_, err = c.Conn.Write(b)
 	return
 }
+
+func (c *shadowConn) CloseRead() error {
+	if sc, ok := c.Conn.(xio.CloseRead); ok {
+		return sc.CloseRead()
+	}
+	return nil
+}
+
+func (c *shadowConn) CloseWrite() error {
+	if sc, ok := c.Conn.(xio.CloseWrite); ok {
+		return sc.CloseWrite()
+	}
+	return nil
+}

limiter/conn/wrapper/conn.go

@@ -7,6 +7,7 @@ import (
 
 	limiter "github.com/go-gost/core/limiter/conn"
 	"github.com/go-gost/core/metadata"
+	xio "github.com/go-gost/x/internal/io"
 )
 
 var (
@@ -49,3 +50,17 @@ func (c *serverConn) Metadata() metadata.Metadata {
 	}
 	return nil
 }
+
+func (c *serverConn) CloseRead() error {
+	if sc, ok := c.Conn.(xio.CloseRead); ok {
+		return sc.CloseRead()
+	}
+	return nil
+}
+
+func (c *serverConn) CloseWrite() error {
+	if sc, ok := c.Conn.(xio.CloseWrite); ok {
+		return sc.CloseWrite()
+	}
+	return nil
+}

limiter/traffic/wrapper/conn.go

@@ -13,6 +13,7 @@ import (
 	"github.com/go-gost/core/metadata"
 	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 (
@@ -113,6 +114,20 @@ func (c *limitConn) ClientAddr() net.Addr {
 	return nil
 }
 
+func (c *limitConn) CloseRead() error {
+	if sc, ok := c.Conn.(xio.CloseRead); ok {
+		return sc.CloseRead()
+	}
+	return nil
+}
+
+func (c *limitConn) CloseWrite() error {
+	if sc, ok := c.Conn.(xio.CloseWrite); ok {
+		return sc.CloseWrite()
+	}
+	return nil
+}
+
 type packetConn struct {
 	net.PacketConn
 	limiter traffic.TrafficLimiter

listener/tun/listener.go

@@ -90,7 +90,7 @@ func (l *tunListener) listenLoop(ready context.CancelCauseFunc) {
 
 			addrs, _ := itf.Addrs()
 			l.log.Infof("name: %s, net: %s, mtu: %d, addrs: %s",
-				itf.Name, ip, itf.MTU, addrs)
+				itf.Name, ip, l.md.config.MTU, addrs)
 
 			var c net.Conn
 			c = &conn{

listener/tun/tun.go

@@ -16,6 +16,7 @@ type tunDevice struct {
 	sizes   []int
 	rbufs   [][]byte
 	wbufs   [][]byte
+	rbuf    []byte
 	wbuf    []byte
 }
 
@@ -32,13 +33,22 @@ func (d *tunDevice) Read(p []byte) (n int, err error) {
 		}
 	}
 
+	if readOffset > 0 {
+		d.rbufs[0] = d.rbuf
+	} else {
 		d.rbufs[0] = p
+	}
+
 	packets, err := d.dev.Read(d.rbufs, d.sizes, readOffset)
 	if err != nil && err != tun.ErrTooManySegments {
 		return
 	}
 	n = d.sizes[0]
 
+	if readOffset > 0 {
+		copy(p, d.rbuf[readOffset:n+readOffset])
+	}
+
 	d.sizes[0] = 0
 	d.packets = packets - 1
 
@@ -80,6 +90,7 @@ func (l *tunListener) createTunDevice() (dev io.ReadWriteCloser, name string, er
 		sizes: make([]int, batchSize),
 		rbufs: rbufs,
 		wbufs: make([][]byte, 1),
+		rbuf:  make([]byte, maxBufSize+readOffset),
 		wbuf:  make([]byte, maxBufSize+writeOffset),
 	}
 	name, err = ifce.Name()

listener/tun/tun_darwin.go

@@ -23,13 +23,13 @@ func (l *tunListener) createTun() (ifce io.ReadWriteCloser, name string, ip net.
 		return
 	}
 
+	if len(l.md.config.Net) > 0 {
 		peer := l.md.config.Peer
 		if peer == "" {
-		peer = ip.String()
+			peer = l.md.config.Net[0].IP.String()
 		}
-	if len(l.md.config.Net) > 0 {
 		cmd := fmt.Sprintf("ifconfig %s inet %s %s mtu %d up",
-			name, l.md.config.Net[0].String(), l.md.config.Peer, l.md.config.MTU)
+			name, l.md.config.Net[0].String(), peer, l.md.config.MTU)
 		l.log.Debug(cmd)
 		args := strings.Split(cmd, " ")
 		if err = exec.Command(args[0], args[1:]...).Run(); err != nil {

listener/tungo/listener.go

@@ -83,7 +83,7 @@ func (l *tunListener) listenLoop(ready context.CancelCauseFunc) {
 
 			addrs, _ := itf.Addrs()
 			l.log.Infof("name: %s, net: %s, mtu: %d, addrs: %s",
-				itf.Name, ip, itf.MTU, addrs)
+				itf.Name, ip, l.md.config.MTU, addrs)
 
 			var c net.Conn
 			c = &conn{

listener/tungo/metadata.go

@@ -30,7 +30,7 @@ func (l *tunListener) parseMetadata(md mdata.Metadata) (err error) {
 		config.MTU = defaultMTU
 	}
 	if config.Name == "" {
-		config.Name = "tungo"
+		config.Name = defaultName
 	}
 	if gw := mdutil.GetString(md, "gw", "tun.gw"); gw != "" {
 		config.Gateway = net.ParseIP(gw)

listener/tungo/tun.go

@@ -3,7 +3,6 @@ package tungo
 import (
 	"io"
 
-	"github.com/go-gost/core/logger"
 	"golang.zx2c4.com/wireguard/tun"
 )
 
@@ -13,14 +12,12 @@ const (
 
 type tunDevice struct {
 	dev     tun.Device
-
 	packets int
 	sizes   []int
 	rbufs   [][]byte
 	wbufs   [][]byte
+	rbuf    []byte
 	wbuf    []byte
-
-	log logger.Logger
 }
 
 func (d *tunDevice) Read(p []byte) (n int, err error) {
@@ -36,17 +33,25 @@ func (d *tunDevice) Read(p []byte) (n int, err error) {
 		}
 	}
 
+	if readOffset > 0 {
+		d.rbufs[0] = d.rbuf
+	} else {
 		d.rbufs[0] = p
+	}
+
 	packets, err := d.dev.Read(d.rbufs, d.sizes, readOffset)
 	if err != nil && err != tun.ErrTooManySegments {
 		return
 	}
 	n = d.sizes[0]
 
+	if readOffset > 0 {
+		copy(p, d.rbuf[readOffset:n+readOffset])
+	}
+
 	d.sizes[0] = 0
 	d.packets = packets - 1
 
-	// d.log.Debugf("read tun: (%d), % x", n, p[:n])
 	return
 }
 
@@ -85,8 +90,8 @@ func (l *tunListener) createTunDevice() (dev io.ReadWriteCloser, name string, er
 		sizes: make([]int, batchSize),
 		rbufs: rbufs,
 		wbufs: make([][]byte, 1),
+		rbuf:  make([]byte, maxBufSize+readOffset),
 		wbuf:  make([]byte, maxBufSize+writeOffset),
-		log:   l.log,
 	}
 	name, err = ifce.Name()
 

listener/tungo/tun_darwin.go

@@ -23,13 +23,13 @@ func (l *tunListener) createTun() (ifce io.ReadWriteCloser, name string, ip net.
 		return
 	}
 
+	if len(l.md.config.Net) > 0 {
 		peer := l.md.config.Peer
 		if peer == "" {
-		peer = ip.String()
+			peer = l.md.config.Net[0].IP.String()
 		}
-	if len(l.md.config.Net) > 0 {
 		cmd := fmt.Sprintf("ifconfig %s inet %s %s mtu %d up",
-			name, l.md.config.Net[0].String(), l.md.config.Peer, l.md.config.MTU)
+			name, l.md.config.Net[0].String(), peer, l.md.config.MTU)
 		l.log.Debug(cmd)
 		args := strings.Split(cmd, " ")
 		if err = exec.Command(args[0], args[1:]...).Run(); err != nil {

listener/tungo/tun_windows.go

@@ -26,7 +26,7 @@ func (l *tunListener) createTun() (ifce io.ReadWriteCloser, name string, ip net.
 	if len(l.md.config.Net) > 0 {
 		ipNet := l.md.config.Net[0]
 		cmd := fmt.Sprintf("netsh interface ip set address name=%s "+
-			"source=static addr=%s mask=%s gateway=none",
+			"source=static addr=%s mask=%s",
 			name, ipNet.IP.String(), ipMask(ipNet.Mask))
 		l.log.Debug(cmd)
 

metrics/wrapper/conn.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/go-gost/core/metadata"
 	"github.com/go-gost/core/metrics"
+	xio "github.com/go-gost/x/internal/io"
 	xnet "github.com/go-gost/x/internal/net"
 	"github.com/go-gost/x/internal/net/udp"
 	xmetrics "github.com/go-gost/x/metrics"
@@ -87,6 +88,20 @@ func (c *serverConn) ClientAddr() net.Addr {
 	return nil
 }
 
+func (c *serverConn) CloseRead() error {
+	if sc, ok := c.Conn.(xio.CloseRead); ok {
+		return sc.CloseRead()
+	}
+	return nil
+}
+
+func (c *serverConn) CloseWrite() error {
+	if sc, ok := c.Conn.(xio.CloseWrite); ok {
+		return sc.CloseWrite()
+	}
+	return nil
+}
+
 type packetConn struct {
 	net.PacketConn
 	service string

observer/stats/wrapper/conn.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/go-gost/core/metadata"
 	"github.com/go-gost/core/observer/stats"
+	xio "github.com/go-gost/x/internal/io"
 	xnet "github.com/go-gost/x/internal/net"
 	"github.com/go-gost/x/internal/net/udp"
 )
@@ -89,6 +90,20 @@ func (c *conn) ClientAddr() net.Addr {
 	return nil
 }
 
+func (c *conn) CloseRead() error {
+	if sc, ok := c.Conn.(xio.CloseRead); ok {
+		return sc.CloseRead()
+	}
+	return nil
+}
+
+func (c *conn) CloseWrite() error {
+	if sc, ok := c.Conn.(xio.CloseWrite); ok {
+		return sc.CloseWrite()
+	}
+	return nil
+}
+
 type packetConn struct {
 	net.PacketConn
 	stats stats.Stats