refactor(wire): один listener на все профили обфускации

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)
+	}
+}

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 {

internal/wire/listen_test.go

@@ -7,7 +7,6 @@ import (
 	"time"
 )
 
-// Серверная обвязка общая для всех профилей - проверяем каждый через реальный UDP.
 func TestListenCodecRoundTrip(t *testing.T) {
 	t.Parallel()
 

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

internal/wire/rtpopus/listen.go

@@ -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) }

internal/wire/rtpopus2/listen.go

@@ -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) }

internal/wire/rtpopus3/listen.go

@@ -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) }