rtsp server: generate RTCP sender reports automatically; stop routing RTCP packets

Makefile

@@ -85,7 +85,6 @@ test-internal:
 	./internal/hls \
 	./internal/logger \
 	./internal/rlimit \
-	./internal/rtcpsenderset \
 	./internal/rtmp
 
 test-core:

go.mod

@@ -4,7 +4,7 @@ go 1.17
 
 require (
 	code.cloudfoundry.org/bytefmt v0.0.0-20211005130812-5bb3c17173e5
-	github.com/aler9/gortsplib v0.0.0-20220324114420-7f6383aa3cc4
+	github.com/aler9/gortsplib v0.0.0-20220324142719-7d9c882cc95b
 	github.com/asticode/go-astits v1.10.0
 	github.com/fsnotify/fsnotify v1.4.9
 	github.com/gin-gonic/gin v1.7.2
@@ -12,7 +12,6 @@ require (
 	github.com/grafov/m3u8 v0.11.1
 	github.com/kballard/go-shellquote v0.0.0-20180428030007-95032a82bc51
 	github.com/notedit/rtmp v0.0.2
-	github.com/pion/rtcp v1.2.9
 	github.com/pion/rtp/v2 v2.0.0-20220302185659-b3d10fc096b0
 	github.com/stretchr/testify v1.7.0
 	golang.org/x/crypto v0.0.0-20201221181555-eec23a3978ad
@@ -37,6 +36,7 @@ require (
 	github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421 // indirect
 	github.com/modern-go/reflect2 v0.0.0-20180701023420-4b7aa43c6742 // indirect
 	github.com/pion/randutil v0.1.0 // indirect
+	github.com/pion/rtcp v1.2.9 // indirect
 	github.com/pion/sdp/v3 v3.0.2 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/ugorji/go/codec v1.1.7 // indirect

go.sum

@@ -4,8 +4,8 @@ github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751 h1:JYp7IbQjafo
 github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
 github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d h1:UQZhZ2O0vMHr2cI+DC1Mbh0TJxzA3RcLoMsFw+aXw7E=
 github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d/go.mod h1:rBZYJk541a8SKzHPHnH3zbiI+7dagKZ0cgpgrD7Fyho=
-github.com/aler9/gortsplib v0.0.0-20220324114420-7f6383aa3cc4 h1:5WaJOApfEcrysFPHDBCN1XGc2JPYtEPjPC/G1LkTok8=
-github.com/aler9/gortsplib v0.0.0-20220324114420-7f6383aa3cc4/go.mod h1:4HE78w95Rqw1B2T90CHwtA4xBPPCRZ7/G8ds8ZdWcFk=
+github.com/aler9/gortsplib v0.0.0-20220324142719-7d9c882cc95b h1:tUbz82mIgMWTVI0pKZGHoL//c+y0VVGIMPHdYeAyvdc=
+github.com/aler9/gortsplib v0.0.0-20220324142719-7d9c882cc95b/go.mod h1:4HE78w95Rqw1B2T90CHwtA4xBPPCRZ7/G8ds8ZdWcFk=
 github.com/aler9/rtmp v0.0.0-20210403095203-3be4a5535927 h1:95mXJ5fUCYpBRdSOnLAQAdJHHKxxxJrVCiaqDi965YQ=
 github.com/aler9/rtmp v0.0.0-20210403095203-3be4a5535927/go.mod h1:vzuE21rowz+lT1NGsWbreIvYulgBpCGnQyeTyFblUHc=
 github.com/asticode/go-astikit v0.20.0 h1:+7N+J4E4lWx2QOkRdOf6DafWJMv6O4RRfgClwQokrH8=

internal/core/hls_muxer.go

@@ -17,7 +17,6 @@ import (
 	"github.com/aler9/gortsplib/pkg/ringbuffer"
 	"github.com/aler9/gortsplib/pkg/rtpaac"
 	"github.com/aler9/gortsplib/pkg/rtph264"
-	"github.com/pion/rtcp"
 	"github.com/pion/rtp/v2"
 
 	"github.com/aler9/rtsp-simple-server/internal/conf"
@@ -541,10 +540,6 @@ func (m *hlsMuxer) onReaderPacketRTP(trackID int, pkt *rtp.Packet) {
 	m.ringBuffer.Push(hlsMuxerTrackIDPayloadPair{trackID, pkt})
 }
 
-// onReaderPacketRTCP implements reader.
-func (m *hlsMuxer) onReaderPacketRTCP(trackID int, pkt rtcp.Packet) {
-}
-
 // onReaderAPIDescribe implements reader.
 func (m *hlsMuxer) onReaderAPIDescribe() interface{} {
 	return struct {

internal/core/hls_source.go

@@ -11,7 +11,6 @@ import (
 
 	"github.com/aler9/rtsp-simple-server/internal/hls"
 	"github.com/aler9/rtsp-simple-server/internal/logger"
-	"github.com/aler9/rtsp-simple-server/internal/rtcpsenderset"
 )
 
 const (
@@ -90,7 +89,6 @@ outer:
 
 func (s *hlsSource) runInner() bool {
 	var stream *stream
-	var rtcpSenders *rtcpsenderset.RTCPSenderSet
 	var videoTrackID int
 	var audioTrackID int
 	var videoEnc *rtph264.Encoder
@@ -99,7 +97,6 @@ func (s *hlsSource) runInner() bool {
 	defer func() {
 		if stream != nil {
 			s.parent.onSourceStaticSetNotReady(pathSourceStaticSetNotReadyReq{source: s})
-			rtcpSenders.Close()
 		}
 	}()
 
@@ -134,7 +131,6 @@ func (s *hlsSource) runInner() bool {
 		s.Log(logger.Info, "ready")
 
 		stream = res.stream
-		rtcpSenders = rtcpsenderset.New(tracks, stream.writePacketRTCP)
 
 		return nil
 	}
@@ -150,7 +146,6 @@ func (s *hlsSource) runInner() bool {
 		}
 
 		for _, pkt := range pkts {
-			rtcpSenders.OnPacketRTP(videoTrackID, pkt)
 			stream.writePacketRTP(videoTrackID, pkt)
 		}
 	}
@@ -166,7 +161,6 @@ func (s *hlsSource) runInner() bool {
 		}
 
 		for _, pkt := range pkts {
-			rtcpSenders.OnPacketRTP(audioTrackID, pkt)
 			stream.writePacketRTP(audioTrackID, pkt)
 		}
 	}

internal/core/reader.go

@@ -1,7 +1,6 @@
 package core
 
 import (
-	"github.com/pion/rtcp"
 	"github.com/pion/rtp/v2"
 )
 
@@ -10,6 +9,5 @@ type reader interface {
 	close()
 	onReaderAccepted()
 	onReaderPacketRTP(int, *rtp.Packet)
-	onReaderPacketRTCP(int, rtcp.Packet)
 	onReaderAPIDescribe() interface{}
 }

internal/core/rtmp_conn.go

@@ -16,13 +16,11 @@ import (
 	"github.com/aler9/gortsplib/pkg/rtpaac"
 	"github.com/aler9/gortsplib/pkg/rtph264"
 	"github.com/notedit/rtmp/av"
-	"github.com/pion/rtcp"
 	"github.com/pion/rtp/v2"
 
 	"github.com/aler9/rtsp-simple-server/internal/conf"
 	"github.com/aler9/rtsp-simple-server/internal/externalcmd"
 	"github.com/aler9/rtsp-simple-server/internal/logger"
-	"github.com/aler9/rtsp-simple-server/internal/rtcpsenderset"
 	"github.com/aler9/rtsp-simple-server/internal/rtmp"
 )
 
@@ -509,14 +507,6 @@ func (c *rtmpConn) runPublish(ctx context.Context) error {
 		return rres.err
 	}
 
-	rtcpSenders := rtcpsenderset.New(tracks, rres.stream.writePacketRTCP)
-	defer rtcpSenders.Close()
-
-	onPacketRTP := func(trackID int, pkt *rtp.Packet) {
-		rtcpSenders.OnPacketRTP(trackID, pkt)
-		rres.stream.writePacketRTP(trackID, pkt)
-	}
-
 	for {
 		c.conn.SetReadDeadline(time.Now().Add(time.Duration(c.readTimeout)))
 		pkt, err := c.conn.ReadPacket()
@@ -558,7 +548,7 @@ func (c *rtmpConn) runPublish(ctx context.Context) error {
 			}
 
 			for _, pkt := range pkts {
-				onPacketRTP(videoTrackID, pkt)
+				rres.stream.writePacketRTP(videoTrackID, pkt)
 			}
 
 		case av.AAC:
@@ -572,7 +562,7 @@ func (c *rtmpConn) runPublish(ctx context.Context) error {
 			}
 
 			for _, pkt := range pkts {
-				onPacketRTP(audioTrackID, pkt)
+				rres.stream.writePacketRTP(audioTrackID, pkt)
 			}
 		}
 	}
@@ -634,10 +624,6 @@ func (c *rtmpConn) onReaderPacketRTP(trackID int, pkt *rtp.Packet) {
 	c.ringBuffer.Push(rtmpConnTrackIDPayloadPair{trackID, pkt})
 }
 
-// onReaderPacketRTCP implements reader.
-func (c *rtmpConn) onReaderPacketRTCP(trackID int, pkt rtcp.Packet) {
-}
-
 // onReaderAPIDescribe implements reader.
 func (c *rtmpConn) onReaderAPIDescribe() interface{} {
 	return struct {

internal/core/rtmp_source.go

@@ -11,11 +11,9 @@ import (
 	"github.com/aler9/gortsplib/pkg/rtpaac"
 	"github.com/aler9/gortsplib/pkg/rtph264"
 	"github.com/notedit/rtmp/av"
-	"github.com/pion/rtp/v2"
 
 	"github.com/aler9/rtsp-simple-server/internal/conf"
 	"github.com/aler9/rtsp-simple-server/internal/logger"
-	"github.com/aler9/rtsp-simple-server/internal/rtcpsenderset"
 	"github.com/aler9/rtsp-simple-server/internal/rtmp"
 )
 
@@ -166,15 +164,6 @@ func (s *rtmpSource) runInner() bool {
 					defer func() {
 						s.parent.onSourceStaticSetNotReady(pathSourceStaticSetNotReadyReq{source: s})
 					}()
-
-					rtcpSenders := rtcpsenderset.New(tracks, res.stream.writePacketRTCP)
-					defer rtcpSenders.Close()
-
-					onPacketRTP := func(trackID int, pkt *rtp.Packet) {
-						rtcpSenders.OnPacketRTP(trackID, pkt)
-						res.stream.writePacketRTP(trackID, pkt)
-					}
-
 					for {
 						conn.SetReadDeadline(time.Now().Add(time.Duration(s.readTimeout)))
 						pkt, err := conn.ReadPacket()
@@ -211,7 +200,7 @@ func (s *rtmpSource) runInner() bool {
 							}
 
 							for _, pkt := range pkts {
-								onPacketRTP(videoTrackID, pkt)
+								res.stream.writePacketRTP(videoTrackID, pkt)
 							}
 
 						case av.AAC:
@@ -225,7 +214,7 @@ func (s *rtmpSource) runInner() bool {
 							}
 
 							for _, pkt := range pkts {
-								onPacketRTP(audioTrackID, pkt)
+								res.stream.writePacketRTP(audioTrackID, pkt)
 							}
 						}
 					}

internal/core/rtsp_server.go

@@ -394,14 +394,6 @@ func (s *rtspServer) OnPacketRTP(ctx *gortsplib.ServerHandlerOnPacketRTPCtx) {
 	se.onPacketRTP(ctx)
 }
 
-// OnPacketRTCP implements gortsplib.ServerHandlerOnPacket.
-func (s *rtspServer) OnPacketRTCP(ctx *gortsplib.ServerHandlerOnPacketRTCPCtx) {
-	s.mutex.RLock()
-	se := s.sessions[ctx.Session]
-	s.mutex.RUnlock()
-	se.onPacketRTCP(ctx)
-}
-
 // onAPISessionsList is called by api and metrics.
 func (s *rtspServer) onAPISessionsList(req rtspServerAPISessionsListReq) rtspServerAPISessionsListRes {
 	select {

internal/core/rtsp_session.go

@@ -9,7 +9,6 @@ import (
 
 	"github.com/aler9/gortsplib"
 	"github.com/aler9/gortsplib/pkg/base"
-	"github.com/pion/rtcp"
 	"github.com/pion/rtp/v2"
 
 	"github.com/aler9/rtsp-simple-server/internal/conf"
@@ -351,11 +350,6 @@ func (s *rtspSession) onReaderPacketRTP(trackID int, pkt *rtp.Packet) {
 	// packets are routed to the session by gortsplib.ServerStream.
 }
 
-// onReaderPacketRTCP implements reader.
-func (s *rtspSession) onReaderPacketRTCP(trackID int, pkt rtcp.Packet) {
-	// packets are routed to the session by gortsplib.ServerStream.
-}
-
 // onReaderAPIDescribe implements reader.
 func (s *rtspSession) onReaderAPIDescribe() interface{} {
 	var typ string
@@ -408,12 +402,3 @@ func (s *rtspSession) onPacketRTP(ctx *gortsplib.ServerHandlerOnPacketRTPCtx) {
 
 	s.stream.writePacketRTP(ctx.TrackID, ctx.Packet)
 }
-
-// onPacketRTCP is called by rtspServer.
-func (s *rtspSession) onPacketRTCP(ctx *gortsplib.ServerHandlerOnPacketRTCPCtx) {
-	if s.ss.State() != gortsplib.ServerSessionStateRecord {
-		return
-	}
-
-	s.stream.writePacketRTCP(ctx.TrackID, ctx.Packet)
-}

internal/core/rtsp_source.go

@@ -14,7 +14,6 @@ import (
 	"github.com/aler9/gortsplib/pkg/base"
 	"github.com/aler9/gortsplib/pkg/h264"
 	"github.com/aler9/gortsplib/pkg/rtph264"
-	"github.com/pion/rtcp"
 	"github.com/pion/rtp/v2"
 
 	"github.com/aler9/rtsp-simple-server/internal/conf"
@@ -209,10 +208,6 @@ func (s *rtspSource) runInner() bool {
 				res.stream.writePacketRTP(trackID, pkt)
 			}
 
-			c.OnPacketRTCP = func(trackID int, pkt rtcp.Packet) {
-				res.stream.writePacketRTCP(trackID, pkt)
-			}
-
 			_, err = c.Play(nil)
 			if err != nil {
 				return err
@@ -305,15 +300,6 @@ func (s *rtspSource) handleMissingH264Params(c *gortsplib.Client, tracks gortspl
 		}
 	}
 
-	c.OnPacketRTCP = func(trackID int, pkt rtcp.Packet) {
-		streamMutex.RLock()
-		defer streamMutex.RUnlock()
-
-		if stream != nil {
-			stream.writePacketRTCP(trackID, pkt)
-		}
-	}
-
 	_, err := c.Play(nil)
 	if err != nil {
 		return err

internal/core/stream.go

@@ -4,7 +4,6 @@ import (
 	"sync"
 
 	"github.com/aler9/gortsplib"
-	"github.com/pion/rtcp"
 	"github.com/pion/rtp/v2"
 )
 
@@ -46,15 +45,6 @@ func (m *streamNonRTSPReadersMap) forwardPacketRTP(trackID int, pkt *rtp.Packet)
 	}
 }
 
-func (m *streamNonRTSPReadersMap) forwardPacketRTCP(trackID int, pkt rtcp.Packet) {
-	m.mutex.RLock()
-	defer m.mutex.RUnlock()
-
-	for c := range m.ma {
-		c.onReaderPacketRTCP(trackID, pkt)
-	}
-}
-
 type stream struct {
 	nonRTSPReaders *streamNonRTSPReadersMap
 	rtspStream     *gortsplib.ServerStream
@@ -96,11 +86,3 @@ func (s *stream) writePacketRTP(trackID int, pkt *rtp.Packet) {
 	// forward to non-RTSP readers
 	s.nonRTSPReaders.forwardPacketRTP(trackID, pkt)
 }
-
-func (s *stream) writePacketRTCP(trackID int, pkt rtcp.Packet) {
-	// forward to RTSP readers
-	s.rtspStream.WritePacketRTCP(trackID, pkt)
-
-	// forward to non-RTSP readers
-	s.nonRTSPReaders.forwardPacketRTCP(trackID, pkt)
-}

internal/rtcpsenderset/rtcpsenderset.go

@@ -1,58 +0,0 @@
-package rtcpsenderset
-
-import (
-	"time"
-
-	"github.com/aler9/gortsplib"
-	"github.com/aler9/gortsplib/pkg/rtcpsender"
-	"github.com/pion/rtcp"
-	"github.com/pion/rtp/v2"
-)
-
-// RTCPSenderSet is a set of RTCP senders.
-type RTCPSenderSet struct {
-	writePacketRTCP func(int, rtcp.Packet)
-	senders         []*rtcpsender.RTCPSender
-
-	// in
-	terminate chan struct{}
-
-	// out
-	done chan struct{}
-}
-
-// New allocates a RTCPSenderSet.
-func New(
-	tracks gortsplib.Tracks,
-	writePacketRTCP func(int, rtcp.Packet),
-) *RTCPSenderSet {
-	s := &RTCPSenderSet{
-		writePacketRTCP: writePacketRTCP,
-		terminate:       make(chan struct{}),
-		done:            make(chan struct{}),
-	}
-
-	s.senders = make([]*rtcpsender.RTCPSender, len(tracks))
-	for i, track := range tracks {
-		ci := i
-
-		s.senders[i] = rtcpsender.New(10*time.Second,
-			track.ClockRate(), func(pkt rtcp.Packet) {
-				writePacketRTCP(ci, pkt)
-			})
-	}
-
-	return s
-}
-
-// Close closes a RTCPSenderSet.
-func (s *RTCPSenderSet) Close() {
-	for _, sender := range s.senders {
-		sender.Close()
-	}
-}
-
-// OnPacketRTP sends a RTP packet to the senders.
-func (s *RTCPSenderSet) OnPacketRTP(trackID int, pkt *rtp.Packet) {
-	s.senders[trackID].ProcessPacketRTP(time.Now(), pkt)
-}