Merge pull request #45 from matrix-org/SimonBrandner/feat/forward-rtcp

Author: SimonBrandner

src/publisher.go

@@ -20,13 +20,19 @@ import (
 	"errors"
 	"io"
 	"sync"
+	"sync/atomic"
+	"time"
 
+	"github.com/pion/rtcp"
 	"github.com/pion/webrtc/v3"
 	"github.com/sirupsen/logrus"
 	"maunium.net/go/mautrix/event"
 )
 
-const bufferSize = 1500
+const (
+	minimalPLIInterval = time.Millisecond * 500
+	bufferSize         = 1500
+)
 
 type Publisher struct {
 	Track *webrtc.TrackRemote
@@ -35,6 +41,8 @@ type Publisher struct {
 	mutex       sync.RWMutex
 	logger      *logrus.Entry
 	subscribers []*Subscriber
+
+	lastPLI atomic.Int64
 }
 
 func NewPublisher(
@@ -57,7 +65,6 @@ func NewPublisher(
 	call.Publishers = append(call.Publishers, publisher)
 	call.mutex.Unlock()
 
-	go WriteRTCP(track, call.PeerConnection, publisher.logger)
 	go publisher.WriteToSubscribers()
 
 	publisher.logger.Info("published track")
@@ -118,6 +125,46 @@ func (p *Publisher) Matches(trackDescription event.SFUTrackDescription) bool {
 	return true
 }
 
+func (p *Publisher) WriteRTCP(packets []rtcp.Packet) {
+	packetsToSend := []rtcp.Packet{}
+	readSSRC := uint32(p.Track.SSRC())
+
+	for _, packet := range packets {
+		switch typedPacket := packet.(type) {
+		// We mung the packets here, so that the SSRCs match what the
+		// receiver expects:
+		// The media SSRC is the SSRC of the media about which the packet is
+		// reporting; therefore, we mung it to be the SSRC of the publishing
+		// participant's track. Without this, it would be SSRC of the SFU's
+		// track which isn't right
+		case *rtcp.PictureLossIndication:
+			// Since we sometimes spam the sender with PLIs, make sure we don't send
+			// them way too often
+			if time.Now().UnixNano()-p.lastPLI.Load() < minimalPLIInterval.Nanoseconds() {
+				continue
+			}
+
+			p.lastPLI.Store(time.Now().UnixNano())
+
+			typedPacket.MediaSSRC = readSSRC
+			packetsToSend = append(packetsToSend, typedPacket)
+		case *rtcp.FullIntraRequest:
+			typedPacket.MediaSSRC = readSSRC
+			packetsToSend = append(packetsToSend, typedPacket)
+		}
+
+		packetsToSend = append(packetsToSend, packet)
+	}
+
+	if len(packetsToSend) != 0 {
+		if err := p.Call.PeerConnection.WriteRTCP(packetsToSend); err != nil {
+			if !errors.Is(err, io.ErrClosedPipe) {
+				p.logger.WithError(err).Warn("failed to write RTCP on track")
+			}
+		}
+	}
+}
+
 func (p *Publisher) WriteToSubscribers() {
 	buff := make([]byte, bufferSize)
 

src/subscriber.go

@@ -17,6 +17,8 @@ limitations under the License.
 package main
 
 import (
+	"errors"
+	"io"
 	"sync"
 
 	"github.com/pion/webrtc/v3"
@@ -59,6 +61,10 @@ func (s *Subscriber) initLoggingWithTrack(track *webrtc.TrackRemote) {
 func (s *Subscriber) Subscribe(publisher *Publisher) {
 	s.initLoggingWithTrack(publisher.Track)
 
+	if s.publisher != nil {
+		s.logger.Error("cannot subscribe, if we already are")
+	}
+
 	track, err := webrtc.NewTrackLocalStaticRTP(
 		publisher.Track.Codec().RTPCodecCapability,
 		publisher.Track.ID(),
@@ -79,6 +85,10 @@ func (s *Subscriber) Subscribe(publisher *Publisher) {
 	s.publisher = publisher
 	s.mutex.Unlock()
 
+	if s.Track.Kind() == webrtc.RTPCodecTypeVideo {
+		go s.forwardRTCP()
+	}
+
 	publisher.AddSubscriber(s)
 
 	s.logger.Info("subscribed")
@@ -104,3 +114,23 @@ func (s *Subscriber) Unsubscribe() {
 
 	s.logger.Info("unsubscribed")
 }
+
+func (s *Subscriber) forwardRTCP() {
+	for {
+		// If we unsubscribed, stop forwarding RTCP packets
+		if s.publisher == nil {
+			return
+		}
+
+		packets, _, err := s.sender.ReadRTCP()
+		if err != nil {
+			if errors.Is(err, io.ErrClosedPipe) || errors.Is(err, io.EOF) {
+				return
+			}
+
+			s.logger.WithError(err).Warn("failed to read RTCP on track")
+		}
+
+		s.publisher.WriteRTCP(packets)
+	}
+}