Forward RTCP packets from receiver to sender

Signed-off-by: Šimon Brandner <[email protected]>

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,36 @@ func (p *Publisher) Matches(trackDescription event.SFUTrackDescription) bool {
 	return true
 }
 
+func (p *Publisher) WriteRTCP(packets []rtcp.Packet) {
+	packetsToSend := []rtcp.Packet{}
+
+	for _, packet := range packets {
+		// Since we sometimes spam the sender with PLIs, make sure we don't send
+		// them way too often
+		if _, ok := packet.(*rtcp.PictureLossIndication); ok {
+			if time.Now().UnixNano()-p.lastPLI.Load() < minimalPLIInterval.Nanoseconds() {
+				continue
+			}
+
+			p.lastPLI.Store(time.Now().UnixNano())
+		}
+
+		packetsToSend = append(packetsToSend, packet)
+	}
+
+	if len(packetsToSend) < 1 {
+		return
+	}
+
+	if err := p.Call.PeerConnection.WriteRTCP(packetsToSend); err != nil {
+		if errors.Is(err, io.ErrClosedPipe) {
+			return
+		}
+
+		p.logger.WithError(err).Warn("failed to write RTCP on track")
+	}
+}
+
 func (p *Publisher) WriteToSubscribers() {
 	buff := make([]byte, bufferSize)
 

src/subscriber.go

@@ -17,8 +17,11 @@ limitations under the License.
 package main
 
 import (
+	"errors"
+	"io"
 	"sync"
 
+	"github.com/pion/rtcp"
 	"github.com/pion/webrtc/v3"
 	"github.com/sirupsen/logrus"
 )
@@ -31,6 +34,8 @@ type Subscriber struct {
 	call      *Call
 	sender    *webrtc.RTPSender
 	publisher *Publisher
+
+	ssrc uint32
 }
 
 func NewSubscriber(call *Call) *Subscriber {
@@ -76,9 +81,12 @@ func (s *Subscriber) Subscribe(publisher *Publisher) {
 	s.mutex.Lock()
 	s.Track = track
 	s.sender = sender
+	s.ssrc = uint32(sender.GetParameters().Encodings[0].SSRC)
 	s.publisher = publisher
 	s.mutex.Unlock()
 
+	go s.writeRTCP()
+
 	publisher.AddSubscriber(s)
 
 	s.logger.Info("subscribed")
@@ -104,3 +112,44 @@ func (s *Subscriber) Unsubscribe() {
 
 	s.logger.Info("unsubscribed")
 }
+
+func (s *Subscriber) writeRTCP() {
+	if s.Track.Kind() != webrtc.RTPCodecTypeVideo {
+		return
+	}
+
+	for {
+		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")
+		}
+
+		packetsToForward := []rtcp.Packet{}
+		readSSRC := uint32(s.publisher.Track.SSRC())
+
+		for _, packet := range packets {
+			switch typedPacket := packet.(type) {
+			// We mung the packets here, so that the SSRCs match what the
+			// receiver expects
+			case *rtcp.PictureLossIndication:
+				typedPacket.SenderSSRC = s.ssrc
+				typedPacket.MediaSSRC = readSSRC
+				packetsToForward = append(packetsToForward, typedPacket)
+			case *rtcp.FullIntraRequest:
+				typedPacket.SenderSSRC = s.ssrc
+				typedPacket.MediaSSRC = readSSRC
+				packetsToForward = append(packetsToForward, typedPacket)
+			}
+		}
+
+		if len(packetsToForward) < 1 {
+			continue
+		}
+
+		s.publisher.WriteRTCP(packetsToForward)
+	}
+}