refactor: унифицировать логирование в internal/proxy/* через logx.Logger

Заменяет прямой log.Printf на logger.Errorf/Debugf в каждом proxy-пакете
(udprelay, tcpfwd, bondclient, bondserver, udpserver, tcpfwdserver). Вывод
остаётся совместимым (под капотом stdlib log), но теперь может быть глобально
перенаправлен сменой реализации Logger.

Тесты + lint проходят.
This commit is contained in:
samosvalishe
2026-05-19 21:57:33 +03:00
parent 51bda46446
commit aff95e87d3
6 changed files with 74 additions and 78 deletions
+8 -9
View File
@@ -8,7 +8,6 @@ import (
"context"
"fmt"
"io"
"log"
"net"
"strconv"
"strings"
@@ -64,11 +63,11 @@ func (h *Handler) Handle(ctx context.Context, tcpConn net.Conn, connID uint64, c
}
stream, err := ps.Sess.OpenStream()
if err != nil {
log.Printf("[bond %d] session %d open stream error: %s", connID, ps.ID, err)
h.Deps.log().Errorf("[bond %d] session %d open stream error: %s", connID, ps.ID, err)
continue
}
if err := bondframe.WriteHello(stream, connID, uint16(i), uint16(len(candidates))); err != nil {
log.Printf("[bond %d] session %d hello error: %s", connID, ps.ID, err)
h.Deps.log().Errorf("[bond %d] session %d hello error: %s", connID, ps.ID, err)
_ = stream.Close()
continue
}
@@ -79,7 +78,7 @@ func (h *Handler) Handle(ctx context.Context, tcpConn net.Conn, connID uint64, c
}
if len(lanes) == 0 {
log.Printf("[bond %d] no usable lanes, rejecting TCP from %s", connID, tcpConn.RemoteAddr())
h.Deps.log().Errorf("[bond %d] no usable lanes, rejecting TCP from %s", connID, tcpConn.RemoteAddr())
return
}
context.AfterFunc(ctx, func() {
@@ -159,7 +158,7 @@ func (h *Handler) copyTCPToBond(ctx context.Context, connID uint64, tcpConn net.
if n > 0 {
l, writeErr := writeBondFrameToNextLane(ctx, lanes, bondframe.FrameData, seq, buf[:n], &laneIdx)
if writeErr != nil {
log.Printf("[bond %d] write data error: %v", connID, writeErr)
h.Deps.log().Errorf("[bond %d] write data error: %v", connID, writeErr)
return
}
l.ps.ToSession.Add(uint64(n))
@@ -177,7 +176,7 @@ func (h *Handler) copyTCPToBond(ctx context.Context, connID uint64, tcpConn net.
writeErr := bondframe.WriteFrame(l.stream, bondframe.FrameFIN, seq, nil)
l.mu.Unlock()
if writeErr != nil && ctx.Err() == nil {
log.Printf("[bond %d] session %d write FIN error: %v", connID, l.ps.ID, writeErr)
h.Deps.log().Errorf("[bond %d] session %d write FIN error: %v", connID, l.ps.ID, writeErr)
}
}
h.Deps.log().Debugf("[bond %d] upload finished chunks=%d", connID, seq)
@@ -238,7 +237,7 @@ func (h *Handler) copyBondToTCP(ctx context.Context, connID uint64, tcpConn net.
switch f.Type {
case bondframe.FrameData:
if len(pending) >= 1024 {
log.Printf("[bond %d] pending map overflow (>1024), closing", connID)
h.Deps.log().Errorf("[bond %d] pending map overflow (>1024), closing", connID)
return
}
pending[f.Seq] = f.Data
@@ -248,7 +247,7 @@ func (h *Handler) copyBondToTCP(ctx context.Context, connID uint64, tcpConn net.
finSeq = &v
}
default:
log.Printf("[bond %d] unknown frame type %d", connID, f.Type)
h.Deps.log().Errorf("[bond %d] unknown frame type %d", connID, f.Type)
return
}
@@ -260,7 +259,7 @@ func (h *Handler) copyBondToTCP(ctx context.Context, connID uint64, tcpConn net.
delete(pending, expect)
if len(data) > 0 {
if _, err := tcpConn.Write(data); err != nil {
log.Printf("[bond %d] local TCP write error: %v", connID, err)
h.Deps.log().Errorf("[bond %d] local TCP write error: %v", connID, err)
return
}
}
+12 -13
View File
@@ -7,7 +7,6 @@ package bondserver
import (
"context"
"io"
"log"
"net"
"sync"
"time"
@@ -64,13 +63,13 @@ func (r *Registry) HandleStreamAfterMagic(ctx context.Context, stream *smux.Stre
func (r *Registry) handleStream(ctx context.Context, stream *smux.Stream, connectAddr string, readHello func(io.Reader) (bondframe.Hello, error)) {
defer func() {
if err := stream.Close(); err != nil && err != smux.ErrGoAway {
log.Printf("failed to close bond smux stream: %v", err)
r.deps.log().Errorf("bondserver: close smux stream: %v", err)
}
}()
hello, err := readHello(stream)
if err != nil {
log.Printf("bond hello error: %v", err)
r.deps.log().Errorf("bondserver: bond hello: %v", err)
return
}
@@ -233,22 +232,22 @@ func (c *serverConn) run() {
backendConn, err := net.DialTimeout("tcp", c.connectAddr, 10*time.Second)
if err != nil {
log.Printf("[bond %d] backend dial error: %s", c.id, err)
c.deps.log().Errorf("[bond %d] backend dial: %s", c.id, err)
return
}
defer func() {
if err := backendConn.Close(); err != nil {
log.Printf("[bond %d] failed to close backend connection: %v", c.id, err)
c.deps.log().Errorf("[bond %d] close backend connection: %v", c.id, err)
}
}()
context.AfterFunc(c.ctx, func() {
now := time.Now()
if err := backendConn.SetDeadline(now); err != nil {
log.Printf("[bond %d] backend deadline error: %v", c.id, err)
c.deps.log().Errorf("[bond %d] backend deadline: %v", c.id, err)
}
for _, lane := range c.snapshotLanes() {
if err := lane.stream.SetDeadline(now); err != nil {
log.Printf("[bond %d] lane %d deadline error: %v", c.id, lane.index, err)
c.deps.log().Errorf("[bond %d] lane %d deadline: %v", c.id, lane.index, err)
}
}
})
@@ -287,7 +286,7 @@ func (c *serverConn) copyBondToBackend(backendConn net.Conn) {
// that emits seq with permanent gaps cannot exhaust memory.
// Matches bondclient.copyBondToTCP.
if len(pending) >= 1024 {
log.Printf("[bond %d] pending map overflow (>1024), closing", c.id)
c.deps.log().Errorf("[bond %d] pending map overflow (>1024), closing", c.id)
return
}
pending[f.Seq] = f.Data
@@ -297,7 +296,7 @@ func (c *serverConn) copyBondToBackend(backendConn net.Conn) {
finSeq = &v
}
default:
log.Printf("[bond %d] unknown frame type %d", c.id, f.Type)
c.deps.log().Errorf("[bond %d] unknown frame type %d", c.id, f.Type)
return
}
@@ -309,7 +308,7 @@ func (c *serverConn) copyBondToBackend(backendConn net.Conn) {
delete(pending, expect)
if len(data) > 0 {
if _, err := backendConn.Write(data); err != nil {
log.Printf("[bond %d] backend write error: %v", c.id, err)
c.deps.log().Errorf("[bond %d] backend write error: %v", c.id, err)
return
}
}
@@ -329,7 +328,7 @@ func (c *serverConn) copyBackendToBond(backendConn net.Conn) {
data := make([]byte, n)
copy(data, buf[:n])
if writeErr := c.writeToNextLane(bondframe.FrameData, seq, data, &laneIdx); writeErr != nil {
log.Printf("[bond %d] lane write data error: %v", c.id, writeErr)
c.deps.log().Errorf("[bond %d] lane write data error: %v", c.id, writeErr)
return
}
seq++
@@ -341,7 +340,7 @@ func (c *serverConn) copyBackendToBond(backendConn net.Conn) {
writeErr := bondframe.WriteFrame(lane.stream, bondframe.FrameFIN, seq, nil)
lane.mu.Unlock()
if writeErr != nil && c.ctx.Err() == nil {
log.Printf("[bond %d] lane %d write FIN error: %v", c.id, lane.index, writeErr)
c.deps.log().Errorf("[bond %d] lane %d write FIN error: %v", c.id, lane.index, writeErr)
}
}
c.deps.log().Debugf("[bond %d] download from backend finished chunks=%d", c.id, seq)
@@ -368,7 +367,7 @@ func (c *serverConn) writeToNextLane(typ byte, seq uint64, data []byte, laneIdx
return nil
}
left := c.removeLane(lane)
log.Printf("[bond %d] lane %d write error: %v (lanes=%d)", c.id, lane.index, err, left)
c.deps.log().Errorf("[bond %d] lane %d write error: %v (lanes=%d)", c.id, lane.index, err, left)
if left == 0 {
return err
}
+18 -18
View File
@@ -70,7 +70,7 @@ func Run(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr, lis
})
}
log.Printf("VLESS mode: waiting for sessions to connect (total: %d)...", numSessions)
deps.log().Errorf("VLESS mode: waiting for sessions to connect (total: %d)...", numSessions)
for {
select {
case <-ctx.Done():
@@ -90,15 +90,15 @@ func Run(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr, lis
wrappedListener, err := ish.WrapListener(listener)
if err != nil {
log.Printf("Warning: failed to wrap listener: %v", err)
deps.log().Errorf("Warning: failed to wrap listener: %v", err)
wrappedListener = listener
}
context.AfterFunc(ctx, func() { _ = wrappedListener.Close() })
if useBond {
log.Printf("VLESS bond mode: listening on %s (striping each TCP connection across active sessions)", listenAddr)
deps.log().Errorf("VLESS bond mode: listening on %s (striping each TCP connection across active sessions)", listenAddr)
} else {
log.Printf("VLESS mode: listening on %s (round-robin across %d sessions)", listenAddr, numSessions)
deps.log().Errorf("VLESS mode: listening on %s (round-robin across %d sessions)", listenAddr, numSessions)
}
var wgConn sync.WaitGroup
@@ -112,20 +112,20 @@ func Run(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr, lis
return
default:
}
log.Printf("TCP accept error: %s", err)
deps.log().Errorf("TCP accept error: %s", err)
continue
}
if useBond {
if deps.BondHandler == nil {
log.Printf("bond requested but no BondHandler set, rejecting")
deps.log().Errorf("bond requested but no BondHandler set, rejecting")
_ = tcpConn.Close()
continue
}
connID := (uint64(time.Now().UnixNano()) << 16) ^ pool.NextConnID()
lanes := pool.Snapshot()
if len(lanes) == 0 {
log.Printf("No active sessions, rejecting connection")
deps.log().Errorf("No active sessions, rejecting connection")
_ = tcpConn.Close()
continue
}
@@ -139,7 +139,7 @@ func Run(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr, lis
ps := pool.Pick()
if ps == nil || ps.Sess.IsClosed() {
log.Printf("No active sessions, rejecting connection")
deps.log().Errorf("No active sessions, rejecting connection")
_ = tcpConn.Close()
continue
}
@@ -163,7 +163,7 @@ func Run(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr, lis
stream, err := sessRef.Sess.OpenStream()
if err != nil {
log.Printf("[session %d] smux open stream error for TCP #%d: %s", sessRef.ID, cid, err)
deps.log().Errorf("[session %d] smux open stream error for TCP #%d: %s", sessRef.ID, cid, err)
return
}
defer func() { _ = stream.Close() }()
@@ -188,7 +188,7 @@ func maintainSession(ctx context.Context, deps *Deps, params *Params, peer *net.
smuxSess, cleanup, err := createSmuxSession(ctx, deps, params, peer, id)
if err != nil {
log.Printf("[session %d] setup error: %s, retrying...", id, err)
deps.log().Errorf("[session %d] setup error: %s, retrying...", id, err)
select {
case <-ctx.Done():
return
@@ -198,7 +198,7 @@ func maintainSession(ctx context.Context, deps *Deps, params *Params, peer *net.
}
ps := pool.Add(id, smuxSess)
log.Printf("[session %d] connected (active: %d)", id, pool.Count())
deps.log().Errorf("[session %d] connected (active: %d)", id, pool.Count())
for !smuxSess.IsClosed() {
select {
@@ -212,7 +212,7 @@ func maintainSession(ctx context.Context, deps *Deps, params *Params, peer *net.
pool.Remove(ps)
cleanup()
log.Printf("[session %d] disconnected (active: %d), reconnecting...", id, pool.Count())
deps.log().Errorf("[session %d] disconnected (active: %d), reconnecting...", id, pool.Count())
select {
case <-ctx.Done():
@@ -285,10 +285,10 @@ func pipe(deps *Deps, ctx context.Context, c1, c2 net.Conn) (int64, int64) {
ctx2, cancel := context.WithCancel(ctx)
context.AfterFunc(ctx2, func() {
if err := c1.SetDeadline(time.Now()); err != nil {
log.Printf("pipe: failed to set deadline c1: %v", err)
deps.log().Errorf("pipe: failed to set deadline c1: %v", err)
}
if err := c2.SetDeadline(time.Now()); err != nil {
log.Printf("pipe: failed to set deadline c2: %v", err)
deps.log().Errorf("pipe: failed to set deadline c2: %v", err)
}
})
@@ -300,7 +300,7 @@ func pipe(deps *Deps, ctx context.Context, c1, c2 net.Conn) (int64, int64) {
n, err := io.Copy(c1, c2)
c1FromC2 = n
if err != nil && deps.log().DebugEnabled() {
log.Printf("pipe: c1<-c2 copy error: %v", err)
deps.log().Errorf("pipe: c1<-c2 copy error: %v", err)
}
})
wg.Go(func() {
@@ -308,15 +308,15 @@ func pipe(deps *Deps, ctx context.Context, c1, c2 net.Conn) (int64, int64) {
n, err := io.Copy(c2, c1)
c2FromC1 = n
if err != nil && deps.log().DebugEnabled() {
log.Printf("pipe: c2<-c1 copy error: %v", err)
deps.log().Errorf("pipe: c2<-c1 copy error: %v", err)
}
})
wg.Wait()
if err := c1.SetDeadline(time.Time{}); err != nil && deps.log().DebugEnabled() {
log.Printf("pipe: failed to reset deadline c1: %v", err)
deps.log().Errorf("pipe: failed to reset deadline c1: %v", err)
}
if err := c2.SetDeadline(time.Time{}); err != nil && deps.log().DebugEnabled() {
log.Printf("pipe: failed to reset deadline c2: %v", err)
deps.log().Errorf("pipe: failed to reset deadline c2: %v", err)
}
return c1FromC2, c2FromC1
}
+9 -10
View File
@@ -7,7 +7,6 @@ package tcpfwdserver
import (
"context"
"io"
"log"
"net"
"sync"
"time"
@@ -37,24 +36,24 @@ func Handle(ctx context.Context, logger logx.Logger, registry *bondserver.Regist
kcpSess, err := kcptun.NewKCPOverDTLS(&stats.CountingConn{Conn: dtlsConn, Stats: st}, true)
if err != nil {
log.Printf("KCP session error: %s", err)
logger.Errorf("tcpfwdserver: KCP session: %s", err)
return
}
defer func() {
if closeErr := kcpSess.Close(); closeErr != nil {
log.Printf("failed to close KCP session: %v", closeErr)
logger.Errorf("tcpfwdserver: close KCP session: %v", closeErr)
}
}()
logger.Debugf("KCP session established (server)")
smuxSess, err := smux.Server(kcpSess, kcptun.DefaultSmuxConfig())
if err != nil {
log.Printf("smux server error: %s", err)
logger.Errorf("tcpfwdserver: smux server: %s", err)
return
}
defer func() {
if err := smuxSess.Close(); err != nil {
log.Printf("failed to close smux session: %v", err)
logger.Errorf("tcpfwdserver: close smux session: %v", err)
}
}()
logger.Debugf("smux session established (server)")
@@ -66,7 +65,7 @@ func Handle(ctx context.Context, logger logx.Logger, registry *bondserver.Regist
select {
case <-ctx.Done():
default:
log.Printf("smux accept error: %s", err)
logger.Errorf("tcpfwdserver: smux accept: %s", err)
}
break
}
@@ -83,7 +82,7 @@ func handleStream(ctx context.Context, logger logx.Logger, registry *bondserver.
var prefix [4]byte
if _, err := io.ReadFull(s, prefix[:]); err != nil {
if err != io.EOF && err != io.ErrUnexpectedEOF {
log.Printf("smux stream prefix read error: %v", err)
logger.Errorf("tcpfwdserver: smux stream prefix read: %v", err)
}
_ = s.Close()
return
@@ -96,18 +95,18 @@ func handleStream(ctx context.Context, logger logx.Logger, registry *bondserver.
defer func() {
if err := s.Close(); err != nil && err != smux.ErrGoAway {
log.Printf("failed to close smux stream: %v", err)
logger.Errorf("tcpfwdserver: close smux stream: %v", err)
}
}()
backendConn, err := net.DialTimeout("tcp", connectAddr, 10*time.Second)
if err != nil {
log.Printf("backend dial error: %s", err)
logger.Errorf("tcpfwdserver: backend dial: %s", err)
return
}
defer func() {
if err := backendConn.Close(); err != nil {
log.Printf("failed to close backend connection: %v", err)
logger.Errorf("tcpfwdserver: close backend connection: %v", err)
}
}()
+16 -16
View File
@@ -91,7 +91,7 @@ func Run(ctx context.Context, dtlsDialer *dtlsdial.Dialer, auth AuthHandler, log
}
context.AfterFunc(ctx, func() {
if closeErr := listenConn.Close(); closeErr != nil {
log.Printf("Failed to close local connection: %s", closeErr)
logger.Errorf("udprelay: close local connection: %s", closeErr)
}
})
@@ -240,7 +240,7 @@ func TURNLoop(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr
if err := <-c; err != nil {
if errors.Is(err, vkauth.ErrFatalCaptchaNoStreams) {
log.Printf("[STREAM %d] Fatal manual captcha error. Shutting down application.", streamID)
deps.log().Errorf("[STREAM %d] Fatal manual captcha error. Shutting down application.", streamID)
if deps.AppCancel != nil {
deps.AppCancel()
}
@@ -248,7 +248,7 @@ func TURNLoop(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr
}
if errors.Is(err, vkauth.ErrCaptchaWaitRequired) {
if !errors.Is(err, vkauth.ErrLockoutActive) {
log.Printf("[STREAM %d] Backing off for 60 seconds to avoid IP ban...", streamID)
deps.log().Errorf("[STREAM %d] Backing off for 60 seconds to avoid IP ban...", streamID)
select {
case <-ctx.Done():
return
@@ -267,7 +267,7 @@ func TURNLoop(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr
}
}
} else {
log.Printf("[STREAM %d] %s", streamID, err)
deps.log().Errorf("[STREAM %d] %s", streamID, err)
time.Sleep(2 * time.Second)
}
}
@@ -301,11 +301,11 @@ func oneDTLS(ctx context.Context, deps *Deps, peer *net.UDPAddr, listenConn net.
var dtlsConn net.Conn = dtlsRaw
defer func() {
if closeErr := dtlsConn.Close(); closeErr != nil {
log.Printf("[STREAM %d] failed to close DTLS connection: %s", streamID, closeErr)
deps.log().Errorf("[STREAM %d] failed to close DTLS connection: %s", streamID, closeErr)
}
log.Printf("[STREAM %d] Closed DTLS connection\n", streamID)
deps.log().Errorf("[STREAM %d] Closed DTLS connection\n", streamID)
}()
log.Printf("[STREAM %d] Established DTLS connection!\n", streamID)
deps.log().Errorf("[STREAM %d] Established DTLS connection!\n", streamID)
if okchan != nil {
go func() {
@@ -319,7 +319,7 @@ func oneDTLS(ctx context.Context, deps *Deps, peer *net.UDPAddr, listenConn net.
wg := sync.WaitGroup{}
context.AfterFunc(dtlsctx, func() {
if err := dtlsConn.SetDeadline(time.Now()); err != nil {
log.Printf("[STREAM %d] Warning: SetDeadline failed: %v", streamID, err)
deps.log().Errorf("[STREAM %d] Warning: SetDeadline failed: %v", streamID, err)
}
})
@@ -348,7 +348,7 @@ func oneDTLS(ctx context.Context, deps *Deps, peer *net.UDPAddr, listenConn net.
if peerAddr := deps.ActiveLocalPeer.Load(); peerAddr != nil {
if addr, ok := peerAddr.(net.Addr); ok {
if _, err := listenConn.WriteTo(buf[:n], addr); err != nil {
log.Printf("[STREAM %d] failed to forward packet to local peer: %v", streamID, err)
deps.log().Errorf("[STREAM %d] failed to forward packet to local peer: %v", streamID, err)
}
}
}
@@ -357,7 +357,7 @@ func oneDTLS(ctx context.Context, deps *Deps, peer *net.UDPAddr, listenConn net.
wg.Wait()
if err := dtlsConn.SetDeadline(time.Time{}); err != nil {
log.Printf("[STREAM %d] Failed to clear DTLS deadline: %s", streamID, err)
deps.log().Errorf("[STREAM %d] Failed to clear DTLS deadline: %s", streamID, err)
}
return nil
}
@@ -388,7 +388,7 @@ func oneTURN(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr,
}()
if deps.log().DebugEnabled() {
log.Printf("[STREAM %d] relayed-address=%s", streamID, relayConn.LocalAddr().String())
deps.log().Errorf("[STREAM %d] relayed-address=%s", streamID, relayConn.LocalAddr().String())
}
wg := sync.WaitGroup{}
@@ -398,13 +398,13 @@ func oneTURN(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr,
context.AfterFunc(turnctx, func() {
if err := relayConn.SetDeadline(time.Now()); err != nil {
log.Printf("Failed to set relay deadline: %s", err)
deps.log().Errorf("Failed to set relay deadline: %s", err)
}
})
var internalPipeAddr atomic.Value
wc, wcErr := common.NewClientWrap(params.WrapKey)
if wcErr != nil {
log.Printf("[STREAM %d] WRAP init failed: %v", streamID, wcErr)
deps.log().Errorf("[STREAM %d] WRAP init failed: %v", streamID, wcErr)
turncancel()
return
}
@@ -434,7 +434,7 @@ func oneTURN(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr,
if wc != nil {
written, wrapErr := wc.WrapInto(wireBuf, out)
if wrapErr != nil {
log.Printf("[STREAM %d] WRAP failed: %v", streamID, wrapErr)
deps.log().Errorf("[STREAM %d] WRAP failed: %v", streamID, wrapErr)
return
}
out = wireBuf[:written]
@@ -471,7 +471,7 @@ func oneTURN(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr,
if wc != nil {
m, wrapErr := wc.Unwrap(payload, plain)
if wrapErr != nil {
log.Printf("[STREAM %d] UNWRAP failed: %v (n=%d)", streamID, wrapErr, n)
deps.log().Errorf("[STREAM %d] UNWRAP failed: %v (n=%d)", streamID, wrapErr, n)
continue
}
payload = plain[:m]
@@ -486,6 +486,6 @@ func oneTURN(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr,
wg.Wait()
if err := relayConn.SetDeadline(time.Time{}); err != nil {
log.Printf("Failed to clear relay deadline: %s", err)
deps.log().Errorf("Failed to clear relay deadline: %s", err)
}
}
+11 -12
View File
@@ -5,7 +5,6 @@ package udpserver
import (
"context"
"fmt"
"log"
"net"
"sync"
"time"
@@ -24,12 +23,12 @@ const (
func Handle(ctx context.Context, logger logx.Logger, conn net.Conn, connectAddr string) {
serverConn, err := net.Dial("udp", connectAddr)
if err != nil {
log.Println(err)
logger.Errorf("udpserver: dial backend: %v", err)
return
}
defer func() {
if err = serverConn.Close(); err != nil {
log.Printf("failed to close outgoing connection: %s", err)
logger.Errorf("udpserver: close outgoing connection: %s", err)
}
}()
@@ -46,28 +45,28 @@ func Handle(ctx context.Context, logger logx.Logger, conn net.Conn, connectAddr
context.AfterFunc(ctx2, func() {
if err := conn.SetDeadline(time.Now()); err != nil {
log.Printf("failed to set incoming deadline: %s", err)
logger.Errorf("udpserver: set incoming deadline: %s", err)
}
if err := serverConn.SetDeadline(time.Now()); err != nil {
log.Printf("failed to set outgoing deadline: %s", err)
logger.Errorf("udpserver: set outgoing deadline: %s", err)
}
})
var wg sync.WaitGroup
wg.Go(func() {
defer cancel()
copyOne(ctx2, conn, serverConn, st.AddTx)
copyOne(ctx2, logger, conn, serverConn, st.AddTx)
})
wg.Go(func() {
defer cancel()
copyOne(ctx2, serverConn, conn, st.AddRx)
copyOne(ctx2, logger, serverConn, conn, st.AddRx)
})
wg.Wait()
}
// copyOne reads from src and writes to dst until ctx fires or either side errors.
// Each read/write resets an idle timeout so a stuck side closes instead of hanging.
func copyOne(ctx context.Context, src, dst net.Conn, count func(int)) {
func copyOne(ctx context.Context, logger logx.Logger, src, dst net.Conn, count func(int)) {
buf := make([]byte, udpRelayBufSize)
for {
select {
@@ -76,22 +75,22 @@ func copyOne(ctx context.Context, src, dst net.Conn, count func(int)) {
default:
}
if err := src.SetReadDeadline(time.Now().Add(udpIdleTimeout)); err != nil {
log.Printf("Failed: %s", err)
logger.Errorf("udpserver: set read deadline: %s", err)
return
}
n, err := src.Read(buf)
if err != nil {
log.Printf("Failed: %s", err)
logger.Debugf("udpserver: read: %s", err)
return
}
if werr := dst.SetWriteDeadline(time.Now().Add(udpIdleTimeout)); werr != nil {
log.Printf("Failed: %s", werr)
logger.Errorf("udpserver: set write deadline: %s", werr)
return
}
written, werr := dst.Write(buf[:n])
count(written)
if werr != nil {
log.Printf("Failed: %s", werr)
logger.Debugf("udpserver: write: %s", werr)
return
}
}