update dependencies (#2113)

README.md

@@ -639,15 +639,23 @@ videotestsrc ! video/x-raw,width=1280,height=720 ! x264enc speed-preset=ultrafas
 audiotestsrc ! audioconvert ! avenc_aac ! mux.
 ```
 
+or FFmpeg:
+
+```
+ffmpeg -re -f lavfi -i testsrc=size=1280x720:rate=30 \
+-pix_fmt yuv420p -c:v libx264 -preset ultrafast -b:v 600k \
+-f mpegts udp://238.0.0.1:1234?pkt_size=1316
+```
+
 Edit `mediamtx.yml` and replace everything inside section `paths` with the following content:
 
 ```yml
 paths:
-  udp:
+  mypath:
     source: udp://238.0.0.1:1234
 ```
 
-The resulting stream will be available in path `/udp`.
+The resulting stream will be available in path `/mypath`.
 
 ## Read from the server
 

go.mod

@@ -7,9 +7,9 @@ require (
 	github.com/abema/go-mp4 v0.11.0
 	github.com/alecthomas/kong v0.8.0
 	github.com/asticode/go-astits v1.11.0
-	github.com/bluenviron/gohlslib v0.2.7
+	github.com/bluenviron/gohlslib v0.3.0
-	github.com/bluenviron/gortsplib/v3 v3.8.2
+	github.com/bluenviron/gortsplib/v3 v3.9.0
-	github.com/bluenviron/mediacommon v0.6.0
+	github.com/bluenviron/mediacommon v0.7.0
 	github.com/fsnotify/fsnotify v1.6.0
 	github.com/gin-gonic/gin v1.9.1
 	github.com/google/uuid v1.3.0

go.sum

@@ -10,12 +10,12 @@ github.com/asticode/go-astikit v0.30.0 h1:DkBkRQRIxYcknlaU7W7ksNfn4gMFsB0tqMJflx
 github.com/asticode/go-astikit v0.30.0/go.mod h1:h4ly7idim1tNhaVkdVBeXQZEE3L0xblP7fCWbgwipF0=
 github.com/asticode/go-astits v1.11.0 h1:GTHUXht0ZXAJXsVbsLIcyfHr1Bchi4QQwMARw2ZWAng=
 github.com/asticode/go-astits v1.11.0/go.mod h1:QSHmknZ51pf6KJdHKZHJTLlMegIrhega3LPWz3ND/iI=
-github.com/bluenviron/gohlslib v0.2.7 h1:wV4ky3PbozwWORWd9pURktMnPIT7j/mBK3ivVljgVkk=
+github.com/bluenviron/gohlslib v0.3.0 h1:ze8cCKszGC2LAWp0B+qIXZIlCZocB7a3BKeBo9E8Sr0=
-github.com/bluenviron/gohlslib v0.2.7/go.mod h1:x5XU60uvx91bQkRI8lrKNEltUtudR0WzaONBOhH+nzQ=
+github.com/bluenviron/gohlslib v0.3.0/go.mod h1:aO69Vu0mMUxWrLmgS6g/S3Y3sfAhyg2SXaMEL7yNlWc=
-github.com/bluenviron/gortsplib/v3 v3.8.2 h1:KSZNOeWkOgotISiJIb38MRIAZGA3gNNN/fqVEwlLg/U=
+github.com/bluenviron/gortsplib/v3 v3.9.0 h1:aAHV6MhsDtgBF6yKaNBBCdvtSpLB8ne4kyUfLQlN7nM=
-github.com/bluenviron/gortsplib/v3 v3.8.2/go.mod h1:hGrlEGPJlMbXonsnHtvSYJBxyYHboxjAULEEr/5Z//Y=
+github.com/bluenviron/gortsplib/v3 v3.9.0/go.mod h1:5h3Zu7jkzwDknYrf+89q2saab//oioKgM9mgvBEX3pg=
-github.com/bluenviron/mediacommon v0.6.0 h1:suWFWHL9WL+sfBQPmleCd5jCY0iEtuKgvPRUaBGoq+g=
+github.com/bluenviron/mediacommon v0.7.0 h1:dJWLLL9oDbAqfK8KuNfnDUQwNbeMAtGeRjZc9Vo95js=
-github.com/bluenviron/mediacommon v0.6.0/go.mod h1:wuLJdxcITiSPgY1MvQqrX+qPlKmNfeV9wNvXth5M98I=
+github.com/bluenviron/mediacommon v0.7.0/go.mod h1:wuLJdxcITiSPgY1MvQqrX+qPlKmNfeV9wNvXth5M98I=
 github.com/bytedance/sonic v1.5.0/go.mod h1:ED5hyg4y6t3/9Ku1R6dU/4KyJ48DZ4jPhfY1O2AihPM=
 github.com/bytedance/sonic v1.9.1 h1:6iJ6NqdoxCDr6mbY8h18oSO+cShGSMRGCEo7F2h0x8s=
 github.com/bytedance/sonic v1.9.1/go.mod h1:i736AoUSYt75HyZLoJW9ERYxcy6eaN6h4BZXU064P/U=

internal/core/hls_muxer.go

@@ -16,7 +16,6 @@ import (
 	"github.com/bluenviron/gortsplib/v3/pkg/formats"
 	"github.com/bluenviron/gortsplib/v3/pkg/media"
 	"github.com/bluenviron/gortsplib/v3/pkg/ringbuffer"
-	"github.com/bluenviron/mediacommon/pkg/codecs/mpeg4audio"
 	"github.com/gin-gonic/gin"
 
 	"github.com/bluenviron/mediamtx/internal/conf"
@@ -442,16 +441,13 @@ func (m *hlsMuxer) createAudioTrack(stream *stream) (*media.Media, *gohlslib.Tra
 				}
 				pts := tunit.PTS - audioStartPTS
 
-				for i, au := range tunit.AUs {
+				err := m.muxer.WriteMPEG4Audio(
-					err := m.muxer.WriteAudio(
 					tunit.NTP,
-						pts+time.Duration(i)*mpeg4audio.SamplesPerAccessUnit*
+					pts,
-							time.Second/time.Duration(audioFormatMPEG4AudioGeneric.ClockRate()),
+					tunit.AUs)
-						au)
 				if err != nil {
 					return fmt.Errorf("muxer error: %v", err)
 				}
-				}
 
 				return nil
 			})
@@ -488,10 +484,10 @@ func (m *hlsMuxer) createAudioTrack(stream *stream) (*media.Media, *gohlslib.Tra
 				}
 				pts := tunit.PTS - audioStartPTS
 
-				err := m.muxer.WriteAudio(
+				err := m.muxer.WriteMPEG4Audio(
 					tunit.NTP,
 					pts,
-					tunit.AU)
+					[][]byte{tunit.AU})
 				if err != nil {
 					return fmt.Errorf("muxer error: %v", err)
 				}
@@ -524,10 +520,10 @@ func (m *hlsMuxer) createAudioTrack(stream *stream) (*media.Media, *gohlslib.Tra
 				}
 				pts := tunit.PTS - audioStartPTS
 
-				err := m.muxer.WriteAudio(
+				err := m.muxer.WriteOpus(
 					tunit.NTP,
 					pts,
-					tunit.Frame)
+					tunit.Packets)
 				if err != nil {
 					return fmt.Errorf("muxer error: %v", err)
 				}

internal/core/hls_source.go

@@ -77,10 +77,10 @@ func (s *hlsSource) run(ctx context.Context, cnf *conf.PathConf, reloadConf chan
 					}},
 				}
 
-				c.OnData(track, func(pts time.Duration, unit interface{}) {
+				c.OnDataH26x(track, func(pts time.Duration, dts time.Duration, au [][]byte) {
 					stream.writeUnit(medi, medi.Formats[0], &formatprocessor.UnitH264{
 						PTS: pts,
-						AU:  unit.([][]byte),
+						AU:  au,
 						NTP: time.Now(),
 					})
 				})
@@ -96,10 +96,10 @@ func (s *hlsSource) run(ctx context.Context, cnf *conf.PathConf, reloadConf chan
 					}},
 				}
 
-				c.OnData(track, func(pts time.Duration, unit interface{}) {
+				c.OnDataH26x(track, func(pts time.Duration, dts time.Duration, au [][]byte) {
 					stream.writeUnit(medi, medi.Formats[0], &formatprocessor.UnitH265{
 						PTS: pts,
-						AU:  unit.([][]byte),
+						AU:  au,
 						NTP: time.Now(),
 					})
 				})
@@ -116,10 +116,10 @@ func (s *hlsSource) run(ctx context.Context, cnf *conf.PathConf, reloadConf chan
 					}},
 				}
 
-				c.OnData(track, func(pts time.Duration, unit interface{}) {
+				c.OnDataMPEG4Audio(track, func(pts time.Duration, dts time.Duration, aus [][]byte) {
 					stream.writeUnit(medi, medi.Formats[0], &formatprocessor.UnitMPEG4AudioGeneric{
 						PTS: pts,
-						AUs: [][]byte{unit.([]byte)},
+						AUs: aus,
 						NTP: time.Now(),
 					})
 				})
@@ -133,10 +133,10 @@ func (s *hlsSource) run(ctx context.Context, cnf *conf.PathConf, reloadConf chan
 					}},
 				}
 
-				c.OnData(track, func(pts time.Duration, unit interface{}) {
+				c.OnDataOpus(track, func(pts time.Duration, dts time.Duration, packets [][]byte) {
 					stream.writeUnit(medi, medi.Formats[0], &formatprocessor.UnitOpus{
 						PTS:     pts,
-						Frame: unit.([]byte),
+						Packets: packets,
 						NTP:     time.Now(),
 					})
 				})

internal/core/mpegts_buffered_reader.go

@@ -0,0 +1,43 @@
+package core
+
+import (
+	"fmt"
+	"io"
+)
+
+// mpegtsBufferedReader is a buffered reader optimized for MPEG-TS.
+type mpegtsBufferedReader struct {
+	r         io.Reader
+	midbuf    []byte
+	midbufpos int
+}
+
+func newMPEGTSBufferedReader(r io.Reader) *mpegtsBufferedReader {
+	return &mpegtsBufferedReader{
+		r:      r,
+		midbuf: make([]byte, 0, 1500),
+	}
+}
+
+// Read implements io.Reader.
+func (r *mpegtsBufferedReader) Read(p []byte) (int, error) {
+	if r.midbufpos < len(r.midbuf) {
+		n := copy(p, r.midbuf[r.midbufpos:])
+		r.midbufpos += n
+		return n, nil
+	}
+
+	mn, err := r.r.Read(r.midbuf[:cap(r.midbuf)])
+	if err != nil {
+		return 0, err
+	}
+
+	if (mn % 188) != 0 {
+		return 0, fmt.Errorf("received packet with size %d not multiple of 188", mn)
+	}
+
+	r.midbuf = r.midbuf[:mn]
+	n := copy(p, r.midbuf)
+	r.midbufpos = n
+	return n, nil
+}

internal/core/udp_source.go

@@ -6,11 +6,8 @@ import (
 	"net"
 	"time"
 
-	"github.com/asticode/go-astits"
 	"github.com/bluenviron/gortsplib/v3/pkg/formats"
 	"github.com/bluenviron/gortsplib/v3/pkg/media"
-	"github.com/bluenviron/mediacommon/pkg/codecs/h264"
-	"github.com/bluenviron/mediacommon/pkg/codecs/mpeg4audio"
 	"github.com/bluenviron/mediacommon/pkg/formats/mpegts"
 	"golang.org/x/net/ipv4"
 
@@ -24,43 +21,6 @@ const (
 	udpMTU       = 1472
 )
 
-var opusDurations = [32]int{
-	480, 960, 1920, 2880, /* Silk NB */
-	480, 960, 1920, 2880, /* Silk MB */
-	480, 960, 1920, 2880, /* Silk WB */
-	480, 960, /* Hybrid SWB */
-	480, 960, /* Hybrid FB */
-	120, 240, 480, 960, /* CELT NB */
-	120, 240, 480, 960, /* CELT NB */
-	120, 240, 480, 960, /* CELT NB */
-	120, 240, 480, 960, /* CELT NB */
-}
-
-func opusGetPacketDuration(pkt []byte) time.Duration {
-	if len(pkt) == 0 {
-		return 0
-	}
-
-	frameDuration := opusDurations[pkt[0]>>3]
-
-	frameCount := 0
-	switch pkt[0] & 3 {
-	case 0:
-		frameCount = 1
-	case 1:
-		frameCount = 2
-	case 2:
-		frameCount = 2
-	case 3:
-		if len(pkt) < 2 {
-			return 0
-		}
-		frameCount = int(pkt[1] & 63)
-	}
-
-	return (time.Duration(frameDuration) * time.Duration(frameCount) * time.Millisecond) / 48
-}
-
 func joinMulticastGroupOnAtLeastOneInterface(p *ipv4.PacketConn, listenIP net.IP) error {
 	intfs, err := net.Interfaces()
 	if err != nil {
@@ -87,37 +47,17 @@ func joinMulticastGroupOnAtLeastOneInterface(p *ipv4.PacketConn, listenIP net.IP
 
 type packetConnReader struct {
 	pc net.PacketConn
-	midbuf    []byte
-	midbufpos int
 }
 
 func newPacketConnReader(pc net.PacketConn) *packetConnReader {
 	return &packetConnReader{
 		pc: pc,
-		midbuf: make([]byte, 0, 1500),
 	}
 }
 
 func (r *packetConnReader) Read(p []byte) (int, error) {
-	if r.midbufpos < len(r.midbuf) {
+	n, _, err := r.pc.ReadFrom(p)
-		n := copy(p, r.midbuf[r.midbufpos:])
+	return n, err
-		r.midbufpos += n
-		return n, nil
-	}
-
-	mn, _, err := r.pc.ReadFrom(r.midbuf[:cap(r.midbuf)])
-	if err != nil {
-		return 0, err
-	}
-
-	if (mn % 188) != 0 {
-		return 0, fmt.Errorf("received packet with size %d not multiple of 188", mn)
-	}
-
-	r.midbuf = r.midbuf[:mn]
-	n := copy(p, r.midbuf)
-	r.midbufpos = n
-	return n, nil
 }
 
 type udpSourceParent interface {
@@ -174,26 +114,42 @@ func (s *udpSource) run(ctx context.Context, cnf *conf.PathConf, _ chan *conf.Pa
 		}
 	}
 
-	dem := astits.NewDemuxer(
-		context.Background(),
-		newPacketConnReader(pc),
-		astits.DemuxerOptPacketSize(188))
-
 	readerErr := make(chan error)
 
 	go func() {
-		readerErr <- func() error {
+		readerErr <- s.runReader(pc)
+	}()
+
+	select {
+	case err := <-readerErr:
+		return err
+
+	case <-ctx.Done():
+		pc.Close()
+		<-readerErr
+		return fmt.Errorf("terminated")
+	}
+}
+
+func (s *udpSource) runReader(pc net.PacketConn) error {
 	pc.SetReadDeadline(time.Now().Add(time.Duration(s.readTimeout)))
-			tracks, err := mpegts.FindTracks(dem)
+	r, err := mpegts.NewReader(newMPEGTSBufferedReader(newPacketConnReader(pc)))
 	if err != nil {
 		return err
 	}
 
 	var medias media.Medias
-			mediaCallbacks := make(map[uint16]func(time.Duration, []byte), len(tracks))
 	var stream *stream
 
-			for _, track := range tracks {
+	var td *mpegts.TimeDecoder
+	decodeTime := func(t int64) time.Duration {
+		if td == nil {
+			td = mpegts.NewTimeDecoder(t)
+		}
+		return td.Decode(t)
+	}
+
+	for _, track := range r.Tracks() {
 		var medi *media.Media
 
 		switch tcodec := track.Codec.(type) {
@@ -206,19 +162,14 @@ func (s *udpSource) run(ctx context.Context, cnf *conf.PathConf, _ chan *conf.Pa
 				}},
 			}
 
-					mediaCallbacks[track.ES.ElementaryPID] = func(pts time.Duration, data []byte) {
+			r.OnDataH26x(track, func(pts int64, _ int64, au [][]byte) error {
-						au, err := h264.AnnexBUnmarshal(data)
-						if err != nil {
-							s.Log(logger.Warn, "%v", err)
-							return
-						}
-
 				stream.writeUnit(medi, medi.Formats[0], &formatprocessor.UnitH264{
-							PTS: pts,
+					PTS: decodeTime(pts),
 					AU:  au,
 					NTP: time.Now(),
 				})
-					}
+				return nil
+			})
 
 		case *mpegts.CodecH265:
 			medi = &media.Media{
@@ -228,19 +179,14 @@ func (s *udpSource) run(ctx context.Context, cnf *conf.PathConf, _ chan *conf.Pa
 				}},
 			}
 
-					mediaCallbacks[track.ES.ElementaryPID] = func(pts time.Duration, data []byte) {
+			r.OnDataH26x(track, func(pts int64, _ int64, au [][]byte) error {
-						au, err := h264.AnnexBUnmarshal(data)
-						if err != nil {
-							s.Log(logger.Warn, "%v", err)
-							return
-						}
-
 				stream.writeUnit(medi, medi.Formats[0], &formatprocessor.UnitH265{
-							PTS: pts,
+					PTS: decodeTime(pts),
 					AU:  au,
 					NTP: time.Now(),
 				})
-					}
+				return nil
+			})
 
 		case *mpegts.CodecMPEG4Audio:
 			medi = &media.Media{
@@ -254,60 +200,32 @@ func (s *udpSource) run(ctx context.Context, cnf *conf.PathConf, _ chan *conf.Pa
 				}},
 			}
 
-					mediaCallbacks[track.ES.ElementaryPID] = func(pts time.Duration, data []byte) {
+			r.OnDataMPEG4Audio(track, func(pts int64, _ int64, aus [][]byte) error {
-						var pkts mpeg4audio.ADTSPackets
-						err := pkts.Unmarshal(data)
-						if err != nil {
-							s.Log(logger.Warn, "%v", err)
-							return
-						}
-
-						aus := make([][]byte, len(pkts))
-						for i, pkt := range pkts {
-							aus[i] = pkt.AU
-						}
-
 				stream.writeUnit(medi, medi.Formats[0], &formatprocessor.UnitMPEG4AudioGeneric{
-							PTS: pts,
+					PTS: decodeTime(pts),
 					AUs: aus,
 					NTP: time.Now(),
 				})
-					}
+				return nil
+			})
 
 		case *mpegts.CodecOpus:
 			medi = &media.Media{
 				Type: media.TypeAudio,
 				Formats: []formats.Format{&formats.Opus{
 					PayloadTyp: 96,
-							IsStereo:   (tcodec.Channels == 2),
+					IsStereo:   (tcodec.ChannelCount == 2),
 				}},
 			}
 
-					mediaCallbacks[track.ES.ElementaryPID] = func(pts time.Duration, data []byte) {
+			r.OnDataOpus(track, func(pts int64, _ int64, packets [][]byte) error {
-						pos := 0
-
-						for {
-							var au mpegts.OpusAccessUnit
-							n, err := au.Unmarshal(data[pos:])
-							if err != nil {
-								s.Log(logger.Warn, "%v", err)
-								return
-							}
-							pos += n
-
 				stream.writeUnit(medi, medi.Formats[0], &formatprocessor.UnitOpus{
-								PTS:   pts,
+					PTS:     decodeTime(pts),
-								Frame: au.Frame,
+					Packets: packets,
 					NTP:     time.Now(),
 				})
-
+				return nil
-							if len(data[pos:]) == 0 {
+			})
-								break
-							}
-
-							pts += opusGetPacketDuration(au.Frame)
-						}
-					}
 		}
 
 		medias = append(medias, medi)
@@ -326,51 +244,13 @@ func (s *udpSource) run(ctx context.Context, cnf *conf.PathConf, _ chan *conf.Pa
 	s.Log(logger.Info, "ready: %s", sourceMediaInfo(medias))
 
 	stream = res.stream
-			var timedec *mpegts.TimeDecoder
 
 	for {
 		pc.SetReadDeadline(time.Now().Add(time.Duration(s.readTimeout)))
-				data, err := dem.NextData()
+		err := r.Read()
 		if err != nil {
 			return err
 		}
-
-				if data.PES == nil {
-					continue
-				}
-
-				if data.PES.Header.OptionalHeader == nil ||
-					data.PES.Header.OptionalHeader.PTSDTSIndicator == astits.PTSDTSIndicatorNoPTSOrDTS ||
-					data.PES.Header.OptionalHeader.PTSDTSIndicator == astits.PTSDTSIndicatorIsForbidden {
-					return fmt.Errorf("PTS is missing")
-				}
-
-				var pts time.Duration
-				if timedec == nil {
-					timedec = mpegts.NewTimeDecoder(data.PES.Header.OptionalHeader.PTS.Base)
-					pts = 0
-				} else {
-					pts = timedec.Decode(data.PES.Header.OptionalHeader.PTS.Base)
-				}
-
-				cb, ok := mediaCallbacks[data.PID]
-				if !ok {
-					continue
-				}
-
-				cb(pts, data.PES.Data)
-			}
-		}()
-	}()
-
-	select {
-	case err := <-readerErr:
-		return err
-
-	case <-ctx.Done():
-		pc.Close()
-		<-readerErr
-		return fmt.Errorf("terminated")
 	}
 }
 

internal/formatprocessor/opus.go

@@ -6,6 +6,7 @@ import (
 
 	"github.com/bluenviron/gortsplib/v3/pkg/formats"
 	"github.com/bluenviron/gortsplib/v3/pkg/formats/rtpsimpleaudio"
+	"github.com/bluenviron/mediacommon/pkg/codecs/opus"
 	"github.com/pion/rtp"
 
 	"github.com/bluenviron/mediamtx/internal/logger"
@@ -16,7 +17,7 @@ type UnitOpus struct {
 	RTPPackets []*rtp.Packet
 	NTP        time.Time
 	PTS        time.Duration
-	Frame      []byte
+	Packets    [][]byte
 }
 
 // GetRTPPackets implements Unit.
@@ -91,12 +92,12 @@ func (t *formatProcessorOpus) Process(unit Unit, hasNonRTSPReaders bool) error {
 				}
 			}
 
-			frame, pts, err := t.decoder.Decode(pkt)
+			packet, pts, err := t.decoder.Decode(pkt)
 			if err != nil {
 				return err
 			}
 
-			tunit.Frame = frame
+			tunit.Packets = [][]byte{packet}
 			tunit.PTS = pts
 		}
 
@@ -105,11 +106,18 @@ func (t *formatProcessorOpus) Process(unit Unit, hasNonRTSPReaders bool) error {
 	}
 
 	// encode into RTP
-	pkt, err := t.encoder.Encode(tunit.Frame, tunit.PTS)
+	var rtpPackets []*rtp.Packet //nolint:prealloc
+	pts := tunit.PTS
+	for _, packet := range tunit.Packets {
+		pkt, err := t.encoder.Encode(packet, pts)
 		if err != nil {
 			return err
 		}
-	tunit.RTPPackets = []*rtp.Packet{pkt}
+
+		rtpPackets = append(rtpPackets, pkt)
+		pts += opus.PacketDuration(packet)
+	}
+	tunit.RTPPackets = rtpPackets
 
 	return nil
 }