对接

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

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) {