update mediacommon; replace Client.OnData with OnDataH26x, OnDataMPEG…

…4Audio, OnDataOpus

README.md

@@ -51,7 +51,7 @@ General features:
 * [Examples](#examples)
 * [API Documentation](#api-documentation)
 * [Standards](#standards)
-* [Links](#links)
+* [Related projects](#related-projects)
 
 ## Examples
 
@@ -72,8 +72,8 @@ https://pkg.go.dev/github.com/bluenviron/gohlslib#pkg-index
 * [Codec standards](https://github.com/bluenviron/mediacommon#standards)
 * [Golang project layout](https://github.com/golang-standards/project-layout)
 
-## Links
-
-Related projects
+## Related projects
 
 * [MediaMTX](https://github.com/bluenviron/mediamtx)
+* [gortsplib](https://github.com/bluenviron/gortsplib)
+* [mediacommon](https://github.com/bluenviron/mediacommon)

client.go

@@ -21,6 +21,12 @@ const (
 	clientMaxDTSRTCDiff               = 10 * time.Second
 )
 
+type onDataH26xFunc func(pts time.Duration, dts time.Duration, au [][]byte)
+
+type onDataMPEG4AudioFunc func(pts time.Duration, dts time.Duration, aus [][]byte)
+
+type onDataOpusFunc func(pts time.Duration, dts time.Duration, packets [][]byte)
+
 func clientAbsoluteURL(base *url.URL, relative string) (*url.URL, error) {
 	u, err := url.Parse(relative)
 	if err != nil {
@@ -50,7 +56,7 @@ type Client struct {
 	ctx         context.Context
 	ctxCancel   func()
 	onTracks    func([]*Track) error
-	onData      map[*Track]func(time.Duration, interface{})
+	onData      map[*Track]interface{}
 	playlistURL *url.URL
 
 	// out
@@ -74,7 +80,7 @@ func (c *Client) Start() error {
 
 	c.ctx, c.ctxCancel = context.WithCancel(context.Background())
 
-	c.onData = make(map[*Track]func(time.Duration, interface{}))
+	c.onData = make(map[*Track]interface{})
 	c.outErr = make(chan error, 1)
 
 	go c.run()
@@ -97,9 +103,19 @@ func (c *Client) OnTracks(cb func([]*Track) error) {
 	c.onTracks = cb
 }
 
-// OnData sets a callback that is called when data arrives.
-func (c *Client) OnData(forma *Track, cb func(time.Duration, interface{})) {
-	c.onData[forma] = cb
+// OnDataH26x sets a callback that is called when data from an H26x track is received.
+func (c *Client) OnDataH26x(forma *Track, onData onDataH26xFunc) {
+	c.onData[forma] = onData
+}
+
+// OnDataMPEG4Audio sets a callback that is called when data from a MPEG-4 Audio track is received.
+func (c *Client) OnDataMPEG4Audio(forma *Track, onData onDataMPEG4AudioFunc) {
+	c.onData[forma] = onData
+}
+
+// OnDataOpus sets a callback that is called when data from an Opus track is received.
+func (c *Client) OnDataOpus(forma *Track, onData onDataOpusFunc) {
+	c.onData[forma] = onData
 }
 
 func (c *Client) run() {

client_downloader_primary.go

@@ -7,7 +7,6 @@ import (
 	"net/http"
 	"net/url"
 	"strings"
-	"time"
 
 	"github.com/bluenviron/gohlslib/pkg/playlist"
 )
@@ -104,7 +103,7 @@ type clientDownloaderPrimary struct {
 	httpClient         *http.Client
 	log                LogFunc
 	onTracks           func([]*Track) error
-	onData             map[*Track]func(time.Duration, interface{})
+	onData             map[*Track]interface{}
 	rp                 *clientRoutinePool
 
 	leadingTimeSync clientTimeSync
@@ -123,7 +122,7 @@ func newClientDownloaderPrimary(
 	log LogFunc,
 	rp *clientRoutinePool,
 	onTracks func([]*Track) error,
-	onData map[*Track]func(time.Duration, interface{}),
+	onData map[*Track]interface{},
 ) *clientDownloaderPrimary {
 	return &clientDownloaderPrimary{
 		primaryPlaylistURL:   primaryPlaylistURL,

client_downloader_stream.go

@@ -7,7 +7,6 @@ import (
 	"net/http"
 	"net/url"
 	"strconv"
-	"time"
 
 	"github.com/bluenviron/gohlslib/pkg/playlist"
 )
@@ -40,7 +39,7 @@ type clientDownloaderStream struct {
 	onStreamTracks       func(context.Context, []*Track) bool
 	onSetLeadingTimeSync func(clientTimeSync)
 	onGetLeadingTimeSync func(context.Context) (clientTimeSync, bool)
-	onData               map[*Track]func(time.Duration, interface{})
+	onData               map[*Track]interface{}
 
 	curSegmentID *int
 }
@@ -55,7 +54,7 @@ func newClientDownloaderStream(
 	onStreamTracks func(context.Context, []*Track) bool,
 	onSetLeadingTimeSync func(clientTimeSync),
 	onGetLeadingTimeSync func(context.Context) (clientTimeSync, bool),
-	onData map[*Track]func(time.Duration, interface{}),
+	onData map[*Track]interface{},
 ) *clientDownloaderStream {
 	return &clientDownloaderStream{
 		isLeading:            isLeading,

client_processor_fmp4.go

@@ -31,12 +31,12 @@ type clientProcessorFMP4 struct {
 	rp                   *clientRoutinePool
 	onSetLeadingTimeSync func(clientTimeSync)
 	onGetLeadingTimeSync func(context.Context) (clientTimeSync, bool)
-	onData               map[*Track]func(time.Duration, interface{})
+	onData               map[*Track]interface{}
 
-	tracks         []*Track
-	init           fmp4.Init
-	leadingTrackID int
-	trackProcs     map[int]*clientProcessorFMP4Track
+	tracks             []*Track
+	init               fmp4.Init
+	leadingTrackID     int
+	prePreProcessFuncs map[int]func(context.Context, *fmp4.PartTrack) error
 
 	// in
 	subpartProcessed chan struct{}
@@ -52,7 +52,7 @@ func newClientProcessorFMP4(
 	onStreamTracks func(context.Context, []*Track) bool,
 	onSetLeadingTimeSync func(clientTimeSync),
 	onGetLeadingTimeSync func(context.Context) (clientTimeSync, bool),
-	onData map[*Track]func(time.Duration, interface{}),
+	onData map[*Track]interface{},
 ) (*clientProcessorFMP4, error) {
 	p := &clientProcessorFMP4{
 		isLeading:            isLeading,
@@ -111,41 +111,16 @@ func (p *clientProcessorFMP4) processSegment(ctx context.Context, byts []byte) e
 	processingCount := 0
 
 	for _, part := range parts {
-		for _, track := range part.Tracks {
-			if p.trackProcs == nil {
-				var ts *clientTimeSyncFMP4
-
-				if p.isLeading {
-					if track.ID != p.leadingTrackID {
-						continue
-					}
-
-					timeScale := func() uint32 {
-						for _, track := range p.init.Tracks {
-							if track.ID == p.leadingTrackID {
-								return track.TimeScale
-							}
-						}
-						return 0
-					}()
-					ts = newClientTimeSyncFMP4(timeScale, track.BaseTime)
-					p.onSetLeadingTimeSync(ts)
-				} else {
-					rawTS, ok := p.onGetLeadingTimeSync(ctx)
-					if !ok {
-						return fmt.Errorf("terminated")
-					}
-
-					ts, ok = rawTS.(*clientTimeSyncFMP4)
-					if !ok {
-						return fmt.Errorf("stream playlists are mixed MPEGTS/FMP4")
-					}
+		for _, partTrack := range part.Tracks {
+			err := p.initializeTrackProcs(ctx, partTrack)
+			if err != nil {
+				if err == errSkipSilently {
+					continue
 				}
-
-				p.initializeTrackProcs(ts)
+				return err
 			}
 
-			proc, ok := p.trackProcs[track.ID]
+			prePreProcess, ok := p.prePreProcessFuncs[partTrack.ID]
 			if !ok {
 				continue
 			}
@@ -154,11 +129,11 @@ func (p *clientProcessorFMP4) processSegment(ctx context.Context, byts []byte) e
 				return fmt.Errorf("too many part tracks at once")
 			}
 
-			select {
-			case proc.queue <- track:
-			case <-ctx.Done():
-				return fmt.Errorf("terminated")
+			err = prePreProcess(ctx, partTrack)
+			if err != nil {
+				return err
 			}
+
 			processingCount++
 		}
 	}
@@ -181,44 +156,127 @@ func (p *clientProcessorFMP4) onPartTrackProcessed(ctx context.Context) {
 	}
 }
 
-func (p *clientProcessorFMP4) initializeTrackProcs(ts *clientTimeSyncFMP4) {
-	p.trackProcs = make(map[int]*clientProcessorFMP4Track)
+func (p *clientProcessorFMP4) initializeTrackProcs(ctx context.Context, track *fmp4.PartTrack) error {
+	if p.prePreProcessFuncs != nil {
+		return nil
+	}
 
-	for i, track := range p.tracks {
-		var cb func(time.Duration, []byte) error
+	var timeSync *clientTimeSyncFMP4
+	isLeadingTrack := (track.ID == p.leadingTrackID)
 
-		cb2, ok := p.onData[track]
-		if !ok {
-			cb2 = func(time.Duration, interface{}) {
+	if p.isLeading {
+		if !isLeadingTrack {
+			return errSkipSilently
+		}
+
+		timeScale := func() uint32 {
+			for _, track := range p.init.Tracks {
+				if isLeadingTrack {
+					return track.TimeScale
+				}
 			}
+			return 0
+		}()
+		timeSync = newClientTimeSyncFMP4(timeScale, track.BaseTime)
+		p.onSetLeadingTimeSync(timeSync)
+	} else {
+		rawTS, ok := p.onGetLeadingTimeSync(ctx)
+		if !ok {
+			return fmt.Errorf("terminated")
+		}
+
+		timeSync, ok = rawTS.(*clientTimeSyncFMP4)
+		if !ok {
+			return fmt.Errorf("stream playlists are mixed MPEG-TS/fMP4")
 		}
+	}
+
+	p.prePreProcessFuncs = make(map[int]func(context.Context, *fmp4.PartTrack) error)
+
+	for i, track := range p.tracks {
+		onData := p.onData[track]
+
+		var postProcess func(pts time.Duration, dts time.Duration, payload []byte) error
 
 		switch track.Codec.(type) {
 		case *codecs.H264, *codecs.H265:
-			cb = func(pts time.Duration, payload []byte) error {
-				nalus, err := h264.AVCCUnmarshal(payload)
+			var onDataCasted onDataH26xFunc = func(pts time.Duration, dts time.Duration, au [][]byte) {}
+			if onData != nil {
+				onDataCasted = onData.(onDataH26xFunc)
+			}
+
+			postProcess = func(pts time.Duration, dts time.Duration, payload []byte) error {
+				au, err := h264.AVCCUnmarshal(payload)
 				if err != nil {
 					return err
 				}
 
-				cb2(pts, nalus)
+				onDataCasted(pts, dts, au)
+				return nil
+			}
+
+		case *codecs.MPEG4Audio:
+			var onDataCasted onDataMPEG4AudioFunc = func(pts time.Duration, dts time.Duration, aus [][]byte) {}
+			if onData != nil {
+				onDataCasted = onData.(onDataMPEG4AudioFunc)
+			}
+
+			postProcess = func(pts time.Duration, dts time.Duration, payload []byte) error {
+				onDataCasted(pts, dts, [][]byte{payload})
 				return nil
 			}
 
-		case *codecs.MPEG4Audio, *codecs.Opus:
-			cb = func(pts time.Duration, payload []byte) error {
-				cb2(pts, payload)
+		case *codecs.Opus:
+			var onDataCasted onDataOpusFunc = func(pts time.Duration, dts time.Duration, packets [][]byte) {}
+			if onData != nil {
+				onDataCasted = onData.(onDataOpusFunc)
+			}
+
+			postProcess = func(pts time.Duration, dts time.Duration, payload []byte) error {
+				onDataCasted(pts, dts, [][]byte{payload})
 				return nil
 			}
 		}
 
-		proc := newClientProcessorFMP4Track(
-			p.init.Tracks[i].TimeScale,
-			ts,
-			p.onPartTrackProcessed,
-			cb,
-		)
-		p.rp.add(proc)
-		p.trackProcs[p.init.Tracks[i].ID] = proc
+		timeScale := p.init.Tracks[i].TimeScale
+
+		preProcess := func(ctx context.Context, partTrack *fmp4.PartTrack) error {
+			rawDTS := partTrack.BaseTime
+
+			for _, sample := range partTrack.Samples {
+				pts, dts, err := timeSync.convertAndSync(ctx, timeScale, rawDTS, sample.PTSOffset)
+				if err != nil {
+					return err
+				}
+
+				rawDTS += uint64(sample.Duration)
+
+				// silently discard packets prior to the first packet of the leading track
+				if pts < 0 {
+					continue
+				}
+
+				err = postProcess(pts, dts, sample.Payload)
+				if err != nil {
+					return err
+				}
+			}
+
+			p.onPartTrackProcessed(ctx)
+			return nil
+		}
+
+		trackProc := newClientTrackProcessor()
+		p.rp.add(trackProc)
+
+		prePreProcess := func(ctx context.Context, partTrack *fmp4.PartTrack) error {
+			return trackProc.push(ctx, func() error {
+				return preProcess(ctx, partTrack)
+			})
+		}
+
+		p.prePreProcessFuncs[p.init.Tracks[i].ID] = prePreProcess
 	}
+
+	return nil
 }