Adding an option to configure max acceptable RTP packet PTS time difference before adjusting them (#716)

* Reducing max acceptable ts diff

* Adding abiliy to configure max acceptable time difference

Author: milos-lk

pkg/synchronizer/synchronizer.go

@@ -21,25 +21,78 @@ import (
 	"github.com/pion/rtcp"
 )
 
+const (
+	DefaultMaxTsDiff = time.Minute
+)
+
+type SynchronizerOption func(*SynchronizerConfig)
+
+// SynchronizerConfig holds configuration for the Synchronizer
+type SynchronizerConfig struct {
+	MaxTsDiff time.Duration
+	OnStarted func()
+}
+
+// WithMaxTsDiff sets the maximum acceptable difference between RTP packets
+// In case exceeded the synchronizer will adjust the PTS offset to keep the audio and video in sync
+func WithMaxTsDiff(maxTsDiff time.Duration) SynchronizerOption {
+	return func(config *SynchronizerConfig) {
+		config.MaxTsDiff = maxTsDiff
+	}
+}
+
+// WithOnStarted sets the callback to be called when the synchronizer starts
+func WithOnStarted(onStarted func()) SynchronizerOption {
+	return func(config *SynchronizerConfig) {
+		config.OnStarted = onStarted
+	}
+}
+
 // a single Synchronizer is shared between all audio and video writers
 type Synchronizer struct {
 	sync.RWMutex
 
 	startedAt int64
 	onStarted func()
 	endedAt   int64
+	config    SynchronizerConfig
 
 	psByIdentity map[string]*participantSynchronizer
 	psBySSRC     map[uint32]*participantSynchronizer
 	ssrcByID     map[string]uint32
 }
 
 func NewSynchronizer(onStarted func()) *Synchronizer {
+	config := SynchronizerConfig{
+		MaxTsDiff: DefaultMaxTsDiff,
+		OnStarted: onStarted,
+	}
+
+	return &Synchronizer{
+		onStarted:    config.OnStarted,
+		psByIdentity: make(map[string]*participantSynchronizer),
+		psBySSRC:     make(map[uint32]*participantSynchronizer),
+		ssrcByID:     make(map[string]uint32),
+		config:       config,
+	}
+}
+
+func NewSynchronizerWithOptions(opts ...SynchronizerOption) *Synchronizer {
+	config := SynchronizerConfig{
+		MaxTsDiff: DefaultMaxTsDiff,
+		OnStarted: nil,
+	}
+
+	for _, opt := range opts {
+		opt(&config)
+	}
+
 	return &Synchronizer{
-		onStarted:    onStarted,
+		onStarted:    config.OnStarted,
 		psByIdentity: make(map[string]*participantSynchronizer),
 		psBySSRC:     make(map[uint32]*participantSynchronizer),
 		ssrcByID:     make(map[string]uint32),
+		config:       config,
 	}
 }
 

pkg/synchronizer/track.go

@@ -29,7 +29,6 @@ import (
 
 const (
 	maxAdjustment = time.Millisecond * 5
-	maxTSDiff     = time.Minute
 )
 
 type TrackRemote interface {
@@ -46,6 +45,9 @@ type TrackSynchronizer struct {
 	logger logger.Logger
 	*rtpConverter
 
+	// config
+	maxTsDiff time.Duration // maximum acceptable difference between RTP packets
+
 	// timing info
 	startTime       time.Time     // time first packet was pushed
 	startRTP        uint32        // RTP timestamp of PTS 0
@@ -69,6 +71,7 @@ func newTrackSynchronizer(s *Synchronizer, track TrackRemote) *TrackSynchronizer
 		track:        track,
 		logger:       logger.GetLogger().WithValues("trackID", track.ID(), "codec", track.Codec().MimeType),
 		rtpConverter: newRTPConverter(int64(track.Codec().ClockRate)),
+		maxTsDiff:    s.config.MaxTsDiff,
 	}
 
 	return t
@@ -115,7 +118,7 @@ func (t *TrackSynchronizer) GetPTS(pkt *rtp.Packet) (time.Duration, error) {
 
 	pts := t.lastPTS + t.toDuration(ts-t.lastTS)
 	estimatedPTS := time.Since(t.startTime)
-	if pts < t.lastPTS || !acceptable(pts-estimatedPTS) {
+	if pts < t.lastPTS || !t.acceptable(pts-estimatedPTS) {
 		pts = estimatedPTS
 		t.startRTP = ts - t.toRTP(pts)
 	}
@@ -155,7 +158,7 @@ func (t *TrackSynchronizer) onSenderReport(pkt *rtcp.SenderReport) {
 	} else {
 		pts = t.lastPTS - t.toDuration(t.lastTS-pkt.RTPTime)
 	}
-	if !acceptable(pts - time.Since(t.startTime)) {
+	if !t.acceptable(pts - time.Since(t.startTime)) {
 		return
 	}
 
@@ -164,16 +167,16 @@ func (t *TrackSynchronizer) onSenderReport(pkt *rtcp.SenderReport) {
 		t.onSR(offset - t.desiredPTSOffset)
 	}
 
-	if !acceptable(offset) {
+	if !t.acceptable(offset) {
 		return
 	}
 
 	t.desiredPTSOffset = offset
 	t.lastSR = pkt.RTPTime
 }
 
-func acceptable(d time.Duration) bool {
-	return d > -maxTSDiff && d < maxTSDiff
+func (t *TrackSynchronizer) acceptable(d time.Duration) bool {
+	return d > -t.maxTsDiff && d < t.maxTsDiff
 }
 
 type rtpConverter struct {