muxer: isolate components

muxer_part.go

@@ -1,6 +1,7 @@
 package gohlslib
 
 import (
+	"fmt"
 	"io"
 	"time"
 
@@ -9,20 +10,22 @@ import (
 )
 
 type muxerPart struct {
-	stream   *muxerStream
-	segment  *muxerSegmentFMP4
-	startDTS time.Duration
-	prefix   string
-	id       uint64
-	storage  storage.Part
+	segmentMaxSize uint64
+	streamID       string
+	streamTracks   []*muxerTrack
+	segment        *muxerSegmentFMP4
+	startDTS       time.Duration
+	prefix         string
+	id             uint64
+	storage        storage.Part
 
 	path          string
 	isIndependent bool
 	endDTS        time.Duration
 }
 
 func (p *muxerPart) initialize() {
-	p.path = partPath(p.prefix, p.stream.id, p.id)
+	p.path = partPath(p.prefix, p.streamID, p.id)
 }
 
 func (p *muxerPart) reader() (io.ReadCloser, error) {
@@ -38,7 +41,7 @@ func (p *muxerPart) finalize(endDTS time.Duration) error {
 		SequenceNumber: uint32(p.id),
 	}
 
-	for i, track := range p.stream.tracks {
+	for i, track := range p.streamTracks {
 		if track.fmp4Samples != nil {
 			part.Tracks = append(part.Tracks, &fmp4.PartTrack{
 				ID:       1 + i,
@@ -60,7 +63,13 @@ func (p *muxerPart) finalize(endDTS time.Duration) error {
 	return nil
 }
 
-func (p *muxerPart) writeSample(track *muxerTrack, sample *fmp4AugmentedSample) {
+func (p *muxerPart) writeSample(track *muxerTrack, sample *fmp4AugmentedSample) error {
+	size := uint64(len(sample.Payload))
+	if (p.segment.size + size) > p.segmentMaxSize {
+		return fmt.Errorf("reached maximum segment size")
+	}
+	p.segment.size += size
+
 	if track.fmp4Samples == nil {
 		track.fmp4StartDTS = sample.dts
 	}
@@ -70,4 +79,6 @@ func (p *muxerPart) writeSample(track *muxerTrack, sample *fmp4AugmentedSample)
 	}
 
 	track.fmp4Samples = append(track.fmp4Samples, &sample.PartSample)
+
+	return nil
 }

muxer_segment.go

@@ -7,7 +7,6 @@ import (
 )
 
 type muxerSegment interface {
-	initialize() error
 	close()
 	finalize(time.Duration) error
 	getPath() string
@@ -21,10 +20,6 @@ type muxerGap struct {
 	duration time.Duration
 }
 
-func (muxerGap) initialize() error {
-	return nil
-}
-
 func (muxerGap) close() {
 }
 

muxer_segment_fmp4.go

@@ -1,20 +1,16 @@
 package gohlslib
 
 import (
-	"fmt"
 	"io"
 	"time"
 
 	"github.com/bluenviron/gohlslib/v2/pkg/storage"
 )
 
 type muxerSegmentFMP4 struct {
-	variant            MuxerVariant
-	segmentMaxSize     uint64
 	prefix             string
-	nextPartID         uint64
 	storageFactory     storage.Factory
-	stream             *muxerStream
+	streamID           string
 	id                 uint64
 	startNTP           time.Time
 	startDTS           time.Duration
@@ -28,24 +24,14 @@ type muxerSegmentFMP4 struct {
 }
 
 func (s *muxerSegmentFMP4) initialize() error {
-	s.path = segmentPath(s.prefix, s.stream.id, s.id, true)
+	s.path = segmentPath(s.prefix, s.streamID, s.id, true)
 
 	var err error
 	s.storage, err = s.storageFactory.NewFile(s.path)
 	if err != nil {
 		return err
 	}
 
-	s.stream.nextPart = &muxerPart{
-		stream:   s.stream,
-		segment:  s,
-		startDTS: s.startDTS,
-		prefix:   s.prefix,
-		id:       s.nextPartID,
-		storage:  s.storage.NewPart(),
-	}
-	s.stream.nextPart.initialize()
-
 	return nil
 }
 
@@ -80,18 +66,3 @@ func (s *muxerSegmentFMP4) finalize(nextDTS time.Duration) error {
 
 	return nil
 }
-
-func (s *muxerSegmentFMP4) writeSample(
-	track *muxerTrack,
-	sample *fmp4AugmentedSample,
-) error {
-	size := uint64(len(sample.Payload))
-	if (s.size + size) > s.segmentMaxSize {
-		return fmt.Errorf("reached maximum segment size")
-	}
-	s.size += size
-
-	s.stream.nextPart.writeSample(track, sample)
-
-	return nil
-}

muxer_segment_mpegts.go

@@ -7,13 +7,15 @@ import (
 	"time"
 
 	"github.com/bluenviron/gohlslib/v2/pkg/storage"
+	"github.com/bluenviron/mediacommon/pkg/formats/mpegts"
 )
 
 type muxerSegmentMPEGTS struct {
 	segmentMaxSize uint64
 	prefix         string
 	storageFactory storage.Factory
-	stream         *muxerStream
+	streamID       string
+	mpegtsWriter   *mpegts.Writer
 	id             uint64
 	startNTP       time.Time
 	startDTS       time.Duration
@@ -28,7 +30,7 @@ type muxerSegmentMPEGTS struct {
 }
 
 func (s *muxerSegmentMPEGTS) initialize() error {
-	s.path = segmentPath(s.prefix, s.stream.id, s.id, false)
+	s.path = segmentPath(s.prefix, s.streamID, s.id, false)
 
 	var err error
 	s.storage, err = s.storageFactory.NewFile(s.path)
@@ -39,8 +41,6 @@ func (s *muxerSegmentMPEGTS) initialize() error {
 	s.storagePart = s.storage.NewPart()
 	s.bw = bufio.NewWriter(s.storagePart.Writer())
 
-	s.stream.mpegtsSwitchableWriter.w = s.bw
-
 	return nil
 }
 
@@ -96,7 +96,7 @@ func (s *muxerSegmentMPEGTS) writeH264(
 	}
 	s.size += size
 
-	err := s.stream.mpegtsWriter.WriteH264(
+	err := s.mpegtsWriter.WriteH264(
 		track.mpegtsTrack,
 		multiplyAndDivide(pts, 90000, int64(track.ClockRate)),
 		multiplyAndDivide(dts, 90000, int64(track.ClockRate)),
@@ -127,7 +127,7 @@ func (s *muxerSegmentMPEGTS) writeMPEG4Audio(
 	}
 	s.size += size
 
-	err := s.stream.mpegtsWriter.WriteMPEG4Audio(
+	err := s.mpegtsWriter.WriteMPEG4Audio(
 		track.mpegtsTrack,
 		multiplyAndDivide(pts, 90000, int64(track.ClockRate)),
 		aus,

muxer_segmenter.go

@@ -77,8 +77,17 @@ type fmp4AugmentedSample struct {
 	ntp time.Time
 }
 
+type muxerSegmenterParent interface {
+	createFirstSegment(nextDTS time.Duration, nextNTP time.Time) error
+	rotateSegments(nextDTS time.Duration, nextNTP time.Time, force bool) error
+	rotateParts(nextDTS time.Duration) error
+}
+
 type muxerSegmenter struct {
-	muxer *Muxer // TODO: remove
+	variant            MuxerVariant
+	segmentMinDuration time.Duration
+	partMinDuration    time.Duration
+	parent             muxerSegmenterParent
 
 	pendingParamsChange            bool
 	fmp4SampleDurations            map[time.Duration]struct{} // low-latency only
@@ -87,7 +96,7 @@ type muxerSegmenter struct {
 }
 
 func (s *muxerSegmenter) initialize() {
-	if s.muxer.Variant != MuxerVariantMPEGTS {
+	if s.variant != MuxerVariantMPEGTS {
 		s.fmp4SampleDurations = make(map[time.Duration]struct{})
 	}
 }
@@ -349,17 +358,17 @@ func (s *muxerSegmenter) writeH264(
 		return fmt.Errorf("unable to extract DTS: %w", err)
 	}
 
-	if s.muxer.Variant == MuxerVariantMPEGTS {
+	if s.variant == MuxerVariantMPEGTS {
 		if track.stream.nextSegment == nil {
-			err = s.muxer.createFirstSegment(timestampToDuration(dts, track.ClockRate), ntp)
+			err = s.parent.createFirstSegment(timestampToDuration(dts, track.ClockRate), ntp)
 			if err != nil {
 				return err
 			}
 		} else if randomAccess && // switch segment
 			((timestampToDuration(dts, track.ClockRate)-
-				track.stream.nextSegment.(*muxerSegmentMPEGTS).startDTS) >= s.muxer.SegmentMinDuration ||
+				track.stream.nextSegment.(*muxerSegmentMPEGTS).startDTS) >= s.segmentMinDuration ||
 				paramsChanged) {
-			err = s.muxer.rotateSegments(timestampToDuration(dts, track.ClockRate), ntp, false)
+			err = s.parent.rotateSegments(timestampToDuration(dts, track.ClockRate), ntp, false)
 			if err != nil {
 				return err
 			}
@@ -435,17 +444,17 @@ func (s *muxerSegmenter) writeMPEG4Audio(
 	pts int64,
 	aus [][]byte,
 ) error {
-	if s.muxer.Variant == MuxerVariantMPEGTS {
+	if s.variant == MuxerVariantMPEGTS {
 		if track.isLeading {
 			if track.stream.nextSegment == nil {
-				err := s.muxer.createFirstSegment(timestampToDuration(pts, track.ClockRate), ntp)
+				err := s.parent.createFirstSegment(timestampToDuration(pts, track.ClockRate), ntp)
 				if err != nil {
 					return err
 				}
 			} else if track.stream.nextSegment.(*muxerSegmentMPEGTS).audioAUCount >= mpegtsSegmentMinAUCount && // switch segment
 				(timestampToDuration(pts, track.ClockRate)-
-					track.stream.nextSegment.(*muxerSegmentMPEGTS).startDTS) >= s.muxer.SegmentMinDuration {
-				err := s.muxer.rotateSegments(timestampToDuration(pts, track.ClockRate), ntp, false)
+					track.stream.nextSegment.(*muxerSegmentMPEGTS).startDTS) >= s.segmentMinDuration {
+				err := s.parent.rotateSegments(timestampToDuration(pts, track.ClockRate), ntp, false)
 				if err != nil {
 					return err
 				}
@@ -496,7 +505,7 @@ func (s *muxerSegmenter) writeMPEG4Audio(
 // iPhone iOS fails if part durations are less than 85% of maximum part duration.
 // find a part duration that is compatible with all sample durations
 func (s *muxerSegmenter) fmp4AdjustPartDuration(sampleDuration time.Duration) {
-	if s.muxer.Variant != MuxerVariantLowLatency || s.fmp4FreezeAdjustedPartDuration {
+	if s.variant != MuxerVariantLowLatency || s.fmp4FreezeAdjustedPartDuration {
 		return
 	}
 
@@ -508,7 +517,7 @@ func (s *muxerSegmenter) fmp4AdjustPartDuration(sampleDuration time.Duration) {
 	if _, ok := s.fmp4SampleDurations[sampleDuration]; !ok {
 		s.fmp4SampleDurations[sampleDuration] = struct{}{}
 		s.fmp4AdjustedPartDuration = findCompatiblePartDuration(
-			s.muxer.PartMinDuration,
+			s.partMinDuration,
 			s.fmp4SampleDurations,
 		)
 	}
@@ -539,7 +548,7 @@ func (s *muxerSegmenter) fmp4WriteSample(
 	if track.isLeading {
 		// create first segment
 		if track.stream.nextSegment == nil {
-			err := s.muxer.createFirstSegment(timestampToDuration(sample.dts, track.ClockRate), sample.ntp)
+			err := s.parent.createFirstSegment(timestampToDuration(sample.dts, track.ClockRate), sample.ntp)
 			if err != nil {
 				return err
 			}
@@ -555,7 +564,7 @@ func (s *muxerSegmenter) fmp4WriteSample(
 		s.fmp4AdjustPartDuration(timestampToDuration(duration, track.ClockRate))
 	}
 
-	err := track.stream.nextSegment.(*muxerSegmentFMP4).writeSample(
+	err := track.stream.nextPart.writeSample(
 		track,
 		sample,
 	)
@@ -567,8 +576,8 @@ func (s *muxerSegmenter) fmp4WriteSample(
 		// switch segment
 		if randomAccess && (paramsChanged ||
 			(timestampToDuration(track.fmp4NextSample.dts, track.ClockRate)-
-				track.stream.nextSegment.(*muxerSegmentFMP4).startDTS) >= s.muxer.SegmentMinDuration) {
-			err = s.muxer.rotateSegments(timestampToDuration(track.fmp4NextSample.dts, track.ClockRate),
+				track.stream.nextSegment.(*muxerSegmentFMP4).startDTS) >= s.segmentMinDuration) {
+			err = s.parent.rotateSegments(timestampToDuration(track.fmp4NextSample.dts, track.ClockRate),
 				track.fmp4NextSample.ntp, paramsChanged)
 			if err != nil {
 				return err
@@ -583,10 +592,10 @@ func (s *muxerSegmenter) fmp4WriteSample(
 			}
 
 			// switch part
-		} else if (s.muxer.Variant == MuxerVariantLowLatency) &&
+		} else if (s.variant == MuxerVariantLowLatency) &&
 			(timestampToDuration(track.fmp4NextSample.dts, track.ClockRate)-
 				track.stream.nextPart.startDTS) >= s.fmp4AdjustedPartDuration {
-			err := s.muxer.rotateParts(timestampToDuration(track.fmp4NextSample.dts, track.ClockRate))
+			err := s.parent.rotateParts(timestampToDuration(track.fmp4NextSample.dts, track.ClockRate))
 			if err != nil {
 				return err
 			}