From 322c7001fb58b489ee8fbb992afadf165719c2de Mon Sep 17 00:00:00 2001 From: samosvalishe Date: Mon, 21 Sep 2026 13:36:27 +0300 Subject: [PATCH] =?UTF-8?q?refactor(wire):=20=D0=BE=D0=B4=D0=B8=D0=BD=20li?= =?UTF-8?q?stener=20=D0=BD=D0=B0=20=D0=B2=D1=81=D0=B5=20=D0=BF=D1=80=D0=BE?= =?UTF-8?q?=D1=84=D0=B8=D0=BB=D0=B8=20=D0=BE=D0=B1=D1=84=D1=83=D1=81=D0=BA?= =?UTF-8?q?=D0=B0=D1=86=D0=B8=D0=B8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/wire/codec.go | 40 ++++++++--- internal/wire/listen.go | 4 -- internal/wire/listen_test.go | 1 - internal/wire/relayconn.go | 14 +++- internal/wire/rtpopus/listen.go | 115 ------------------------------- internal/wire/rtpopus2/listen.go | 115 ------------------------------- internal/wire/rtpopus3/listen.go | 113 ------------------------------ 7 files changed, 40 insertions(+), 362 deletions(-) delete mode 100644 internal/wire/rtpopus/listen.go delete mode 100644 internal/wire/rtpopus2/listen.go delete mode 100644 internal/wire/rtpopus3/listen.go diff --git a/internal/wire/codec.go b/internal/wire/codec.go index 9d3619c..faf3b00 100644 --- a/internal/wire/codec.go +++ b/internal/wire/codec.go @@ -55,20 +55,38 @@ func Listen(profile string, addr *net.UDPAddr, key []byte, serverTiming ...time. timing = serverTiming[0] } - var listener dtlsnet.PacketListener - var err error - switch profile { - case ProfileRTPOpus: - listener, err = rtpopus.Listen(addr, key) - case ProfileRTPOpus2: - listener, err = rtpopus2.Listen(addr, key) - case ProfileRTPOpus3: - listener, err = rtpopus3.Listen(addr, key) - default: - return nil, fmt.Errorf("wire: profile %q has no server listener", profile) + newCodec, err := serverCodecFactory(profile, key) + if err != nil { + return nil, err } + listener, err := ListenCodec(addr, newCodec) if err != nil { return nil, err } return shape.WrapPacketListener(listener, timing), nil } + +func serverCodecFactory(profile string, key []byte) (NewServerCodec, error) { + switch profile { + case ProfileRTPOpus: + st, err := rtpopus.NewState(key) + if err != nil { + return nil, err + } + return func() (Codec, error) { return rtpopus.NewConnFromState(st, true) }, nil + case ProfileRTPOpus2: + st, err := rtpopus2.NewState(key) + if err != nil { + return nil, err + } + return func() (Codec, error) { return rtpopus2.NewConnFromState(st, true) }, nil + case ProfileRTPOpus3: + st, err := rtpopus3.NewState(key) + if err != nil { + return nil, err + } + return func() (Codec, error) { return rtpopus3.NewConnFromState(st, true) }, nil + default: + return nil, fmt.Errorf("wire: profile %q has no server listener", profile) + } +} diff --git a/internal/wire/listen.go b/internal/wire/listen.go index 56003a9..dd551c2 100644 --- a/internal/wire/listen.go +++ b/internal/wire/listen.go @@ -8,12 +8,8 @@ import ( pionudp "github.com/pion/transport/v4/udp" ) -// NewServerCodec создаёт кодек для очередного принятого соединения: состояние RTP-сессии -// у каждого клиента своё, общий у профиля только ключ. type NewServerCodec func() (Codec, error) -// ListenCodec - единственная серверная обвязка на все профили: они отличались только -// именем пакета в текстах ошибок. func ListenCodec(addr *net.UDPAddr, newCodec NewServerCodec) (dtlsnet.PacketListener, error) { inner, err := pionudp.Listen("udp", addr) if err != nil { diff --git a/internal/wire/listen_test.go b/internal/wire/listen_test.go index 68001b3..dc0a58d 100644 --- a/internal/wire/listen_test.go +++ b/internal/wire/listen_test.go @@ -7,7 +7,6 @@ import ( "time" ) -// Серверная обвязка общая для всех профилей - проверяем каждый через реальный UDP. func TestListenCodecRoundTrip(t *testing.T) { t.Parallel() diff --git a/internal/wire/relayconn.go b/internal/wire/relayconn.go index 41d0bc8..75848de 100644 --- a/internal/wire/relayconn.go +++ b/internal/wire/relayconn.go @@ -20,6 +20,13 @@ type RelayPacketConn struct { Codec Codec } +func (r *RelayPacketConn) dst(addr net.Addr) net.Addr { + if r.Peer != nil { + return r.Peer + } + return addr +} + func (r *RelayPacketConn) ReadFrom(b []byte) (int, net.Addr, error) { if r.Codec == nil { return r.Relay.ReadFrom(b) @@ -44,9 +51,10 @@ func (r *RelayPacketConn) ReadFrom(b []byte) (int, net.Addr, error) { return m, addr, nil } -func (r *RelayPacketConn) WriteTo(b []byte, _ net.Addr) (int, error) { +func (r *RelayPacketConn) WriteTo(b []byte, addr net.Addr) (int, error) { + peer := r.dst(addr) if r.Codec == nil { - return r.Relay.WriteTo(b, r.Peer) + return r.Relay.WriteTo(b, peer) } wireLen := r.Codec.MaxWire(len(b)) @@ -63,7 +71,7 @@ func (r *RelayPacketConn) WriteTo(b []byte, _ net.Addr) (int, error) { if err != nil { return 0, err } - if _, err = r.Relay.WriteTo(out[:n], r.Peer); err != nil { + if _, err = r.Relay.WriteTo(out[:n], peer); err != nil { return 0, err } return len(b), nil diff --git a/internal/wire/rtpopus/listen.go b/internal/wire/rtpopus/listen.go deleted file mode 100644 index 228fcf5..0000000 --- a/internal/wire/rtpopus/listen.go +++ /dev/null @@ -1,115 +0,0 @@ -// SPDX-License-Identifier: MIT - -package rtpopus - -import ( - "errors" - "fmt" - "net" - "sync" - "time" - - dtlsnet "github.com/pion/dtls/v3/pkg/net" - pionudp "github.com/pion/transport/v4/udp" -) - -var bufPool = sync.Pool{ - New: func() any { - b := make([]byte, 1600+Overhead) - return &b - }, -} - -// Listen возвращает dtls PacketListener, AEAD-оборачивающий каждый принятый PacketConn. -func Listen(addr *net.UDPAddr, key []byte) (dtlsnet.PacketListener, error) { - state, err := NewState(key) - if err != nil { - return nil, err - } - inner, err := pionudp.Listen("udp", addr) - if err != nil { - return nil, fmt.Errorf("rtpopus:udp listen: %w", err) - } - return &packetListener{ - inner: dtlsnet.PacketListenerFromListener(inner), - state: state, - }, nil -} - -type packetListener struct { - inner dtlsnet.PacketListener - state *State -} - -func (l *packetListener) Accept() (net.PacketConn, net.Addr, error) { - pc, addr, err := l.inner.Accept() - if err != nil { - return pc, addr, err - } - conn, err := NewConnFromState(l.state, true) - if err != nil { - return nil, addr, err - } - return &packetConn{inner: pc, conn: conn}, addr, nil -} - -func (l *packetListener) Close() error { return l.inner.Close() } -func (l *packetListener) Addr() net.Addr { return l.inner.Addr() } - -type packetConn struct { - inner net.PacketConn - conn *Conn -} - -func (c *packetConn) ReadFrom(p []byte) (int, net.Addr, error) { - bp := bufPool.Get().(*[]byte) //nolint:errcheck // pool New always returns *[]byte - buf := *bp - need := len(p) + Overhead - if cap(buf) < need { - buf = make([]byte, need) - *bp = buf - } - defer bufPool.Put(bp) - - n, addr, err := c.inner.ReadFrom(buf[:cap(buf)]) - if err != nil { - return 0, addr, err - } - wire := buf[:n] - if len(wire) < Overhead { - return 0, addr, errors.New("rtpopus:packet too short") - } - m, err := c.conn.Unwrap(wire, p) - if err != nil { - return 0, addr, err - } - return m, addr, nil -} - -func (c *packetConn) WriteTo(p []byte, addr net.Addr) (int, error) { - wireLen := Overhead + len(p) - - bp := bufPool.Get().(*[]byte) //nolint:errcheck // pool New always returns *[]byte - out := *bp - if cap(out) < wireLen { - out = make([]byte, wireLen) - *bp = out - } - out = out[:wireLen] - defer bufPool.Put(bp) - - n, err := c.conn.WrapInto(out, p) - if err != nil { - return 0, err - } - if _, err := c.inner.WriteTo(out[:n], addr); err != nil { - return 0, err - } - return len(p), nil -} - -func (c *packetConn) Close() error { return c.inner.Close() } -func (c *packetConn) LocalAddr() net.Addr { return c.inner.LocalAddr() } -func (c *packetConn) SetDeadline(t time.Time) error { return c.inner.SetDeadline(t) } -func (c *packetConn) SetReadDeadline(t time.Time) error { return c.inner.SetReadDeadline(t) } -func (c *packetConn) SetWriteDeadline(t time.Time) error { return c.inner.SetWriteDeadline(t) } diff --git a/internal/wire/rtpopus2/listen.go b/internal/wire/rtpopus2/listen.go deleted file mode 100644 index 21ccd81..0000000 --- a/internal/wire/rtpopus2/listen.go +++ /dev/null @@ -1,115 +0,0 @@ -// SPDX-License-Identifier: MIT - -package rtpopus2 - -import ( - "errors" - "fmt" - "net" - "sync" - "time" - - dtlsnet "github.com/pion/dtls/v3/pkg/net" - pionudp "github.com/pion/transport/v4/udp" -) - -var bufPool = sync.Pool{ - New: func() any { - b := make([]byte, 1600+overhead) - return &b - }, -} - -// Listen возвращает dtls PacketListener, AEAD-оборачивающий каждый принятый PacketConn. -func Listen(addr *net.UDPAddr, key []byte) (dtlsnet.PacketListener, error) { - state, err := NewState(key) - if err != nil { - return nil, err - } - inner, err := pionudp.Listen("udp", addr) - if err != nil { - return nil, fmt.Errorf("rtpopus2:udp listen: %w", err) - } - return &packetListener{ - inner: dtlsnet.PacketListenerFromListener(inner), - state: state, - }, nil -} - -type packetListener struct { - inner dtlsnet.PacketListener - state *State -} - -func (l *packetListener) Accept() (net.PacketConn, net.Addr, error) { - pc, addr, err := l.inner.Accept() - if err != nil { - return pc, addr, err - } - conn, err := NewConnFromState(l.state, true) - if err != nil { - return nil, addr, err - } - return &packetConn{inner: pc, conn: conn}, addr, nil -} - -func (l *packetListener) Close() error { return l.inner.Close() } -func (l *packetListener) Addr() net.Addr { return l.inner.Addr() } - -type packetConn struct { - inner net.PacketConn - conn *Conn -} - -func (c *packetConn) ReadFrom(p []byte) (int, net.Addr, error) { - bp := bufPool.Get().(*[]byte) //nolint:errcheck // pool New always returns *[]byte - buf := *bp - need := len(p) + overhead - if cap(buf) < need { - buf = make([]byte, need) - *bp = buf - } - defer bufPool.Put(bp) - - n, addr, err := c.inner.ReadFrom(buf[:cap(buf)]) - if err != nil { - return 0, addr, err - } - wire := buf[:n] - if len(wire) < overhead { - return 0, addr, errors.New("rtpopus2:packet too short") - } - m, err := c.conn.Unwrap(wire, p) - if err != nil { - return 0, addr, err - } - return m, addr, nil -} - -func (c *packetConn) WriteTo(p []byte, addr net.Addr) (int, error) { - wireLen := overhead + len(p) - - bp := bufPool.Get().(*[]byte) //nolint:errcheck // pool New always returns *[]byte - out := *bp - if cap(out) < wireLen { - out = make([]byte, wireLen) - *bp = out - } - out = out[:wireLen] - defer bufPool.Put(bp) - - n, err := c.conn.WrapInto(out, p) - if err != nil { - return 0, err - } - if _, err := c.inner.WriteTo(out[:n], addr); err != nil { - return 0, err - } - return len(p), nil -} - -func (c *packetConn) Close() error { return c.inner.Close() } -func (c *packetConn) LocalAddr() net.Addr { return c.inner.LocalAddr() } -func (c *packetConn) SetDeadline(t time.Time) error { return c.inner.SetDeadline(t) } -func (c *packetConn) SetReadDeadline(t time.Time) error { return c.inner.SetReadDeadline(t) } -func (c *packetConn) SetWriteDeadline(t time.Time) error { return c.inner.SetWriteDeadline(t) } diff --git a/internal/wire/rtpopus3/listen.go b/internal/wire/rtpopus3/listen.go deleted file mode 100644 index d0c7835..0000000 --- a/internal/wire/rtpopus3/listen.go +++ /dev/null @@ -1,113 +0,0 @@ -package rtpopus3 - -import ( - "errors" - "fmt" - "net" - "sync" - "time" - - dtlsnet "github.com/pion/dtls/v3/pkg/net" - pionudp "github.com/pion/transport/v4/udp" -) - -var bufPool = sync.Pool{ - New: func() any { - b := make([]byte, 1600+overhead) - return &b - }, -} - -// Listen возвращает dtls PacketListener с поддержкой профиля rtpopus3. -func Listen(addr *net.UDPAddr, key []byte) (dtlsnet.PacketListener, error) { - state, err := NewState(key) - if err != nil { - return nil, err - } - inner, err := pionudp.Listen("udp", addr) - if err != nil { - return nil, fmt.Errorf("rtpopus3:udp listen: %w", err) - } - return &packetListener{ - inner: dtlsnet.PacketListenerFromListener(inner), - state: state, - }, nil -} - -type packetListener struct { - inner dtlsnet.PacketListener - state *State -} - -func (l *packetListener) Accept() (net.PacketConn, net.Addr, error) { - pc, addr, err := l.inner.Accept() - if err != nil { - return pc, addr, err - } - conn, err := NewConnFromState(l.state, true) - if err != nil { - return nil, addr, err - } - return &packetConn{inner: pc, conn: conn}, addr, nil -} - -func (l *packetListener) Close() error { return l.inner.Close() } -func (l *packetListener) Addr() net.Addr { return l.inner.Addr() } - -type packetConn struct { - inner net.PacketConn - conn *Conn -} - -func (c *packetConn) ReadFrom(p []byte) (int, net.Addr, error) { - bp := bufPool.Get().(*[]byte) //nolint:errcheck // pool New always returns *[]byte - buf := *bp - need := len(p) + overhead - if cap(buf) < need { - buf = make([]byte, need) - *bp = buf - } - defer bufPool.Put(bp) - - n, addr, err := c.inner.ReadFrom(buf[:cap(buf)]) - if err != nil { - return 0, addr, err - } - wire := buf[:n] - if len(wire) < overhead { - return 0, addr, errors.New("rtpopus3:packet too short") - } - m, err := c.conn.Unwrap(wire, p) - if err != nil { - return 0, addr, err - } - return m, addr, nil -} - -func (c *packetConn) WriteTo(p []byte, addr net.Addr) (int, error) { - wireLen := overhead + len(p) - - bp := bufPool.Get().(*[]byte) //nolint:errcheck // pool New always returns *[]byte - out := *bp - if cap(out) < wireLen { - out = make([]byte, wireLen) - *bp = out - } - out = out[:wireLen] - defer bufPool.Put(bp) - - n, err := c.conn.WrapInto(out, p) - if err != nil { - return 0, err - } - if _, err := c.inner.WriteTo(out[:n], addr); err != nil { - return 0, err - } - return len(p), nil -} - -func (c *packetConn) Close() error { return c.inner.Close() } -func (c *packetConn) LocalAddr() net.Addr { return c.inner.LocalAddr() } -func (c *packetConn) SetDeadline(t time.Time) error { return c.inner.SetDeadline(t) } -func (c *packetConn) SetReadDeadline(t time.Time) error { return c.inner.SetReadDeadline(t) } -func (c *packetConn) SetWriteDeadline(t time.Time) error { return c.inner.SetWriteDeadline(t) }