diff --git a/deploy/deploy.sh b/deploy/deploy.sh index b8bb1a5..e0f4d15 100644 --- a/deploy/deploy.sh +++ b/deploy/deploy.sh @@ -26,7 +26,7 @@ FRONTEND_PORT=9501 # admin 管理后台 nginx 容器 BACKEND_PORT=9502 # 后端 API/WebSocket 容器 LIVEWEB_PORT=9503 # live 用户端(PC 直播页) nginx 容器 LIVE_UDP_PORT="${KRA_LIVE_UDP_PORT:-9504}" # WebRTC 音视频固定 UDP 端口(与 9501-9503 顺延配套) -LIVE_PUBLIC_IP="${KRA_LIVE_PUBLIC_IP:-}" # 留空时从站点域名解析 +LIVE_PUBLIC_IP="${KRA_LIVE_PUBLIC_IP:-192.168.2.226}" # 留空时从站点域名解析 LIVE_HOST="${KRA_LIVE_HOST:-live.fuhua91.com}" NETWORK="live-net" BACKEND_NAME="live-backend" @@ -53,7 +53,14 @@ if [ ! -f "$CONFIG_DIR/config.yaml" ]; then fi if [ -z "$LIVE_PUBLIC_IP" ]; then - LIVE_PUBLIC_IP="$(getent ahostsv4 "$LIVE_HOST" | awk 'NR == 1 { print $1 }')" + # 1) 域名解析,过滤内网/回环地址(服务器 /etc/hosts 可能指向内网) + LIVE_PUBLIC_IP="$(getent ahostsv4 "$LIVE_HOST" | awk '{print $1}' | grep -E '^[0-9]+(\.[0-9]+){3}$' \ + | grep -vE '^(10\.|127\.|169\.254\.|172\.(1[6-9]|2[0-9]|3[01])\.|192\.168\.)' | head -n1)" +fi +if [ -z "$LIVE_PUBLIC_IP" ]; then + # 2) 域名取不到公网 IP 时,回退查询公网出口 IP(云主机 EIP 1:1 NAT 场景一致) + LIVE_PUBLIC_IP="$(curl -4 -fsS --max-time 5 https://api.ipify.org 2>/dev/null \ + || curl -4 -fsS --max-time 5 https://ifconfig.me 2>/dev/null || true)" fi if ! printf '%s' "$LIVE_PUBLIC_IP" | grep -Eq '^[0-9]+(\.[0-9]+){3}$'; then echo "错误: 无法确定直播服务公网 IPv4,请通过 KRA_LIVE_PUBLIC_IP 显式设置" diff --git a/go.mod b/go.mod index cb7dbb2..58096f2 100644 --- a/go.mod +++ b/go.mod @@ -28,6 +28,7 @@ require ( github.com/minio/minio-go/v7 v7.0.92 github.com/mojocn/base64Captcha v1.3.8 github.com/olahol/melody v1.4.0 + github.com/pion/rtcp v1.2.17 github.com/pion/webrtc/v4 v4.2.20 github.com/qiniu/go-sdk/v7 v7.25.2 github.com/rabbitmq/amqp091-go v1.14.0 @@ -153,7 +154,6 @@ require ( github.com/pion/logging v0.2.4 // indirect github.com/pion/mdns/v2 v2.2.0 // indirect github.com/pion/randutil v0.1.0 // indirect - github.com/pion/rtcp v1.2.17 // indirect github.com/pion/rtp v1.10.5 // indirect github.com/pion/sctp v1.11.1 // indirect github.com/pion/sdp/v3 v3.0.19 // indirect diff --git a/internal/integration/live/server.go b/internal/integration/live/server.go index fad1bc8..e3b51ee 100644 --- a/internal/integration/live/server.go +++ b/internal/integration/live/server.go @@ -21,6 +21,7 @@ import ( platformws "kra/pkg/websocket" "github.com/olahol/melody" + "github.com/pion/rtcp" "github.com/pion/webrtc/v4" ) @@ -51,9 +52,10 @@ type room struct { } type publishedTrack struct { - id string - kind webrtc.RTPCodecType - local *webrtc.TrackLocalStaticRTP + id string + kind webrtc.RTPCodecType + mediaSSRC uint32 + local *webrtc.TrackLocalStaticRTP } type peer struct { @@ -519,7 +521,7 @@ func (m *Manager) handleOffer(session *melody.Session, payload sessionDescriptio } else { for _, track := range m.trackSnapshot(p.room) { if sender, addErr := pc.AddTrack(track.local); addErr == nil { - go drainRTCP(sender) + go m.forwardSubscriberRTCP(p.room, track, sender) } } } @@ -555,7 +557,14 @@ func (m *Manager) handleOffer(session *melody.Session, payload sessionDescriptio func (m *Manager) handleICEConnectionStateChange(p *peer, pc *webrtc.PeerConnection, state webrtc.ICEConnectionState) { m.liveLogger().Info("live ICE state changed", "role", p.role, "stream_id", p.streamID, "user_id", p.userID, "state", state.String()) - if state != webrtc.ICEConnectionStateFailed && state != webrtc.ICEConnectionStateClosed { + if state == webrtc.ICEConnectionStateConnected && p.role == "subscriber" && p.hasPeerConnection(pc) { + m.requestRoomVideoKeyFrames(p.room) + return + } + // The viewer recreates its PeerConnection when publisher tracks change. + // Closing that connection is part of renegotiation; Leave or WebSocket + // disconnect remains responsible for cleaning up the signaling peer. + if state != webrtc.ICEConnectionStateFailed { return } m.closePeerConnection(p, pc, "ice_"+state.String()) @@ -568,7 +577,7 @@ func (m *Manager) onPublisherTrack(p *peer, remote *webrtc.TrackRemote) { m.log("live track creation failed", "error", err) return } - entry := &publishedTrack{id: remote.ID(), kind: remote.Kind(), local: local} + entry := &publishedTrack{id: remote.ID(), kind: remote.Kind(), mediaSSRC: uint32(remote.SSRC()), local: local} m.liveLogger().Info("live publisher track received", "stream_id", p.streamID, "user_id", p.userID, "track_id", remote.ID(), "kind", strings.ToLower(remote.Kind().String()), "codec", codec.MimeType) p.room.mu.Lock() p.room.tracks[remote.Kind()] = entry @@ -883,10 +892,45 @@ func errorCode(err error) string { } } -func drainRTCP(sender *webrtc.RTPSender) { +func (m *Manager) forwardSubscriberRTCP(r *room, track *publishedTrack, sender *webrtc.RTPSender) { for { - if _, _, err := sender.ReadRTCP(); err != nil { + packets, _, err := sender.ReadRTCP() + if err != nil { return } + for _, packet := range packets { + switch packet.(type) { + case *rtcp.PictureLossIndication, *rtcp.FullIntraRequest: + m.requestKeyFrame(r, track.mediaSSRC) + } + } } } + +func (m *Manager) requestRoomVideoKeyFrames(r *room) { + for _, track := range m.trackSnapshot(r) { + if track.kind == webrtc.RTPCodecTypeVideo { + m.requestKeyFrame(r, track.mediaSSRC) + } + } +} + +func (m *Manager) requestKeyFrame(r *room, mediaSSRC uint32) { + publisher := m.publisher(r) + if publisher == nil { + return + } + pc := publisher.peerConnection() + if pc == nil { + return + } + if err := pc.WriteRTCP([]rtcp.Packet{&rtcp.PictureLossIndication{MediaSSRC: mediaSSRC}}); err != nil { + m.liveLogger().Debug("live key frame request failed", "stream_id", r.streamID, "error", err) + } +} + +func (p *peer) peerConnection() *webrtc.PeerConnection { + p.pcMu.RLock() + defer p.pcMu.RUnlock() + return p.pc +} diff --git a/internal/integration/live/server_test.go b/internal/integration/live/server_test.go index 8e663fb..c73a31b 100644 --- a/internal/integration/live/server_test.go +++ b/internal/integration/live/server_test.go @@ -480,6 +480,27 @@ func TestClosedSupersededPeerConnectionDoesNotRemovePeer(t *testing.T) { } } +func TestClosedCurrentPeerConnectionDoesNotRemoveSignalingPeer(t *testing.T) { + pc, err := webrtc.NewPeerConnection(webrtc.Configuration{}) + if err != nil { + t.Fatal(err) + } + defer pc.Close() + + p := &peer{pc: pc} + manager := &Manager{} + manager.handleICEConnectionStateChange(p, pc, webrtc.ICEConnectionStateClosed) + + cleanupAvailable := false + p.cleanup.Do(func() { cleanupAvailable = true }) + if !cleanupAvailable { + t.Fatal("closed media connection consumed signaling peer cleanup") + } + if !p.hasPeerConnection(pc) { + t.Fatal("closed media connection removed signaling peer before renegotiation") + } +} + var _ platformws.Hub = (*liveHubStub)(nil) func TestNewPionAPIRejectsConflictingUDPConfiguration(t *testing.T) {