对接

internal/integration/live/server.go

@@ -5,7 +5,9 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
 	"log/slog"
+	"net"
 	"os"
 	"strconv"
 	"strings"
@@ -29,6 +31,7 @@ type Manager struct {
 	chat        *bizchat.Usecase
 	logger      *slog.Logger
 	api         *webrtc.API
+	iceMux      io.Closer
 	mu          sync.RWMutex
 	rooms       map[string]*room
 	peers       map[*melody.Session]*peer
@@ -89,19 +92,57 @@ type chatSendPayload struct {
 	Content string `json:"content"`
 }
 
-func newPionAPI() (*webrtc.API, error) {
+func newPionAPI() (*webrtc.API, io.Closer, error) {
 	settings := webrtc.SettingEngine{}
+	fixedPort := envPort("KRA_LIVE_UDP_PORT")
 	minPort := envPort("KRA_LIVE_UDP_MIN_PORT")
 	maxPort := envPort("KRA_LIVE_UDP_MAX_PORT")
+	if fixedPort > 0 && (minPort > 0 || maxPort > 0) {
+		return nil, nil, errors.New("KRA_LIVE_UDP_PORT cannot be combined with KRA_LIVE_UDP_MIN_PORT/KRA_LIVE_UDP_MAX_PORT")
+	}
 	if minPort > 0 || maxPort > 0 {
 		if minPort == 0 || maxPort == 0 || minPort > maxPort {
-			return nil, fmt.Errorf("invalid KRA_LIVE_UDP_MIN_PORT/KRA_LIVE_UDP_MAX_PORT")
+			return nil, nil, fmt.Errorf("invalid KRA_LIVE_UDP_MIN_PORT/KRA_LIVE_UDP_MAX_PORT")
 		}
 		if err := settings.SetEphemeralUDPPortRange(minPort, maxPort); err != nil {
-			return nil, fmt.Errorf("configure live udp port range: %w", err)
+			return nil, nil, fmt.Errorf("configure live udp port range: %w", err)
 		}
 	}
-	return webrtc.NewAPI(webrtc.WithSettingEngine(settings)), nil
+	publicIPs, err := livePublicIPs()
+	if err != nil {
+		return nil, nil, err
+	}
+	if len(publicIPs) > 0 {
+		settings.SetNAT1To1IPs(publicIPs, webrtc.ICECandidateTypeHost)
+	}
+	var mux io.Closer
+	if fixedPort > 0 {
+		packetConn, listenErr := net.ListenPacket("udp4", fmt.Sprintf(":%d", fixedPort))
+		if listenErr != nil {
+			return nil, nil, fmt.Errorf("listen on live udp port %d: %w", fixedPort, listenErr)
+		}
+		udpMux := webrtc.NewICEUDPMux(nil, packetConn)
+		settings.SetICEUDPMux(udpMux)
+		mux = udpMux
+	}
+	return webrtc.NewAPI(webrtc.WithSettingEngine(settings)), mux, nil
+}
+
+func livePublicIPs() ([]string, error) {
+	value := strings.TrimSpace(os.Getenv("KRA_LIVE_PUBLIC_IP"))
+	if value == "" {
+		return nil, nil
+	}
+	items := strings.Split(value, ",")
+	result := make([]string, 0, len(items))
+	for _, item := range items {
+		item = strings.TrimSpace(item)
+		if net.ParseIP(item) == nil {
+			return nil, fmt.Errorf("invalid KRA_LIVE_PUBLIC_IP %q", item)
+		}
+		result = append(result, item)
+	}
+	return result, nil
 }
 
 func envPort(key string) uint16 {
@@ -133,7 +174,7 @@ func New(hub platformws.Hub, streams *bizlive.StreamUsecase, rooms *bizlive.Room
 	if chat == nil {
 		return nil, nil, errors.New("live chat usecase is unavailable")
 	}
-	api, err := newPionAPI()
+	api, iceMux, err := newPionAPI()
 	if err != nil {
 		return nil, nil, err
 	}
@@ -141,12 +182,14 @@ func New(hub platformws.Hub, streams *bizlive.StreamUsecase, rooms *bizlive.Room
 		logger = slog.Default()
 	}
 	logger = logger.With("mod", "live")
-	m := &Manager{hub: hub, streams: streams, roomCatalog: rooms, chat: chat, logger: logger, api: api, rooms: make(map[string]*room), peers: make(map[*melody.Session]*peer), observers: make(map[*melody.Session]struct{}), stop: make(chan struct{})}
+	m := &Manager{hub: hub, streams: streams, roomCatalog: rooms, chat: chat, logger: logger, api: api, iceMux: iceMux, rooms: make(map[string]*room), peers: make(map[*melody.Session]*peer), observers: make(map[*melody.Session]struct{}), stop: make(chan struct{})}
 	hub.OnMessage(m.onMessage)
 	hub.OnDisconnect(m.onDisconnect)
 	go m.heartbeatLoop()
 	logger.Info("live signaling manager initialized",
 		"stun_configured", len(liveICEServers()) > 0,
+		"public_ips", strings.TrimSpace(os.Getenv("KRA_LIVE_PUBLIC_IP")),
+		"udp_port", envPort("KRA_LIVE_UDP_PORT"),
 		"udp_min_port", envPort("KRA_LIVE_UDP_MIN_PORT"),
 		"udp_max_port", envPort("KRA_LIVE_UDP_MAX_PORT"))
 	return m, m.Close, nil
@@ -164,6 +207,11 @@ func (m *Manager) Close() {
 	for _, p := range peers {
 		m.closePeer(p, "server_shutdown")
 	}
+	if m.iceMux != nil {
+		if err := m.iceMux.Close(); err != nil {
+			m.liveLogger().Warn("live UDP mux close failed", "error", err)
+		}
+	}
 }
 
 func (m *Manager) onMessage(session *melody.Session, raw []byte) {

internal/integration/live/server_test.go

@@ -3,6 +3,7 @@ package live
 import (
 	"context"
 	"encoding/json"
+	"strings"
 	"sync"
 	"testing"
 
@@ -480,3 +481,29 @@ func TestClosedSupersededPeerConnectionDoesNotRemovePeer(t *testing.T) {
 }
 
 var _ platformws.Hub = (*liveHubStub)(nil)
+
+func TestNewPionAPIRejectsConflictingUDPConfiguration(t *testing.T) {
+	t.Setenv("KRA_LIVE_UDP_PORT", "40000")
+	t.Setenv("KRA_LIVE_UDP_MIN_PORT", "40000")
+	t.Setenv("KRA_LIVE_UDP_MAX_PORT", "40010")
+
+	_, closer, err := newPionAPI()
+	if closer != nil {
+		_ = closer.Close()
+	}
+	if err == nil || !strings.Contains(err.Error(), "cannot be combined") {
+		t.Fatalf("newPionAPI error = %v, want conflicting UDP configuration", err)
+	}
+}
+
+func TestNewPionAPIRejectsInvalidPublicIP(t *testing.T) {
+	t.Setenv("KRA_LIVE_PUBLIC_IP", "live.example.com")
+
+	_, closer, err := newPionAPI()
+	if closer != nil {
+		_ = closer.Close()
+	}
+	if err == nil || !strings.Contains(err.Error(), "invalid KRA_LIVE_PUBLIC_IP") {
+		t.Fatalf("newPionAPI error = %v, want invalid public IP", err)
+	}
+}