External/mediamtx
1
0
Fork 1
mirror of https://github.com/bluenviron/mediamtx.git synced 2025-12-25 04:22:00 -08:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

rtmp server, hls muxer: fix DTS in case of B-frames and remove PTS-DTS offset

Browse source
This commit is contained in:
aler9 2022-06-03 11:31:39 +02:00
parent 4073013f68
commit 9bd8b2cfb6
9 changed files with 126 additions and 109 deletions
Download patch file Download diff file Expand all files Collapse all files

2
go.mod
View file

@ -5,7 +5,7 @@ go 1.17
require (
code.cloudfoundry.org/bytefmt v0.0.0-20211005130812-5bb3c17173e5
github.com/abema/go-mp4 v0.7.2
github.com/aler9/gortsplib v0.0.0-20220602202513-787c516d791e
github.com/aler9/gortsplib v0.0.0-20220603090811-322a7e2f958e
github.com/asticode/go-astits v1.10.1-0.20220319093903-4abe66a9b757
github.com/fsnotify/fsnotify v1.4.9
github.com/gin-gonic/gin v1.7.2

4
go.sum
View file

@ -6,8 +6,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-20220602202513-787c516d791e h1:OCMAhEoYzacPQ6et2EKt/xy4kXksSoYFLXP9ROlH/1c=
github.com/aler9/gortsplib v0.0.0-20220602202513-787c516d791e/go.mod h1:i1e4CEs42IrbidMUNTSNOKmeGPCOHVX9P3BvPxzyMtI=
github.com/aler9/gortsplib v0.0.0-20220603090811-322a7e2f958e h1:2dtEhIrXnyogUOTsFXDwBludmmQiox2y4fqgn5wDaqQ=
github.com/aler9/gortsplib v0.0.0-20220603090811-322a7e2f958e/go.mod h1:i1e4CEs42IrbidMUNTSNOKmeGPCOHVX9P3BvPxzyMtI=
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/aler9/writerseeker v0.0.0-20220601075008-6f0e685b9c82 h1:9WgSzBLo3a9ToSVV7sRTBYZ1GGOZUpq4+5H3SN0UZq4=

37
internal/core/rtmp_conn.go
View file

@ -27,7 +27,6 @@ import (
const (
rtmpConnPauseAfterAuthError = 2 * time.Second
rtmpPTSDTSOffset = 400 * time.Millisecond // 2 samples @ 5fps
)
func pathNameAndQuery(inURL *url.URL) (string, url.Values, string) {
@ -330,7 +329,7 @@ func (c *rtmpConn) runRead(ctx context.Context) error {
var videoInitialPTS *time.Duration
videoFirstIDRFound := false
var videoFirstIDRPTS time.Duration
var videoStartDTS time.Duration
var videoDTSExtractor *h264.DTSExtractor
for {
@ -352,9 +351,10 @@ func (c *rtmpConn) runRead(ctx context.Context) error {
v := data.h264PTS
videoInitialPTS = &v
}
pts := data.h264PTS - *videoInitialPTS
pts := data.h264PTS - *videoInitialPTS
idrPresent := h264.IDRPresent(data.h264NALUs)
var dts time.Duration
// wait until we receive an IDR
if !videoFirstIDRFound {
@ -363,16 +363,26 @@ func (c *rtmpConn) runRead(ctx context.Context) error {
}
videoFirstIDRFound = true
videoFirstIDRPTS = pts
videoDTSExtractor = h264.NewDTSExtractor()
}
// normalize as this is expected in RTMP
pts -= videoFirstIDRPTS
var err error
dts, err = videoDTSExtractor.Extract(data.h264NALUs, pts)
if err != nil {
return err
}
dts, err := videoDTSExtractor.Extract(data.h264NALUs, pts)
if err != nil {
return err
videoStartDTS = dts
dts = 0
pts -= videoStartDTS
} else {
var err error
dts, err = videoDTSExtractor.Extract(data.h264NALUs, pts)
if err != nil {
return err
}
dts -= videoStartDTS
pts -= videoStartDTS
}
// insert a H264DecoderConfig before every IDR
@ -412,7 +422,7 @@ func (c *rtmpConn) runRead(ctx context.Context) error {
Type: av.H264,
Data: avcc,
Time: dts,
CTime: rtmpPTSDTSOffset + pts - dts,
CTime: pts - dts,
})
if err != nil {
return err
@ -430,7 +440,7 @@ func (c *rtmpConn) runRead(ctx context.Context) error {
continue
}
pts -= videoFirstIDRPTS
pts -= videoStartDTS
if pts < 0 {
continue
}
@ -440,8 +450,7 @@ func (c *rtmpConn) runRead(ctx context.Context) error {
err := c.conn.WritePacket(av.Packet{
Type: av.AAC,
Data: au,
Time: rtmpPTSDTSOffset + pts +
time.Duration(i)*aac.SamplesPerAccessUnit*time.Second/time.Duration(audioTrack.ClockRate()),
Time: pts + time.Duration(i)*aac.SamplesPerAccessUnit*time.Second/time.Duration(audioTrack.ClockRate()),
})
if err != nil {
return err

22
internal/hls/muxer_test.go
View file

@ -133,8 +133,8 @@ func TestMuxerVideoAudio(t *testing.T) {
require.NoError(t, err)
require.Equal(t, &astits.Packet{
AdaptationField: &astits.PacketAdaptationField{
Length: 119,
StuffingLength: 112,
Length: 124,
StuffingLength: 117,
HasPCR: true,
PCR: &astits.ClockReference{},
RandomAccessIndicator: true,
@ -146,14 +146,14 @@ func TestMuxerVideoAudio(t *testing.T) {
PID: 256,
},
Payload: []byte{
0x00, 0x00, 0x01, 0xe0, 0x00, 0x00, 0x80, 0xc0,
0x0a, 0x31, 0x00, 0x05, 0x32, 0x81, 0x11, 0x00,
0x03, 0x19, 0x41, 0x00, 0x00, 0x00, 0x01, 0x09,
0xf0, 0x00, 0x00, 0x00, 0x01, 0x67, 0x42, 0xc0,
0x28, 0xd9, 0x00, 0x78, 0x02, 0x27, 0xe5, 0x84,
0x00, 0x00, 0x03, 0x00, 0x04, 0x00, 0x00, 0x03,
0x00, 0xf0, 0x3c, 0x60, 0xc9, 0x20, 0x00, 0x00,
0x00, 0x01, 0x08, 0x00, 0x00, 0x00, 0x01, 0x05,
0x00, 0x00, 0x01, 0xe0, 0x00, 0x00, 0x80, 0x80,
0x05, 0x21, 0x00, 0x03, 0x19, 0x41, 0x00, 0x00,
0x00, 0x01, 0x09, 0xf0, 0x00, 0x00, 0x00, 0x01,
0x67, 0x42, 0xc0, 0x28, 0xd9, 0x00, 0x78, 0x02,
0x27, 0xe5, 0x84, 0x00, 0x00, 0x03, 0x00, 0x04,
0x00, 0x00, 0x03, 0x00, 0xf0, 0x3c, 0x60, 0xc9,
0x20, 0x00, 0x00, 0x00, 0x01, 0x08, 0x00, 0x00,
0x00, 0x01, 0x05,
},
}, pkt)
@ -174,7 +174,7 @@ func TestMuxerVideoAudio(t *testing.T) {
},
Payload: []byte{
0x00, 0x00, 0x01, 0xc0, 0x00, 0x1e, 0x80, 0x80,
0x05, 0x21, 0x00, 0x09, 0xf1, 0xa1, 0xff, 0xf1,
0x05, 0x21, 0x00, 0x07, 0xd8, 0x5f, 0xff, 0xf1,
0x50, 0x80, 0x01, 0x7f, 0xfc, 0x01, 0x02, 0x03,
0x04, 0xff, 0xf1, 0x50, 0x80, 0x01, 0x7f, 0xfc,
0x05, 0x06, 0x07, 0x08,

12
internal/hls/muxer_variant_fmp4_part.go
View file

@ -20,7 +20,6 @@ func mp4PartGenerateVideoTraf(
w *mp4Writer,
trackID int,
videoSamples []*fmp4VideoSample,
startDTS time.Duration,
) (*mp4.Trun, int, error) {
/*
traf
@ -51,7 +50,7 @@ func mp4PartGenerateVideoTraf(
Version: 1,
},
// sum of decode durations of all earlier samples
BaseMediaDecodeTimeV1: uint64(durationGoToMp4(startDTS, fmp4VideoTimescale)),
BaseMediaDecodeTimeV1: uint64(durationGoToMp4(videoSamples[0].dts, fmp4VideoTimescale)),
})
if err != nil {
return nil, 0, err
@ -184,7 +183,6 @@ func mp4PartGenerate(
audioTrack *gortsplib.TrackAAC,
videoSamples []*fmp4VideoSample,
audioSamples []*fmp4AudioSample,
startDTS time.Duration,
) ([]byte, error) {
/*
moof
@ -215,7 +213,7 @@ func mp4PartGenerate(
if videoTrack != nil {
var err error
videoTrun, videoTrunOffset, err = mp4PartGenerateVideoTraf(
w, trackID, videoSamples, startDTS)
w, trackID, videoSamples)
if err != nil {
return nil, err
}
@ -303,7 +301,6 @@ type muxerVariantFMP4Part struct {
videoTrack *gortsplib.TrackH264
audioTrack *gortsplib.TrackAAC
id uint64
startDTS time.Duration
isIndependent bool
videoSamples []*fmp4VideoSample
@ -316,13 +313,11 @@ func newMuxerVariantFMP4Part(
videoTrack *gortsplib.TrackH264,
audioTrack *gortsplib.TrackAAC,
id uint64,
startDTS time.Duration,
) *muxerVariantFMP4Part {
p := &muxerVariantFMP4Part{
videoTrack: videoTrack,
audioTrack: audioTrack,
id: id,
startDTS: startDTS,
}
if videoTrack == nil {
@ -363,8 +358,7 @@ func (p *muxerVariantFMP4Part) finalize() error {
p.videoTrack,
p.audioTrack,
p.videoSamples,
p.audioSamples,
p.startDTS)
p.audioSamples)
if err != nil {
return err
}

5
internal/hls/muxer_variant_fmp4_segment.go
View file

@ -83,7 +83,6 @@ func newMuxerVariantFMP4Segment(
s.videoTrack,
s.audioTrack,
s.genPartID(),
s.startDTS,
)
return s
@ -118,7 +117,7 @@ func (s *muxerVariantFMP4Segment) finalize(
s.currentPart = nil
if s.videoTrack != nil {
s.renderedDuration = nextVideoSample.pts - s.startDTS
s.renderedDuration = nextVideoSample.dts - s.startDTS
} else {
s.renderedDuration = 0
for _, pa := range s.parts {
@ -155,7 +154,6 @@ func (s *muxerVariantFMP4Segment) writeH264(sample *fmp4VideoSample, adjustedPar
s.videoTrack,
s.audioTrack,
s.genPartID(),
sample.next.dts,
)
}
@ -188,7 +186,6 @@ func (s *muxerVariantFMP4Segment) writeAAC(sample *fmp4AudioSample, adjustedPart
s.videoTrack,
s.audioTrack,
s.genPartID(),
sample.next.pts,
)
}

51
internal/hls/muxer_variant_fmp4_segmenter.go
View file

@ -54,6 +54,7 @@ type muxerVariantFMP4Segmenter struct {
onSegmentFinalized func(*muxerVariantFMP4Segment)
onPartFinalized func(*muxerVariantFMP4Part)
startDTS time.Duration
videoFirstIDRReceived bool
videoDTSExtractor *h264.DTSExtractor
videoSPS []byte
@ -146,15 +147,28 @@ func (m *muxerVariantFMP4Segmenter) writeH264Entry(sample *fmp4VideoSample) erro
m.videoFirstIDRReceived = true
m.videoDTSExtractor = h264.NewDTSExtractor()
m.videoSPS = append([]byte(nil), m.videoTrack.SPS()...)
}
// fill DTS
var err error
sample.dts, err = m.videoDTSExtractor.Extract(sample.nalus, sample.pts)
if err != nil {
return err
var err error
sample.dts, err = m.videoDTSExtractor.Extract(sample.nalus, sample.pts)
if err != nil {
return err
}
sample.nalus = nil
m.startDTS = sample.dts
sample.dts = 0
sample.pts -= m.startDTS
} else {
var err error
sample.dts, err = m.videoDTSExtractor.Extract(sample.nalus, sample.pts)
if err != nil {
return err
}
sample.nalus = nil
sample.dts -= m.startDTS
sample.pts -= m.startDTS
}
sample.nalus = nil
// put samples into a queue in order to
// - allow to compute sample duration
@ -173,7 +187,7 @@ func (m *muxerVariantFMP4Segmenter) writeH264Entry(sample *fmp4VideoSample) erro
m.lowLatency,
m.genSegmentID(),
now,
0,
sample.dts,
m.segmentMaxSize,
m.videoTrack,
m.audioTrack,
@ -184,7 +198,7 @@ func (m *muxerVariantFMP4Segmenter) writeH264Entry(sample *fmp4VideoSample) erro
m.adjustPartDuration(sample.duration())
err = m.currentSegment.writeH264(sample, m.adjustedPartDuration)
err := m.currentSegment.writeH264(sample, m.adjustedPartDuration)
if err != nil {
return err
}
@ -194,7 +208,7 @@ func (m *muxerVariantFMP4Segmenter) writeH264Entry(sample *fmp4VideoSample) erro
sps := m.videoTrack.SPS()
spsChanged := !bytes.Equal(m.videoSPS, sps)
if (sample.next.pts-m.currentSegment.startDTS) >= m.segmentDuration ||
if (sample.next.dts-m.currentSegment.startDTS) >= m.segmentDuration ||
spsChanged {
err := m.currentSegment.finalize(sample.next, nil)
if err != nil {
@ -208,7 +222,7 @@ func (m *muxerVariantFMP4Segmenter) writeH264Entry(sample *fmp4VideoSample) erro
m.lowLatency,
m.genSegmentID(),
now,
sample.next.pts,
sample.next.dts,
m.segmentMaxSize,
m.videoTrack,
m.audioTrack,
@ -242,6 +256,15 @@ func (m *muxerVariantFMP4Segmenter) writeAAC(pts time.Duration, aus [][]byte) er
}
func (m *muxerVariantFMP4Segmenter) writeAACEntry(sample *fmp4AudioSample) error {
if m.videoTrack != nil {
// wait for the video track
if !m.videoFirstIDRReceived {
return nil
}
sample.pts -= m.startDTS
}
// put samples into a queue in order to
// allow to compute the sample duration
sample, m.nextAudioSample = m.nextAudioSample, sample
@ -259,7 +282,7 @@ func (m *muxerVariantFMP4Segmenter) writeAACEntry(sample *fmp4AudioSample) error
m.lowLatency,
m.genSegmentID(),
now,
0,
sample.pts,
m.segmentMaxSize,
m.videoTrack,
m.audioTrack,
@ -274,9 +297,7 @@ func (m *muxerVariantFMP4Segmenter) writeAACEntry(sample *fmp4AudioSample) error
}
}
m.adjustPartDuration(sample.duration())
err := m.currentSegment.writeAAC(sample, m.adjustedPartDuration)
err := m.currentSegment.writeAAC(sample, m.partDuration)
if err != nil {
return err
}

36
internal/hls/muxer_variant_mpegts_segment.go
View file

@ -14,8 +14,7 @@ import (
)
const (
mpegtsPCROffset = 400 * time.Millisecond // 2 samples @ 5fps
mpegtsPTSDTSOffset = 400 * time.Millisecond // 2 samples @ 5fps
mpegtsPCROffset = 400 * time.Millisecond // 2 samples @ 5fps
)
type muxerVariantMPEGTSSegment struct {
@ -27,8 +26,8 @@ type muxerVariantMPEGTSSegment struct {
startTime time.Time
name string
buf bytes.Buffer
startPTS *time.Duration
endPTS time.Duration
startDTS *time.Duration
endDTS time.Duration
pcrSendCounter int
audioAUCount int
}
@ -58,7 +57,7 @@ func newMuxerVariantMPEGTSSegment(
}
func (t *muxerVariantMPEGTSSegment) duration() time.Duration {
return t.endPTS - *t.startPTS
return t.endDTS - *t.startDTS
}
func (t *muxerVariantMPEGTSSegment) write(p []byte) (int, error) {
@ -110,13 +109,13 @@ func (t *muxerVariantMPEGTSSegment) writeH264(
MarkerBits: 2,
}
if dts == (pts + mpegtsPTSDTSOffset) {
if dts == pts {
oh.PTSDTSIndicator = astits.PTSDTSIndicatorOnlyPTS
oh.PTS = &astits.ClockReference{Base: int64((pts + mpegtsPTSDTSOffset + mpegtsPCROffset).Seconds() * 90000)}
oh.PTS = &astits.ClockReference{Base: int64((pts + mpegtsPCROffset).Seconds() * 90000)}
} else {
oh.PTSDTSIndicator = astits.PTSDTSIndicatorBothPresent
oh.DTS = &astits.ClockReference{Base: int64((dts + mpegtsPCROffset).Seconds() * 90000)}
oh.PTS = &astits.ClockReference{Base: int64((pts + mpegtsPTSDTSOffset + mpegtsPCROffset).Seconds() * 90000)}
oh.PTS = &astits.ClockReference{Base: int64((pts + mpegtsPCROffset).Seconds() * 90000)}
}
_, err = t.writeData(&astits.MuxerData{
@ -134,13 +133,11 @@ func (t *muxerVariantMPEGTSSegment) writeH264(
return err
}
if t.startPTS == nil {
t.startPTS = &pts
if t.startDTS == nil {
t.startDTS = &dts
}
if pts > t.endPTS {
t.endPTS = pts
}
t.endDTS = dts
return nil
}
@ -188,8 +185,7 @@ func (t *muxerVariantMPEGTSSegment) writeAAC(
OptionalHeader: &astits.PESOptionalHeader{
MarkerBits: 2,
PTSDTSIndicator: astits.PTSDTSIndicatorOnlyPTS,
PTS: &astits.ClockReference{Base: int64((pts + mpegtsPTSDTSOffset +
mpegtsPCROffset).Seconds() * 90000)},
PTS: &astits.ClockReference{Base: int64((pts + mpegtsPCROffset).Seconds() * 90000)},
},
PacketLength: uint16(len(enc) + 8),
StreamID: 192, // audio
@ -203,14 +199,12 @@ func (t *muxerVariantMPEGTSSegment) writeAAC(
if t.videoTrack == nil {
t.audioAUCount += len(aus)
}
if t.startPTS == nil {
t.startPTS = &pts
}
if t.startDTS == nil {
t.startDTS = &pts
}
if pts > t.endPTS {
t.endPTS = pts
t.endDTS = pts
}
return nil

66
internal/hls/muxer_variant_mpegts_segmenter.go
View file

@ -30,7 +30,7 @@ type muxerVariantMPEGTSSegmenter struct {
currentSegment *muxerVariantMPEGTSSegment
videoDTSExtractor *h264.DTSExtractor
startPCR time.Time
startPTS time.Duration
startDTS time.Duration
}
func newMuxerVariantMPEGTSSegmenter(
@ -80,6 +80,7 @@ func newMuxerVariantMPEGTSSegmenter(
func (m *muxerVariantMPEGTSSegmenter) writeH264(pts time.Duration, nalus [][]byte) error {
now := time.Now()
idrPresent := h264.IDRPresent(nalus)
var dts time.Duration
if m.currentSegment == nil {
// skip groups silently until we find one with a IDR
@ -87,44 +88,49 @@ func (m *muxerVariantMPEGTSSegmenter) writeH264(pts time.Duration, nalus [][]byt
return nil
}
m.videoDTSExtractor = h264.NewDTSExtractor()
var err error
dts, err = m.videoDTSExtractor.Extract(nalus, pts)
if err != nil {
return err
}
m.startPCR = now
m.startDTS = dts
dts = 0
pts -= m.startDTS
// create first segment
m.currentSegment = newMuxerVariantMPEGTSSegment(now, m.segmentMaxSize,
m.videoTrack, m.audioTrack, m.writer.WriteData)
m.startPCR = now
m.startPTS = pts
m.videoDTSExtractor = h264.NewDTSExtractor()
pts = 0
} else {
// normalize in order to sync with PCR
pts -= m.startPTS
var err error
dts, err = m.videoDTSExtractor.Extract(nalus, pts)
if err != nil {
return err
}
dts -= m.startDTS
pts -= m.startDTS
// switch segment
if idrPresent &&
m.currentSegment.startPTS != nil &&
(pts-*m.currentSegment.startPTS) >= m.segmentDuration {
m.currentSegment.endPTS = pts
(dts-*m.currentSegment.startDTS) >= m.segmentDuration {
m.currentSegment.endDTS = dts
m.onSegmentReady(m.currentSegment)
m.currentSegment = newMuxerVariantMPEGTSSegment(now, m.segmentMaxSize,
m.videoTrack, m.audioTrack, m.writer.WriteData)
}
}
dts, err := m.videoDTSExtractor.Extract(nalus, pts)
if err != nil {
return err
}
err = m.currentSegment.writeH264(
err := m.currentSegment.writeH264(
now.Sub(m.startPCR),
dts,
pts,
idrPresent,
nalus)
if err != nil {
if m.currentSegment.buf.Len() > 0 {
m.onSegmentReady(m.currentSegment)
}
m.currentSegment = nil
return err
}
@ -136,20 +142,20 @@ func (m *muxerVariantMPEGTSSegmenter) writeAAC(pts time.Duration, aus [][]byte)
if m.videoTrack == nil {
if m.currentSegment == nil {
m.startPCR = now
m.startDTS = pts
pts = 0
// create first segment
m.currentSegment = newMuxerVariantMPEGTSSegment(now, m.segmentMaxSize,
m.videoTrack, m.audioTrack, m.writer.WriteData)
m.startPCR = now
m.startPTS = pts
pts = 0
} else {
pts -= m.startPTS
pts -= m.startDTS
// switch segment
if m.currentSegment.audioAUCount >= mpegtsSegmentMinAUCount &&
m.currentSegment.startPTS != nil &&
(pts-*m.currentSegment.startPTS) >= m.segmentDuration {
m.currentSegment.endPTS = pts
(pts-*m.currentSegment.startDTS) >= m.segmentDuration {
m.currentSegment.endDTS = pts
m.onSegmentReady(m.currentSegment)
m.currentSegment = newMuxerVariantMPEGTSSegment(now, m.segmentMaxSize,
m.videoTrack, m.audioTrack, m.writer.WriteData)
@ -161,15 +167,11 @@ func (m *muxerVariantMPEGTSSegmenter) writeAAC(pts time.Duration, aus [][]byte)
return nil
}
pts -= m.startPTS
pts -= m.startDTS
}
err := m.currentSegment.writeAAC(now.Sub(m.startPCR), pts, aus)
if err != nil {
if m.currentSegment.buf.Len() > 0 {
m.onSegmentReady(m.currentSegment)
}
m.currentSegment = nil
return err
}