From 68036b68c1c5fa83c649ff41b8921d8133a90d8d Mon Sep 17 00:00:00 2001 From: Alex X Date: Wed, 19 Nov 2025 12:58:23 +0300 Subject: [PATCH 1/2] Fix timestamp processing for HTTP-FLV --- pkg/flv/flv_test.go | 21 +++++++++++++++++++++ pkg/flv/producer.go | 8 ++++++-- 2 files changed, 27 insertions(+), 2 deletions(-) create mode 100644 pkg/flv/flv_test.go diff --git a/pkg/flv/flv_test.go b/pkg/flv/flv_test.go new file mode 100644 index 00000000..389272b0 --- /dev/null +++ b/pkg/flv/flv_test.go @@ -0,0 +1,21 @@ +package flv + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +func TestTimeToRTP(t *testing.T) { + // Reolink camera has 20 FPS + // Video timestamp increases by 50ms, SampleRate 90000, RTP timestamp increases by 4500 + // Audio timestamp increases by 64ms, SampleRate 16000, RTP timestamp increases by 1024 + frameN := 1 + for i := 0; i < 32; i++ { + // 1000ms/(90000/4500) = 50ms + require.Equal(t, uint32(frameN*4500), TimeToRTP(uint32(frameN*50), 90000)) + // 1000ms/(16000/1024) = 64ms + require.Equal(t, uint32(frameN*1024), TimeToRTP(uint32(frameN*64), 16000)) + frameN *= 2 + } +} diff --git a/pkg/flv/producer.go b/pkg/flv/producer.go index 9748c94a..38c601d5 100644 --- a/pkg/flv/producer.go +++ b/pkg/flv/producer.go @@ -299,8 +299,12 @@ func (c *Producer) readPacket() (*rtp.Packet, error) { return pkt, nil } -func TimeToRTP(timeMS uint32, clockRate uint32) uint32 { - return timeMS * clockRate / 1000 +// TimeToRTP convert time in milliseconds to RTP time +func TimeToRTP(timeMS, clockRate uint32) uint32 { + // for clockRates 90000, 16000, 8000, etc. - we can use: + // return timeMS * (clockRate / 1000) + // but for clockRates 44100, 22050, 11025 - we should use: + return uint32(uint64(timeMS) * uint64(clockRate) / 1000) } func isExHeader(data []byte) bool { From aa0ece2d1e396cd4b044f2e069d4ec394586ce77 Mon Sep 17 00:00:00 2001 From: Alex X Date: Thu, 20 Nov 2025 17:52:08 +0300 Subject: [PATCH 2/2] Fix adts producer for VLC player support #1643 --- pkg/aac/adts.go | 15 +++++++++++++-- pkg/aac/producer.go | 36 +++++++++++++++++++++++++----------- pkg/aac/rtp.go | 11 +++++++---- 3 files changed, 45 insertions(+), 17 deletions(-) diff --git a/pkg/aac/adts.go b/pkg/aac/adts.go index 6688d319..8bdc3a3d 100644 --- a/pkg/aac/adts.go +++ b/pkg/aac/adts.go @@ -8,8 +8,19 @@ import ( "github.com/pion/rtp" ) +const ADTSHeaderSize = 7 + func IsADTS(b []byte) bool { - return len(b) > 7 && b[0] == 0xFF && b[1]&0xF6 == 0xF0 + // AAAAAAAA AAAABCCD EEFFFFGH HHIJKLMM MMMMMMMM MMMOOOOO OOOOOOPP (QQQQQQQQ QQQQQQQQ) + // A 12 Syncword, all bits must be set to 1. + // C 2 Layer, always set to 0. + return len(b) >= ADTSHeaderSize && b[0] == 0xFF && b[1]&0b1111_0110 == 0xF0 +} + +func HasCRC(b []byte) bool { + // AAAAAAAA AAAABCCD EEFFFFGH HHIJKLMM MMMMMMMM MMMOOOOO OOOOOOPP (QQQQQQQQ QQQQQQQQ) + // D 1 Protection absence, set to 1 if there is no CRC and 0 if there is CRC. + return b[1]&0b1 == 0 } func ADTSToCodec(b []byte) *core.Codec { @@ -58,7 +69,7 @@ func ADTSToCodec(b []byte) *core.Codec { func ReadADTSSize(b []byte) uint16 { // AAAAAAAA AAAABCCD EEFFFFGH HHIJKLMM MMMMMMMM MMMOOOOO OOOOOOPP (QQQQQQQQ QQQQQQQQ) _ = b[5] // bounds - return uint16(b[3]&0x03)<<(8+3) | uint16(b[4])<<3 | uint16(b[5]>>5) + return uint16(b[3]&0b11)<<11 | uint16(b[4])<<3 | uint16(b[5]>>5) } func WriteADTSSize(b []byte, size uint16) { diff --git a/pkg/aac/producer.go b/pkg/aac/producer.go index efd2d175..a2c73f92 100644 --- a/pkg/aac/producer.go +++ b/pkg/aac/producer.go @@ -2,7 +2,7 @@ package aac import ( "bufio" - "encoding/binary" + "errors" "io" "github.com/AlexxIT/go2rtc/pkg/core" @@ -17,16 +17,22 @@ type Producer struct { func Open(r io.Reader) (*Producer, error) { rd := bufio.NewReader(r) - b, err := rd.Peek(8) + b, err := rd.Peek(ADTSHeaderSize) if err != nil { return nil, err } + codec := ADTSToCodec(b) + if codec == nil { + return nil, errors.New("adts: wrong header") + } + codec.PayloadType = core.PayloadTypeRAW + medias := []*core.Media{ { Kind: core.KindAudio, Direction: core.DirectionRecvonly, - Codecs: []*core.Codec{ADTSToCodec(b)}, + Codecs: []*core.Codec{codec}, }, } return &Producer{ @@ -42,14 +48,25 @@ func Open(r io.Reader) (*Producer, error) { func (c *Producer) Start() error { for { - b, err := c.rd.Peek(6) - if err != nil { + // read ADTS header + adts := make([]byte, ADTSHeaderSize) + if _, err := io.ReadFull(c.rd, adts); err != nil { return err } - auSize := ReadADTSSize(b) - payload := make([]byte, 2+2+auSize) - if _, err = io.ReadFull(c.rd, payload[4:]); err != nil { + auSize := ReadADTSSize(adts) - ADTSHeaderSize + + if HasCRC(adts) { + // skip CRC after header + if _, err := c.rd.Discard(2); err != nil { + return err + } + auSize -= 2 + } + + // read AAC payload after header + payload := make([]byte, auSize) + if _, err := io.ReadFull(c.rd, payload); err != nil { return err } @@ -59,9 +76,6 @@ func (c *Producer) Start() error { continue } - payload[1] = 16 // header size in bits - binary.BigEndian.PutUint16(payload[2:], auSize<<3) - pkt := &rtp.Packet{ Header: rtp.Header{Timestamp: core.Now90000()}, Payload: payload, diff --git a/pkg/aac/rtp.go b/pkg/aac/rtp.go index 1faa2e27..08846c06 100644 --- a/pkg/aac/rtp.go +++ b/pkg/aac/rtp.go @@ -8,7 +8,6 @@ import ( ) const RTPPacketVersionAAC = 0 -const ADTSHeaderSize = 7 func RTPDepay(handler core.HandlerFunc) core.HandlerFunc { var timestamp uint32 @@ -65,7 +64,8 @@ func RTPDepay(handler core.HandlerFunc) core.HandlerFunc { } func RTPPay(handler core.HandlerFunc) core.HandlerFunc { - sequencer := rtp.NewRandomSequencer() + var seq uint16 + var ts uint32 return func(packet *rtp.Packet) { if packet.Version != RTPPacketVersionAAC { @@ -85,12 +85,15 @@ func RTPPay(handler core.HandlerFunc) core.HandlerFunc { Header: rtp.Header{ Version: 2, Marker: true, - SequenceNumber: sequencer.NextSequenceNumber(), - Timestamp: packet.Timestamp, + SequenceNumber: seq, + Timestamp: ts, }, Payload: payload, } handler(&clone) + + seq++ + ts += AUTime } }