pimhwang
/
mediamtx
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

move format processors into dedicated folder

pull/1364/head
aler9 4 years ago
parent
535cbe41e8
commit
e3fff72b7c
26 changed files with 690 additions and 912 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 13
      internal/core/data.go
  2. 34
      internal/core/formatprocessor.go
  3. 91
      internal/core/formatprocessor_generic_test.go
  4. 450
      internal/core/formatprocessor_h264_test.go
  5. 51
      internal/core/hls_muxer.go
  6. 17
      internal/core/hls_source.go
  7. 9
      internal/core/rpicamera_source.go
  8. 63
      internal/core/rtmp_conn.go
  9. 17
      internal/core/rtmp_source.go
  10. 31
      internal/core/rtsp_server_test.go
  11. 43
      internal/core/rtsp_session.go
  12. 43
      internal/core/rtsp_source.go
  13. 6
      internal/core/stream.go
  14. 20
      internal/core/streamformat.go
  15. 49
      internal/core/webrtc_conn.go
  16. 27
      internal/formatprocessor/generic.go
  17. 51
      internal/formatprocessor/generic_test.go
  18. 51
      internal/formatprocessor/h264.go
  19. 140
      internal/formatprocessor/h264_test.go
  20. 51
      internal/formatprocessor/h265.go
  21. 125
      internal/formatprocessor/h265_test.go
  22. 43
      internal/formatprocessor/mpeg4audio.go
  23. 43
      internal/formatprocessor/opus.go
  24. 48
      internal/formatprocessor/processor.go
  25. 43
      internal/formatprocessor/vp8.go
  26. 43
      internal/formatprocessor/vp9.go

13
internal/core/data.go
Unescape View File

@ -1,13 +0,0 @@
package core
import (
"time"
"github.com/pion/rtp"
)
// data is the data unit routed across the server.
type data interface {
getRTPPackets() []*rtp.Packet
getNTP() time.Time
}

34
internal/core/formatprocessor.go
Unescape View File

@ -1,34 +0,0 @@
package core
import (
"github.com/aler9/gortsplib/v2/pkg/format"
)
type formatProcessor interface {
process(data, bool) error
}
func newFormatProcessor(forma format.Format, generateRTPPackets bool) (formatProcessor, error) {
switch forma := forma.(type) {
case *format.H264:
return newFormatProcessorH264(forma, generateRTPPackets)
case *format.H265:
return newFormatProcessorH265(forma, generateRTPPackets)
case *format.VP8:
return newFormatProcessorVP8(forma, generateRTPPackets)
case *format.VP9:
return newFormatProcessorVP9(forma, generateRTPPackets)
case *format.MPEG4Audio:
return newFormatProcessorMPEG4Audio(forma, generateRTPPackets)
case *format.Opus:
return newFormatProcessorOpus(forma, generateRTPPackets)
default:
return newFormatProcessorGeneric(forma, generateRTPPackets)
}
}

91
internal/core/formatprocessor_generic_test.go
Unescape View File

@ -1,91 +0,0 @@
package core
import (
"testing"
"github.com/aler9/gortsplib/v2"
"github.com/aler9/gortsplib/v2/pkg/format"
"github.com/aler9/gortsplib/v2/pkg/media"
"github.com/aler9/gortsplib/v2/pkg/url"
"github.com/pion/rtp"
"github.com/stretchr/testify/require"
)
func TestFormatProcessorRemovePadding(t *testing.T) {
p, ok := newInstance("rtmpDisable: yes\n" +
"hlsDisable: yes\n" +
"webrtcDisable: yes\n" +
"paths:\n" +
" all:\n")
require.Equal(t, true, ok)
defer p.Close()
forma := &format.Generic{
PayloadTyp: 96,
RTPMap: "private/90000",
}
forma.Init()
medi := &media.Media{
Type: media.TypeApplication,
Formats: []format.Format{forma},
}
source := gortsplib.Client{}
err := source.StartRecording(
"rtsp://localhost:8554/stream",
media.Medias{medi})
require.NoError(t, err)
defer source.Close()
c := gortsplib.Client{}
u, err := url.Parse("rtsp://127.0.0.1:8554/stream")
require.NoError(t, err)
err = c.Start(u.Scheme, u.Host)
require.NoError(t, err)
defer c.Close()
medias, baseURL, _, err := c.Describe(u)
require.NoError(t, err)
err = c.SetupAll(medias, baseURL)
require.NoError(t, err)
packetRecv := make(chan struct{})
c.OnPacketRTP(medias[0], medias[0].Formats[0], func(pkt *rtp.Packet) {
require.Equal(t, &rtp.Packet{
Header: rtp.Header{
Version: 2,
Marker: true,
PayloadType: 96,
SequenceNumber: 123,
Timestamp: 45343,
SSRC: 563423,
CSRC: []uint32{},
},
Payload: []byte{0x01, 0x02, 0x03, 0x04},
}, pkt)
close(packetRecv)
})
_, err = c.Play(nil)
require.NoError(t, err)
source.WritePacketRTP(medi, &rtp.Packet{
Header: rtp.Header{
Version: 2,
Marker: true,
PayloadType: 96,
SequenceNumber: 123,
Timestamp: 45343,
SSRC: 563423,
Padding: true,
},
Payload: []byte{0x01, 0x02, 0x03, 0x04},
PaddingSize: 20,
})
<-packetRecv
}

450
internal/core/formatprocessor_h264_test.go
Unescape View File

@ -1,450 +0,0 @@
package core
import (
"bytes"
"net"
"testing"
"github.com/aler9/gortsplib/v2"
"github.com/aler9/gortsplib/v2/pkg/base"
"github.com/aler9/gortsplib/v2/pkg/codecs/h265"
"github.com/aler9/gortsplib/v2/pkg/conn"
"github.com/aler9/gortsplib/v2/pkg/format"
"github.com/aler9/gortsplib/v2/pkg/headers"
"github.com/aler9/gortsplib/v2/pkg/media"
"github.com/aler9/gortsplib/v2/pkg/url"
"github.com/pion/rtp"
"github.com/stretchr/testify/require"
)
func TestFormatProcessorDynamicParams(t *testing.T) {
checkTrack := func(t *testing.T, forma format.Format) {
c := gortsplib.Client{}
u, err := url.Parse("rtsp://127.0.0.1:8554/stream")
require.NoError(t, err)
err = c.Start(u.Scheme, u.Host)
require.NoError(t, err)
defer c.Close()
medias, _, _, err := c.Describe(u)
require.NoError(t, err)
forma1 := medias[0].Formats[0]
require.Equal(t, forma, forma1)
}
for _, ca := range []string{"h264", "h265"} {
t.Run(ca, func(t *testing.T) {
p, ok := newInstance("rtmpDisable: yes\n" +
"hlsDisable: yes\n" +
"webrtcDisable: yes\n" +
"paths:\n" +
" all:\n")
require.Equal(t, true, ok)
defer p.Close()
formah264 := &format.H264{
PayloadTyp: 96,
PacketizationMode: 1,
}
formah265 := &format.H265{
PayloadTyp: 96,
}
var forma format.Format
if ca == "h264" {
forma = formah264
} else {
forma = formah265
}
medi := &media.Media{
Type: media.TypeVideo,
Formats: []format.Format{forma},
}
source := gortsplib.Client{}
err := source.StartRecording(
"rtsp://localhost:8554/stream",
media.Medias{medi})
require.NoError(t, err)
defer source.Close()
if ca == "h264" {
enc := formah264.CreateEncoder()
pkts, err := enc.Encode([][]byte{{7, 1, 2, 3}}, 0) // SPS
require.NoError(t, err)
source.WritePacketRTP(medi, pkts[0])
pkts, err = enc.Encode([][]byte{{8}}, 0) // PPS
require.NoError(t, err)
source.WritePacketRTP(medi, pkts[0])
checkTrack(t, &format.H264{
PayloadTyp: 96,
PacketizationMode: 1,
SPS: []byte{7, 1, 2, 3},
PPS: []byte{8},
})
pkts, err = enc.Encode([][]byte{{7, 4, 5, 6}}, 0) // SPS
require.NoError(t, err)
source.WritePacketRTP(medi, pkts[0])
pkts, err = enc.Encode([][]byte{{8, 1}}, 0) // PPS
require.NoError(t, err)
source.WritePacketRTP(medi, pkts[0])
checkTrack(t, &format.H264{
PayloadTyp: 96,
PacketizationMode: 1,
SPS: []byte{7, 4, 5, 6},
PPS: []byte{8, 1},
})
} else {
enc := formah265.CreateEncoder()
pkts, err := enc.Encode([][]byte{{byte(h265.NALUType_VPS_NUT) << 1, 1, 2, 3}}, 0)
require.NoError(t, err)
source.WritePacketRTP(medi, pkts[0])
pkts, err = enc.Encode([][]byte{{byte(h265.NALUType_SPS_NUT) << 1, 4, 5, 6}}, 0)
require.NoError(t, err)
source.WritePacketRTP(medi, pkts[0])
pkts, err = enc.Encode([][]byte{{byte(h265.NALUType_PPS_NUT) << 1, 7, 8, 9}}, 0)
require.NoError(t, err)
source.WritePacketRTP(medi, pkts[0])
checkTrack(t, &format.H265{
PayloadTyp: 96,
VPS: []byte{byte(h265.NALUType_VPS_NUT) << 1, 1, 2, 3},
SPS: []byte{byte(h265.NALUType_SPS_NUT) << 1, 4, 5, 6},
PPS: []byte{byte(h265.NALUType_PPS_NUT) << 1, 7, 8, 9},
})
pkts, err = enc.Encode([][]byte{{byte(h265.NALUType_VPS_NUT) << 1, 10, 11, 12}}, 0)
require.NoError(t, err)
source.WritePacketRTP(medi, pkts[0])
pkts, err = enc.Encode([][]byte{{byte(h265.NALUType_SPS_NUT) << 1, 13, 14, 15}}, 0)
require.NoError(t, err)
source.WritePacketRTP(medi, pkts[0])
pkts, err = enc.Encode([][]byte{{byte(h265.NALUType_PPS_NUT) << 1, 16, 17, 18}}, 0)
require.NoError(t, err)
source.WritePacketRTP(medi, pkts[0])
checkTrack(t, &format.H265{
PayloadTyp: 96,
VPS: []byte{byte(h265.NALUType_VPS_NUT) << 1, 10, 11, 12},
SPS: []byte{byte(h265.NALUType_SPS_NUT) << 1, 13, 14, 15},
PPS: []byte{byte(h265.NALUType_PPS_NUT) << 1, 16, 17, 18},
})
}
})
}
}
func TestFormatProcessorOversizedPackets(t *testing.T) {
for _, ca := range []string{"h264", "h265"} {
t.Run(ca, func(t *testing.T) {
p, ok := newInstance("rtmpDisable: yes\n" +
"hlsDisable: yes\n" +
"webrtcDisable: yes\n" +
"paths:\n" +
" all:\n")
require.Equal(t, true, ok)
defer p.Close()
nconn, err := net.Dial("tcp", "localhost:8554")
require.NoError(t, err)
defer nconn.Close()
conn := conn.NewConn(nconn)
var forma format.Format
if ca == "h264" {
forma = &format.H264{
PayloadTyp: 96,
SPS: []byte{0x01, 0x02, 0x03, 0x04},
PPS: []byte{0x01, 0x02, 0x03, 0x04},
PacketizationMode: 1,
}
} else {
forma = &format.H265{
PayloadTyp: 96,
VPS: []byte{byte(h265.NALUType_VPS_NUT) << 1, 10, 11, 12},
SPS: []byte{byte(h265.NALUType_SPS_NUT) << 1, 13, 14, 15},
PPS: []byte{byte(h265.NALUType_PPS_NUT) << 1, 16, 17, 18},
}
}
medi := &media.Media{
Type: media.TypeVideo,
Formats: []format.Format{forma},
}
medias := media.Medias{medi}
medias.SetControls()
res, err := writeReqReadRes(conn, base.Request{
Method: base.Announce,
URL: mustParseURL("rtsp://localhost:8554/stream"),
Header: base.Header{
"CSeq": base.HeaderValue{"1"},
"Content-Type": base.HeaderValue{"application/sdp"},
},
Body: mustMarshalSDP(medias.Marshal(false)),
})
require.NoError(t, err)
require.Equal(t, base.StatusOK, res.StatusCode)
var sx headers.Session
inTH := &headers.Transport{
Delivery: func() *headers.TransportDelivery {
v := headers.TransportDeliveryUnicast
return &v
}(),
Mode: func() *headers.TransportMode {
v := headers.TransportModeRecord
return &v
}(),
Protocol: headers.TransportProtocolTCP,
InterleavedIDs: &[2]int{0, 1},
}
res, err = writeReqReadRes(conn, base.Request{
Method: base.Setup,
URL: mustParseURL("rtsp://localhost:8554/stream/" + medias[0].Control),
Header: base.Header{
"CSeq": base.HeaderValue{"2"},
"Transport": inTH.Marshal(),
},
})
require.NoError(t, err)
require.Equal(t, base.StatusOK, res.StatusCode)
err = sx.Unmarshal(res.Header["Session"])
require.NoError(t, err)
res, err = writeReqReadRes(conn, base.Request{
Method: base.Record,
URL: mustParseURL("rtsp://localhost:8554/stream"),
Header: base.Header{
"CSeq": base.HeaderValue{"3"},
"Session": base.HeaderValue{sx.Session},
},
})
require.NoError(t, err)
require.Equal(t, base.StatusOK, res.StatusCode)
c := gortsplib.Client{}
u, err := url.Parse("rtsp://127.0.0.1:8554/stream")
require.NoError(t, err)
err = c.Start(u.Scheme, u.Host)
require.NoError(t, err)
defer c.Close()
medias, baseURL, _, err := c.Describe(u)
require.NoError(t, err)
err = c.SetupAll(medias, baseURL)
require.NoError(t, err)
packetRecv := make(chan struct{})
i := 0
var expected []*rtp.Packet
if ca == "h264" {
expected = []*rtp.Packet{
{
Header: rtp.Header{
Version: 2,
Marker: true,
PayloadType: 96,
SequenceNumber: 123,
Timestamp: 45343,
SSRC: 563423,
CSRC: []uint32{},
},
Payload: []byte{0x01, 0x02, 0x03, 0x04},
},
{
Header: rtp.Header{
Version: 2,
Marker: false,
PayloadType: 96,
SequenceNumber: 124,
Timestamp: 45343,
SSRC: 563423,
CSRC: []uint32{},
},
Payload: append(
append([]byte{0x1c, 0x80}, bytes.Repeat([]byte{0x01, 0x02, 0x03, 0x04}, 364)...),
[]byte{0x01, 0x02}...,
),
},
{
Header: rtp.Header{
Version: 2,
Marker: true,
PayloadType: 96,
SequenceNumber: 125,
Timestamp: 45343,
SSRC: 563423,
CSRC: []uint32{},
},
Payload: append(
[]byte{0x1c, 0x40, 0x03, 0x04},
bytes.Repeat([]byte{0x01, 0x02, 0x03, 0x04}, 136)...,
),
},
}
} else {
expected = []*rtp.Packet{
{
Header: rtp.Header{
Version: 2,
Marker: true,
PayloadType: 96,
SequenceNumber: 123,
Timestamp: 45343,
SSRC: 563423,
CSRC: []uint32{},
},
Payload: []byte{0x01, 0x02, 0x03, 0x04},
},
{
Header: rtp.Header{
Version: 2,
Marker: false,
PayloadType: 96,
SequenceNumber: 124,
Timestamp: 45343,
SSRC: 563423,
CSRC: []uint32{},
},
Payload: append(
append([]byte{0x63, 0x02, 0x80, 0x03, 0x04}, bytes.Repeat([]byte{0x01, 0x02, 0x03, 0x04}, 363)...),
[]byte{0x01, 0x02, 0x03}...,
),
},
{
Header: rtp.Header{
Version: 2,
Marker: true,
PayloadType: 96,
SequenceNumber: 125,
Timestamp: 45343,
SSRC: 563423,
CSRC: []uint32{},
},
Payload: append(
[]byte{0x63, 0x02, 0x40, 0x04},
bytes.Repeat([]byte{0x01, 0x02, 0x03, 0x04}, 135)...,
),
},
}
}
c.OnPacketRTP(medias[0], medias[0].Formats[0], func(pkt *rtp.Packet) {
require.Equal(t, expected[i], pkt)
i++
if i >= len(expected) {
close(packetRecv)
}
})
_, err = c.Play(nil)
require.NoError(t, err)
var tosend []*rtp.Packet
if ca == "h264" {
tosend = []*rtp.Packet{
{
Header: rtp.Header{
Version: 2,
Marker: true,
PayloadType: 96,
SequenceNumber: 123,
Timestamp: 45343,
SSRC: 563423,
Padding: true,
},
Payload: []byte{0x01, 0x02, 0x03, 0x04},
},
{
Header: rtp.Header{
Version: 2,
Marker: false,
PayloadType: 96,
SequenceNumber: 124,
Timestamp: 45343,
SSRC: 563423,
Padding: true,
},
Payload: append([]byte{0x1c, 0b10000000}, bytes.Repeat([]byte{0x01, 0x02, 0x03, 0x04}, 2000/4)...),
},
{
Header: rtp.Header{
Version: 2,
Marker: true,
PayloadType: 96,
SequenceNumber: 125,
Timestamp: 45343,
SSRC: 563423,
Padding: true,
},
Payload: []byte{0x1c, 0b01000000, 0x01, 0x02, 0x03, 0x04},
},
}
} else {
tosend = []*rtp.Packet{
{
Header: rtp.Header{
Version: 2,
Marker: true,
PayloadType: 96,
SequenceNumber: 123,
Timestamp: 45343,
SSRC: 563423,
Padding: true,
},
Payload: []byte{0x01, 0x02, 0x03, 0x04},
},
{
Header: rtp.Header{
Version: 2,
Marker: true,
PayloadType: 96,
SequenceNumber: 124,
Timestamp: 45343,
SSRC: 563423,
Padding: true,
},
Payload: bytes.Repeat([]byte{0x01, 0x02, 0x03, 0x04}, 2000/4),
},
}
}
for _, pkt := range tosend {
byts, _ := pkt.Marshal()
err = conn.WriteInterleavedFrame(&base.InterleavedFrame{
Channel: 0,
Payload: byts,
}, make([]byte, 2048))
require.NoError(t, err)
}
<-packetRecv
})
}
}

51
internal/core/hls_muxer.go
Unescape View File

@ -19,6 +19,7 @@ import (
"github.com/gin-gonic/gin" "github.com/gin-gonic/gin"
"github.com/aler9/rtsp-simple-server/internal/conf" "github.com/aler9/rtsp-simple-server/internal/conf"
"github.com/aler9/rtsp-simple-server/internal/formatprocessor"
"github.com/aler9/rtsp-simple-server/internal/hls" "github.com/aler9/rtsp-simple-server/internal/hls"
"github.com/aler9/rtsp-simple-server/internal/logger" "github.com/aler9/rtsp-simple-server/internal/logger"
) )
@ -321,21 +322,21 @@ func (m *hlsMuxer) setupVideoMedia(stream *stream) (*media.Media, format.Format)
videoStartPTSFilled := false videoStartPTSFilled := false
var videoStartPTS time.Duration var videoStartPTS time.Duration
stream.readerAdd(m, videoMedia, videoFormatH265, func(dat data) { stream.readerAdd(m, videoMedia, videoFormatH265, func(dat formatprocessor.Data) {
m.ringBuffer.Push(func() error { m.ringBuffer.Push(func() error {
tdata := dat.(*dataH265) tdata := dat.(*formatprocessor.DataH265)
if tdata.au == nil { if tdata.AU == nil {
return nil return nil
} }
if !videoStartPTSFilled { if !videoStartPTSFilled {
videoStartPTSFilled = true videoStartPTSFilled = true
videoStartPTS = tdata.pts videoStartPTS = tdata.PTS
} }
pts := tdata.pts - videoStartPTS pts := tdata.PTS - videoStartPTS
err := m.muxer.WriteH26x(tdata.ntp, pts, tdata.au) err := m.muxer.WriteH26x(tdata.NTP, pts, tdata.AU)
if err != nil { if err != nil {
return fmt.Errorf("muxer error: %v", err) return fmt.Errorf("muxer error: %v", err)
} }
@ -354,21 +355,21 @@ func (m *hlsMuxer) setupVideoMedia(stream *stream) (*media.Media, format.Format)
videoStartPTSFilled := false videoStartPTSFilled := false
var videoStartPTS time.Duration var videoStartPTS time.Duration
stream.readerAdd(m, videoMedia, videoFormatH264, func(dat data) { stream.readerAdd(m, videoMedia, videoFormatH264, func(dat formatprocessor.Data) {
m.ringBuffer.Push(func() error { m.ringBuffer.Push(func() error {
tdata := dat.(*dataH264) tdata := dat.(*formatprocessor.DataH264)
if tdata.au == nil { if tdata.AU == nil {
return nil return nil
} }
if !videoStartPTSFilled { if !videoStartPTSFilled {
videoStartPTSFilled = true videoStartPTSFilled = true
videoStartPTS = tdata.pts videoStartPTS = tdata.PTS
} }
pts := tdata.pts - videoStartPTS pts := tdata.PTS - videoStartPTS
err := m.muxer.WriteH26x(tdata.ntp, pts, tdata.au) err := m.muxer.WriteH26x(tdata.NTP, pts, tdata.AU)
if err != nil { if err != nil {
return fmt.Errorf("muxer error: %v", err) return fmt.Errorf("muxer error: %v", err)
} }
@ -391,23 +392,23 @@ func (m *hlsMuxer) setupAudioMedia(stream *stream) (*media.Media, format.Format)
audioStartPTSFilled := false audioStartPTSFilled := false
var audioStartPTS time.Duration var audioStartPTS time.Duration
stream.readerAdd(m, audioMedia, audioFormatMPEG4Audio, func(dat data) { stream.readerAdd(m, audioMedia, audioFormatMPEG4Audio, func(dat formatprocessor.Data) {
m.ringBuffer.Push(func() error { m.ringBuffer.Push(func() error {
tdata := dat.(*dataMPEG4Audio) tdata := dat.(*formatprocessor.DataMPEG4Audio)
if tdata.aus == nil { if tdata.AUs == nil {
return nil return nil
} }
if !audioStartPTSFilled { if !audioStartPTSFilled {
audioStartPTSFilled = true audioStartPTSFilled = true
audioStartPTS = tdata.pts audioStartPTS = tdata.PTS
} }
pts := tdata.pts - audioStartPTS pts := tdata.PTS - audioStartPTS
for i, au := range tdata.aus { for i, au := range tdata.AUs {
err := m.muxer.WriteAudio( err := m.muxer.WriteAudio(
tdata.ntp, tdata.NTP,
pts+time.Duration(i)*mpeg4audio.SamplesPerAccessUnit* pts+time.Duration(i)*mpeg4audio.SamplesPerAccessUnit*
time.Second/time.Duration(audioFormatMPEG4Audio.ClockRate()), time.Second/time.Duration(audioFormatMPEG4Audio.ClockRate()),
au) au)
@ -430,20 +431,20 @@ func (m *hlsMuxer) setupAudioMedia(stream *stream) (*media.Media, format.Format)
audioStartPTSFilled := false audioStartPTSFilled := false
var audioStartPTS time.Duration var audioStartPTS time.Duration
stream.readerAdd(m, audioMedia, audioFormatOpus, func(dat data) { stream.readerAdd(m, audioMedia, audioFormatOpus, func(dat formatprocessor.Data) {
m.ringBuffer.Push(func() error { m.ringBuffer.Push(func() error {
tdata := dat.(*dataOpus) tdata := dat.(*formatprocessor.DataOpus)
if !audioStartPTSFilled { if !audioStartPTSFilled {
audioStartPTSFilled = true audioStartPTSFilled = true
audioStartPTS = tdata.pts audioStartPTS = tdata.PTS
} }
pts := tdata.pts - audioStartPTS pts := tdata.PTS - audioStartPTS
err := m.muxer.WriteAudio( err := m.muxer.WriteAudio(
tdata.ntp, tdata.NTP,
pts, pts,
tdata.frame) tdata.Frame)
if err != nil { if err != nil {
return fmt.Errorf("muxer error: %v", err) return fmt.Errorf("muxer error: %v", err)
} }

17
internal/core/hls_source.go
Unescape View File

@ -7,6 +7,7 @@ import (
"github.com/aler9/gortsplib/v2/pkg/format" "github.com/aler9/gortsplib/v2/pkg/format"
"github.com/aler9/gortsplib/v2/pkg/media" "github.com/aler9/gortsplib/v2/pkg/media"
"github.com/aler9/rtsp-simple-server/internal/formatprocessor"
"github.com/aler9/rtsp-simple-server/internal/hls" "github.com/aler9/rtsp-simple-server/internal/hls"
"github.com/aler9/rtsp-simple-server/internal/logger" "github.com/aler9/rtsp-simple-server/internal/logger"
) )
@ -85,10 +86,10 @@ func (s *hlsSource) run(ctx context.Context) error {
} }
onVideoData := func(pts time.Duration, au [][]byte) { onVideoData := func(pts time.Duration, au [][]byte) {
err := stream.writeData(videoMedia, videoMedia.Formats[0], &dataH264{ err := stream.writeData(videoMedia, videoMedia.Formats[0], &formatprocessor.DataH264{
pts: pts, PTS: pts,
au: au, AU: au,
ntp: time.Now(), NTP: time.Now(),
}) })
if err != nil { if err != nil {
s.Log(logger.Warn, "%v", err) s.Log(logger.Warn, "%v", err)
@ -96,10 +97,10 @@ func (s *hlsSource) run(ctx context.Context) error {
} }
onAudioData := func(pts time.Duration, au []byte) { onAudioData := func(pts time.Duration, au []byte) {
err := stream.writeData(audioMedia, audioMedia.Formats[0], &dataMPEG4Audio{ err := stream.writeData(audioMedia, audioMedia.Formats[0], &formatprocessor.DataMPEG4Audio{
pts: pts, PTS: pts,
aus: [][]byte{au}, AUs: [][]byte{au},
ntp: time.Now(), NTP: time.Now(),
}) })
if err != nil { if err != nil {
s.Log(logger.Warn, "%v", err) s.Log(logger.Warn, "%v", err)

9
internal/core/rpicamera_source.go
Unescape View File

@ -7,6 +7,7 @@ import (
"github.com/aler9/gortsplib/v2/pkg/format" "github.com/aler9/gortsplib/v2/pkg/format"
"github.com/aler9/gortsplib/v2/pkg/media" "github.com/aler9/gortsplib/v2/pkg/media"
"github.com/aler9/rtsp-simple-server/internal/formatprocessor"
"github.com/aler9/rtsp-simple-server/internal/logger" "github.com/aler9/rtsp-simple-server/internal/logger"
"github.com/aler9/rtsp-simple-server/internal/rpicamera" "github.com/aler9/rtsp-simple-server/internal/rpicamera"
) )
@ -62,10 +63,10 @@ func (s *rpiCameraSource) run(ctx context.Context) error {
stream = res.stream stream = res.stream
} }
err := stream.writeData(medi, medi.Formats[0], &dataH264{ err := stream.writeData(medi, medi.Formats[0], &formatprocessor.DataH264{
pts: dts, PTS: dts,
au: au, AU: au,
ntp: time.Now(), NTP: time.Now(),
}) })
if err != nil { if err != nil {
s.Log(logger.Warn, "%v", err) s.Log(logger.Warn, "%v", err)

63
internal/core/rtmp_conn.go
Unescape View File

@ -20,6 +20,7 @@ import (
"github.com/aler9/rtsp-simple-server/internal/conf" "github.com/aler9/rtsp-simple-server/internal/conf"
"github.com/aler9/rtsp-simple-server/internal/externalcmd" "github.com/aler9/rtsp-simple-server/internal/externalcmd"
"github.com/aler9/rtsp-simple-server/internal/formatprocessor"
"github.com/aler9/rtsp-simple-server/internal/logger" "github.com/aler9/rtsp-simple-server/internal/logger"
"github.com/aler9/rtsp-simple-server/internal/rtmp" "github.com/aler9/rtsp-simple-server/internal/rtmp"
"github.com/aler9/rtsp-simple-server/internal/rtmp/h264conf" "github.com/aler9/rtsp-simple-server/internal/rtmp/h264conf"
@ -277,24 +278,24 @@ func (c *rtmpConn) runRead(ctx context.Context, u *url.URL) error {
var videoStartPTS time.Duration var videoStartPTS time.Duration
var videoDTSExtractor *h264.DTSExtractor var videoDTSExtractor *h264.DTSExtractor
res.stream.readerAdd(c, videoMedia, videoFormat, func(dat data) { res.stream.readerAdd(c, videoMedia, videoFormat, func(dat formatprocessor.Data) {
ringBuffer.Push(func() error { ringBuffer.Push(func() error {
tdata := dat.(*dataH264) tdata := dat.(*formatprocessor.DataH264)
if tdata.au == nil { if tdata.AU == nil {
return nil return nil
} }
if !videoStartPTSFilled { if !videoStartPTSFilled {
videoStartPTSFilled = true videoStartPTSFilled = true
videoStartPTS = tdata.pts videoStartPTS = tdata.PTS
} }
pts := tdata.pts - videoStartPTS pts := tdata.PTS - videoStartPTS
idrPresent := false idrPresent := false
nonIDRPresent := false nonIDRPresent := false
for _, nalu := range tdata.au { for _, nalu := range tdata.AU {
typ := h264.NALUType(nalu[0] & 0x1F) typ := h264.NALUType(nalu[0] & 0x1F)
switch typ { switch typ {
case h264.NALUTypeIDR: case h264.NALUTypeIDR:
@ -317,7 +318,7 @@ func (c *rtmpConn) runRead(ctx context.Context, u *url.URL) error {
videoDTSExtractor = h264.NewDTSExtractor() videoDTSExtractor = h264.NewDTSExtractor()
var err error var err error
dts, err = videoDTSExtractor.Extract(tdata.au, pts) dts, err = videoDTSExtractor.Extract(tdata.AU, pts)
if err != nil { if err != nil {
return err return err
} }
@ -331,7 +332,7 @@ func (c *rtmpConn) runRead(ctx context.Context, u *url.URL) error {
} }
var err error var err error
dts, err = videoDTSExtractor.Extract(tdata.au, pts) dts, err = videoDTSExtractor.Extract(tdata.AU, pts)
if err != nil { if err != nil {
return err return err
} }
@ -340,7 +341,7 @@ func (c *rtmpConn) runRead(ctx context.Context, u *url.URL) error {
pts -= videoStartDTS pts -= videoStartDTS
} }
avcc, err := h264.AVCCMarshal(tdata.au) avcc, err := h264.AVCCMarshal(tdata.AU)
if err != nil { if err != nil {
return err return err
} }
@ -370,19 +371,19 @@ func (c *rtmpConn) runRead(ctx context.Context, u *url.URL) error {
audioStartPTSFilled := false audioStartPTSFilled := false
var audioStartPTS time.Duration var audioStartPTS time.Duration
res.stream.readerAdd(c, audioMedia, audioFormat, func(dat data) { res.stream.readerAdd(c, audioMedia, audioFormat, func(dat formatprocessor.Data) {
ringBuffer.Push(func() error { ringBuffer.Push(func() error {
tdata := dat.(*dataMPEG4Audio) tdata := dat.(*formatprocessor.DataMPEG4Audio)
if tdata.aus == nil { if tdata.AUs == nil {
return nil return nil
} }
if !audioStartPTSFilled { if !audioStartPTSFilled {
audioStartPTSFilled = true audioStartPTSFilled = true
audioStartPTS = tdata.pts audioStartPTS = tdata.PTS
} }
pts := tdata.pts - audioStartPTS pts := tdata.PTS - audioStartPTS
if videoFormat != nil { if videoFormat != nil {
if !videoFirstIDRFound { if !videoFirstIDRFound {
@ -395,7 +396,7 @@ func (c *rtmpConn) runRead(ctx context.Context, u *url.URL) error {
} }
} }
for i, au := range tdata.aus { for i, au := range tdata.AUs {
c.nconn.SetWriteDeadline(time.Now().Add(time.Duration(c.writeTimeout))) c.nconn.SetWriteDeadline(time.Now().Add(time.Duration(c.writeTimeout)))
err := c.conn.WriteMessage(&message.MsgAudio{ err := c.conn.WriteMessage(&message.MsgAudio{
ChunkStreamID: message.MsgAudioChunkStreamID, ChunkStreamID: message.MsgAudioChunkStreamID,
@ -539,10 +540,10 @@ func (c *rtmpConn) runPublish(ctx context.Context, u *url.URL) error {
if _, ok := videoFormat.(*format.H264); ok { if _, ok := videoFormat.(*format.H264); ok {
onVideoData = func(pts time.Duration, au [][]byte) { onVideoData = func(pts time.Duration, au [][]byte) {
err = rres.stream.writeData(videoMedia, videoFormat, &dataH264{ err = rres.stream.writeData(videoMedia, videoFormat, &formatprocessor.DataH264{
pts: pts, PTS: pts,
au: au, AU: au,
ntp: time.Now(), NTP: time.Now(),
}) })
if err != nil { if err != nil {
c.log(logger.Warn, "%v", err) c.log(logger.Warn, "%v", err)
@ -550,10 +551,10 @@ func (c *rtmpConn) runPublish(ctx context.Context, u *url.URL) error {
} }
} else { } else {
onVideoData = func(pts time.Duration, au [][]byte) { onVideoData = func(pts time.Duration, au [][]byte) {
err = rres.stream.writeData(videoMedia, videoFormat, &dataH265{ err = rres.stream.writeData(videoMedia, videoFormat, &formatprocessor.DataH265{
pts: pts, PTS: pts,
au: au, AU: au,
ntp: time.Now(), NTP: time.Now(),
}) })
if err != nil { if err != nil {
c.log(logger.Warn, "%v", err) c.log(logger.Warn, "%v", err)
@ -582,10 +583,10 @@ func (c *rtmpConn) runPublish(ctx context.Context, u *url.URL) error {
conf.PPS, conf.PPS,
} }
err := rres.stream.writeData(videoMedia, videoFormat, &dataH264{ err := rres.stream.writeData(videoMedia, videoFormat, &formatprocessor.DataH264{
pts: tmsg.DTS + tmsg.PTSDelta, PTS: tmsg.DTS + tmsg.PTSDelta,
au: au, AU: au,
ntp: time.Now(), NTP: time.Now(),
}) })
if err != nil { if err != nil {
c.log(logger.Warn, "%v", err) c.log(logger.Warn, "%v", err)
@ -629,10 +630,10 @@ func (c *rtmpConn) runPublish(ctx context.Context, u *url.URL) error {
return fmt.Errorf("received an audio packet, but track is not set up") return fmt.Errorf("received an audio packet, but track is not set up")
} }
err := rres.stream.writeData(audioMedia, audioFormat, &dataMPEG4Audio{ err := rres.stream.writeData(audioMedia, audioFormat, &formatprocessor.DataMPEG4Audio{
pts: tmsg.DTS, PTS: tmsg.DTS,
aus: [][]byte{tmsg.Payload}, AUs: [][]byte{tmsg.Payload},
ntp: time.Now(), NTP: time.Now(),
}) })
if err != nil { if err != nil {
c.log(logger.Warn, "%v", err) c.log(logger.Warn, "%v", err)

17
internal/core/rtmp_source.go
Unescape View File

@ -17,6 +17,7 @@ import (
"github.com/notedit/rtmp/format/flv/flvio" "github.com/notedit/rtmp/format/flv/flvio"
"github.com/aler9/rtsp-simple-server/internal/conf" "github.com/aler9/rtsp-simple-server/internal/conf"
"github.com/aler9/rtsp-simple-server/internal/formatprocessor"
"github.com/aler9/rtsp-simple-server/internal/logger" "github.com/aler9/rtsp-simple-server/internal/logger"
"github.com/aler9/rtsp-simple-server/internal/rtmp" "github.com/aler9/rtsp-simple-server/internal/rtmp"
"github.com/aler9/rtsp-simple-server/internal/rtmp/message" "github.com/aler9/rtsp-simple-server/internal/rtmp/message"
@ -181,10 +182,10 @@ func (s *rtmpSource) run(ctx context.Context) error {
return fmt.Errorf("unable to decode AVCC: %v", err) return fmt.Errorf("unable to decode AVCC: %v", err)
} }
err = res.stream.writeData(videoMedia, videoFormat, &dataH264{ err = res.stream.writeData(videoMedia, videoFormat, &formatprocessor.DataH264{
pts: tmsg.DTS + tmsg.PTSDelta, PTS: tmsg.DTS + tmsg.PTSDelta,
au: au, AU: au,
ntp: time.Now(), NTP: time.Now(),
}) })
if err != nil { if err != nil {
s.Log(logger.Warn, "%v", err) s.Log(logger.Warn, "%v", err)
@ -197,10 +198,10 @@ func (s *rtmpSource) run(ctx context.Context) error {
return fmt.Errorf("received an AAC packet, but track is not set up") return fmt.Errorf("received an AAC packet, but track is not set up")
} }
err := res.stream.writeData(audioMedia, audioFormat, &dataMPEG4Audio{ err := res.stream.writeData(audioMedia, audioFormat, &formatprocessor.DataMPEG4Audio{
pts: tmsg.DTS, PTS: tmsg.DTS,
aus: [][]byte{tmsg.Payload}, AUs: [][]byte{tmsg.Payload},
ntp: time.Now(), NTP: time.Now(),
}) })
if err != nil { if err != nil {
s.Log(logger.Warn, "%v", err) s.Log(logger.Warn, "%v", err)

31
internal/core/rtsp_server_test.go
Unescape View File

@ -4,43 +4,12 @@ import (
"testing" "testing"
"github.com/aler9/gortsplib/v2" "github.com/aler9/gortsplib/v2"
"github.com/aler9/gortsplib/v2/pkg/base"
"github.com/aler9/gortsplib/v2/pkg/conn"
"github.com/aler9/gortsplib/v2/pkg/media" "github.com/aler9/gortsplib/v2/pkg/media"
"github.com/aler9/gortsplib/v2/pkg/sdp"
"github.com/aler9/gortsplib/v2/pkg/url" "github.com/aler9/gortsplib/v2/pkg/url"
"github.com/pion/rtp" "github.com/pion/rtp"
"github.com/stretchr/testify/require" "github.com/stretchr/testify/require"
) )
func mustParseURL(s string) *url.URL {
u, err := url.Parse(s)
if err != nil {
panic(err)
}
return u
}
func mustMarshalSDP(sdp *sdp.SessionDescription) []byte {
byts, err := sdp.Marshal()
if err != nil {
panic(err)
}
return byts
}
func writeReqReadRes(
conn *conn.Conn,
req base.Request,
) (*base.Response, error) {
err := conn.WriteRequest(&req)
if err != nil {
return nil, err
}
return conn.ReadResponse()
}
func TestRTSPServerAuth(t *testing.T) { func TestRTSPServerAuth(t *testing.T) {
for _, ca := range []string{ for _, ca := range []string{
"internal", "internal",

43
internal/core/rtsp_session.go
Unescape View File

@ -17,6 +17,7 @@ import (
"github.com/aler9/rtsp-simple-server/internal/conf" "github.com/aler9/rtsp-simple-server/internal/conf"
"github.com/aler9/rtsp-simple-server/internal/externalcmd" "github.com/aler9/rtsp-simple-server/internal/externalcmd"
"github.com/aler9/rtsp-simple-server/internal/formatprocessor"
"github.com/aler9/rtsp-simple-server/internal/logger" "github.com/aler9/rtsp-simple-server/internal/logger"
) )
@ -310,9 +311,9 @@ func (s *rtspSession) onRecord(ctx *gortsplib.ServerHandlerOnRecordCtx) (*base.R
switch forma.(type) { switch forma.(type) {
case *format.H264: case *format.H264:
ctx.Session.OnPacketRTP(medi, forma, func(pkt *rtp.Packet) { ctx.Session.OnPacketRTP(medi, forma, func(pkt *rtp.Packet) {
err := s.stream.writeData(cmedia, cformat, &dataH264{ err := s.stream.writeData(cmedia, cformat, &formatprocessor.DataH264{
rtpPackets: []*rtp.Packet{pkt}, RTPPackets: []*rtp.Packet{pkt},
ntp: time.Now(), NTP: time.Now(),
}) })
if err != nil { if err != nil {
s.log(logger.Warn, "%v", err) s.log(logger.Warn, "%v", err)
@ -321,9 +322,9 @@ func (s *rtspSession) onRecord(ctx *gortsplib.ServerHandlerOnRecordCtx) (*base.R
case *format.H265: case *format.H265:
ctx.Session.OnPacketRTP(medi, forma, func(pkt *rtp.Packet) { ctx.Session.OnPacketRTP(medi, forma, func(pkt *rtp.Packet) {
err := s.stream.writeData(cmedia, cformat, &dataH265{ err := s.stream.writeData(cmedia, cformat, &formatprocessor.DataH265{
rtpPackets: []*rtp.Packet{pkt}, RTPPackets: []*rtp.Packet{pkt},
ntp: time.Now(), NTP: time.Now(),
}) })
if err != nil { if err != nil {
s.log(logger.Warn, "%v", err) s.log(logger.Warn, "%v", err)
@ -332,9 +333,9 @@ func (s *rtspSession) onRecord(ctx *gortsplib.ServerHandlerOnRecordCtx) (*base.R
case *format.VP8: case *format.VP8:
ctx.Session.OnPacketRTP(medi, forma, func(pkt *rtp.Packet) { ctx.Session.OnPacketRTP(medi, forma, func(pkt *rtp.Packet) {
err := s.stream.writeData(cmedia, cformat, &dataVP8{ err := s.stream.writeData(cmedia, cformat, &formatprocessor.DataVP8{
rtpPackets: []*rtp.Packet{pkt}, RTPPackets: []*rtp.Packet{pkt},
ntp: time.Now(), NTP: time.Now(),
}) })
if err != nil { if err != nil {
s.log(logger.Warn, "%v", err) s.log(logger.Warn, "%v", err)
@ -343,9 +344,9 @@ func (s *rtspSession) onRecord(ctx *gortsplib.ServerHandlerOnRecordCtx) (*base.R
case *format.VP9: case *format.VP9:
ctx.Session.OnPacketRTP(medi, forma, func(pkt *rtp.Packet) { ctx.Session.OnPacketRTP(medi, forma, func(pkt *rtp.Packet) {
err := s.stream.writeData(cmedia, cformat, &dataVP9{ err := s.stream.writeData(cmedia, cformat, &formatprocessor.DataVP9{
rtpPackets: []*rtp.Packet{pkt}, RTPPackets: []*rtp.Packet{pkt},
ntp: time.Now(), NTP: time.Now(),
}) })
if err != nil { if err != nil {
s.log(logger.Warn, "%v", err) s.log(logger.Warn, "%v", err)
@ -354,9 +355,9 @@ func (s *rtspSession) onRecord(ctx *gortsplib.ServerHandlerOnRecordCtx) (*base.R
case *format.MPEG4Audio: case *format.MPEG4Audio:
ctx.Session.OnPacketRTP(medi, forma, func(pkt *rtp.Packet) { ctx.Session.OnPacketRTP(medi, forma, func(pkt *rtp.Packet) {
err := s.stream.writeData(cmedia, cformat, &dataMPEG4Audio{ err := s.stream.writeData(cmedia, cformat, &formatprocessor.DataMPEG4Audio{
rtpPackets: []*rtp.Packet{pkt}, RTPPackets: []*rtp.Packet{pkt},
ntp: time.Now(), NTP: time.Now(),
}) })
if err != nil { if err != nil {
s.log(logger.Warn, "%v", err) s.log(logger.Warn, "%v", err)
@ -365,9 +366,9 @@ func (s *rtspSession) onRecord(ctx *gortsplib.ServerHandlerOnRecordCtx) (*base.R
case *format.Opus: case *format.Opus:
ctx.Session.OnPacketRTP(medi, forma, func(pkt *rtp.Packet) { ctx.Session.OnPacketRTP(medi, forma, func(pkt *rtp.Packet) {
err := s.stream.writeData(cmedia, cformat, &dataOpus{ err := s.stream.writeData(cmedia, cformat, &formatprocessor.DataOpus{
rtpPackets: []*rtp.Packet{pkt}, RTPPackets: []*rtp.Packet{pkt},
ntp: time.Now(), NTP: time.Now(),
}) })
if err != nil { if err != nil {
s.log(logger.Warn, "%v", err) s.log(logger.Warn, "%v", err)
@ -376,9 +377,9 @@ func (s *rtspSession) onRecord(ctx *gortsplib.ServerHandlerOnRecordCtx) (*base.R
default: default:
ctx.Session.OnPacketRTP(medi, forma, func(pkt *rtp.Packet) { ctx.Session.OnPacketRTP(medi, forma, func(pkt *rtp.Packet) {
err := s.stream.writeData(cmedia, cformat, &dataGeneric{ err := s.stream.writeData(cmedia, cformat, &formatprocessor.DataGeneric{
rtpPackets: []*rtp.Packet{pkt}, RTPPackets: []*rtp.Packet{pkt},
ntp: time.Now(), NTP: time.Now(),
}) })
if err != nil { if err != nil {
s.log(logger.Warn, "%v", err) s.log(logger.Warn, "%v", err)

43
internal/core/rtsp_source.go
Unescape View File

@ -16,6 +16,7 @@ import (
"github.com/aler9/gortsplib/v2/pkg/url" "github.com/aler9/gortsplib/v2/pkg/url"
"github.com/aler9/rtsp-simple-server/internal/conf" "github.com/aler9/rtsp-simple-server/internal/conf"
"github.com/aler9/rtsp-simple-server/internal/formatprocessor"
"github.com/aler9/rtsp-simple-server/internal/logger" "github.com/aler9/rtsp-simple-server/internal/logger"
) )
@ -150,9 +151,9 @@ func (s *rtspSource) run(ctx context.Context) error {
switch forma.(type) { switch forma.(type) {
case *format.H264: case *format.H264:
c.OnPacketRTP(medi, forma, func(pkt *rtp.Packet) { c.OnPacketRTP(medi, forma, func(pkt *rtp.Packet) {
err := res.stream.writeData(cmedia, cformat, &dataH264{ err := res.stream.writeData(cmedia, cformat, &formatprocessor.DataH264{
rtpPackets: []*rtp.Packet{pkt}, RTPPackets: []*rtp.Packet{pkt},
ntp: time.Now(), NTP: time.Now(),
}) })
if err != nil { if err != nil {
s.Log(logger.Warn, "%v", err) s.Log(logger.Warn, "%v", err)
@ -161,9 +162,9 @@ func (s *rtspSource) run(ctx context.Context) error {
case *format.H265: case *format.H265:
c.OnPacketRTP(medi, forma, func(pkt *rtp.Packet) { c.OnPacketRTP(medi, forma, func(pkt *rtp.Packet) {
err := res.stream.writeData(cmedia, cformat, &dataH265{ err := res.stream.writeData(cmedia, cformat, &formatprocessor.DataH265{
rtpPackets: []*rtp.Packet{pkt}, RTPPackets: []*rtp.Packet{pkt},
ntp: time.Now(), NTP: time.Now(),
}) })
if err != nil { if err != nil {
s.Log(logger.Warn, "%v", err) s.Log(logger.Warn, "%v", err)
@ -172,9 +173,9 @@ func (s *rtspSource) run(ctx context.Context) error {
case *format.VP8: case *format.VP8:
c.OnPacketRTP(medi, forma, func(pkt *rtp.Packet) { c.OnPacketRTP(medi, forma, func(pkt *rtp.Packet) {
err := res.stream.writeData(cmedia, cformat, &dataVP8{ err := res.stream.writeData(cmedia, cformat, &formatprocessor.DataVP8{
rtpPackets: []*rtp.Packet{pkt}, RTPPackets: []*rtp.Packet{pkt},
ntp: time.Now(), NTP: time.Now(),
}) })
if err != nil { if err != nil {
s.Log(logger.Warn, "%v", err) s.Log(logger.Warn, "%v", err)
@ -183,9 +184,9 @@ func (s *rtspSource) run(ctx context.Context) error {
case *format.VP9: case *format.VP9:
c.OnPacketRTP(medi, forma, func(pkt *rtp.Packet) { c.OnPacketRTP(medi, forma, func(pkt *rtp.Packet) {
err := res.stream.writeData(cmedia, cformat, &dataVP9{ err := res.stream.writeData(cmedia, cformat, &formatprocessor.DataVP9{
rtpPackets: []*rtp.Packet{pkt}, RTPPackets: []*rtp.Packet{pkt},
ntp: time.Now(), NTP: time.Now(),
}) })
if err != nil { if err != nil {
s.Log(logger.Warn, "%v", err) s.Log(logger.Warn, "%v", err)
@ -194,9 +195,9 @@ func (s *rtspSource) run(ctx context.Context) error {
case *format.MPEG4Audio: case *format.MPEG4Audio:
c.OnPacketRTP(medi, forma, func(pkt *rtp.Packet) { c.OnPacketRTP(medi, forma, func(pkt *rtp.Packet) {
err := res.stream.writeData(cmedia, cformat, &dataMPEG4Audio{ err := res.stream.writeData(cmedia, cformat, &formatprocessor.DataMPEG4Audio{
rtpPackets: []*rtp.Packet{pkt}, RTPPackets: []*rtp.Packet{pkt},
ntp: time.Now(), NTP: time.Now(),
}) })
if err != nil { if err != nil {
s.Log(logger.Warn, "%v", err) s.Log(logger.Warn, "%v", err)
@ -205,9 +206,9 @@ func (s *rtspSource) run(ctx context.Context) error {
case *format.Opus: case *format.Opus:
c.OnPacketRTP(medi, forma, func(pkt *rtp.Packet) { c.OnPacketRTP(medi, forma, func(pkt *rtp.Packet) {
err := res.stream.writeData(cmedia, cformat, &dataOpus{ err := res.stream.writeData(cmedia, cformat, &formatprocessor.DataOpus{
rtpPackets: []*rtp.Packet{pkt}, RTPPackets: []*rtp.Packet{pkt},
ntp: time.Now(), NTP: time.Now(),
}) })
if err != nil { if err != nil {
s.Log(logger.Warn, "%v", err) s.Log(logger.Warn, "%v", err)
@ -216,9 +217,9 @@ func (s *rtspSource) run(ctx context.Context) error {
default: default:
c.OnPacketRTP(medi, forma, func(pkt *rtp.Packet) { c.OnPacketRTP(medi, forma, func(pkt *rtp.Packet) {
err := res.stream.writeData(cmedia, cformat, &dataGeneric{ err := res.stream.writeData(cmedia, cformat, &formatprocessor.DataGeneric{
rtpPackets: []*rtp.Packet{pkt}, RTPPackets: []*rtp.Packet{pkt},
ntp: time.Now(), NTP: time.Now(),
}) })
if err != nil { if err != nil {
s.Log(logger.Warn, "%v", err) s.Log(logger.Warn, "%v", err)

6
internal/core/stream.go
Unescape View File

@ -4,6 +4,8 @@ import (
"github.com/aler9/gortsplib/v2" "github.com/aler9/gortsplib/v2"
"github.com/aler9/gortsplib/v2/pkg/format" "github.com/aler9/gortsplib/v2/pkg/format"
"github.com/aler9/gortsplib/v2/pkg/media" "github.com/aler9/gortsplib/v2/pkg/media"
"github.com/aler9/rtsp-simple-server/internal/formatprocessor"
) )
type stream struct { type stream struct {
@ -43,7 +45,7 @@ func (s *stream) medias() media.Medias {
return s.rtspStream.Medias() return s.rtspStream.Medias()
} }
func (s *stream) readerAdd(r reader, medi *media.Media, forma format.Format, cb func(data)) { func (s *stream) readerAdd(r reader, medi *media.Media, forma format.Format, cb func(formatprocessor.Data)) {
sm := s.smedias[medi] sm := s.smedias[medi]
sf := sm.formats[forma] sf := sm.formats[forma]
sf.readerAdd(r, cb) sf.readerAdd(r, cb)
@ -57,7 +59,7 @@ func (s *stream) readerRemove(r reader) {
} }
} }
func (s *stream) writeData(medi *media.Media, forma format.Format, data data) error { func (s *stream) writeData(medi *media.Media, forma format.Format, data formatprocessor.Data) error {
sm := s.smedias[medi] sm := s.smedias[medi]
sf := sm.formats[forma] sf := sm.formats[forma]
return sf.writeData(s, medi, data) return sf.writeData(s, medi, data)

20
internal/core/streamformat.go
Unescape View File

@ -6,29 +6,31 @@ import (
"github.com/aler9/gortsplib/v2/pkg/format" "github.com/aler9/gortsplib/v2/pkg/format"
"github.com/aler9/gortsplib/v2/pkg/media" "github.com/aler9/gortsplib/v2/pkg/media"
"github.com/aler9/rtsp-simple-server/internal/formatprocessor"
) )
type streamFormat struct { type streamFormat struct {
proc formatProcessor proc formatprocessor.Processor
mutex sync.RWMutex mutex sync.RWMutex
nonRTSPReaders map[reader]func(data) nonRTSPReaders map[reader]func(formatprocessor.Data)
} }
func newStreamFormat(forma format.Format, generateRTPPackets bool) (*streamFormat, error) { func newStreamFormat(forma format.Format, generateRTPPackets bool) (*streamFormat, error) {
proc, err := newFormatProcessor(forma, generateRTPPackets) proc, err := formatprocessor.New(forma, generateRTPPackets)
if err != nil { if err != nil {
return nil, err return nil, err
} }
sf := &streamFormat{ sf := &streamFormat{
proc: proc, proc: proc,
nonRTSPReaders: make(map[reader]func(data)), nonRTSPReaders: make(map[reader]func(formatprocessor.Data)),
} }
return sf, nil return sf, nil
} }
func (sf *streamFormat) readerAdd(r reader, cb func(data)) { func (sf *streamFormat) readerAdd(r reader, cb func(formatprocessor.Data)) {
sf.mutex.Lock() sf.mutex.Lock()
defer sf.mutex.Unlock() defer sf.mutex.Unlock()
sf.nonRTSPReaders[r] = cb sf.nonRTSPReaders[r] = cb
@ -40,21 +42,21 @@ func (sf *streamFormat) readerRemove(r reader) {
delete(sf.nonRTSPReaders, r) delete(sf.nonRTSPReaders, r)
} }
func (sf *streamFormat) writeData(s *stream, medi *media.Media, data data) error { func (sf *streamFormat) writeData(s *stream, medi *media.Media, data formatprocessor.Data) error {
sf.mutex.RLock() sf.mutex.RLock()
defer sf.mutex.RUnlock() defer sf.mutex.RUnlock()
hasNonRTSPReaders := len(sf.nonRTSPReaders) > 0 hasNonRTSPReaders := len(sf.nonRTSPReaders) > 0
err := sf.proc.process(data, hasNonRTSPReaders) err := sf.proc.Process(data, hasNonRTSPReaders)
if err != nil { if err != nil {
return err return err
} }
// forward RTP packets to RTSP readers // forward RTP packets to RTSP readers
for _, pkt := range data.getRTPPackets() { for _, pkt := range data.GetRTPPackets() {
atomic.AddUint64(s.bytesReceived, uint64(pkt.MarshalSize())) atomic.AddUint64(s.bytesReceived, uint64(pkt.MarshalSize()))
s.rtspStream.WritePacketRTPWithNTP(medi, pkt, data.getNTP()) s.rtspStream.WritePacketRTPWithNTP(medi, pkt, data.GetNTP())
} }
// forward decoded frames to non-RTSP readers // forward decoded frames to non-RTSP readers

49
internal/core/webrtc_conn.go
Unescape View File

@ -29,6 +29,7 @@ import (
"github.com/pion/webrtc/v3" "github.com/pion/webrtc/v3"
"github.com/aler9/rtsp-simple-server/internal/conf" "github.com/aler9/rtsp-simple-server/internal/conf"
"github.com/aler9/rtsp-simple-server/internal/formatprocessor"
"github.com/aler9/rtsp-simple-server/internal/logger" "github.com/aler9/rtsp-simple-server/internal/logger"
) )
@ -96,7 +97,7 @@ type webRTCTrack struct {
media *media.Media media *media.Media
format format.Format format format.Format
webRTCTrack *webrtc.TrackLocalStaticRTP webRTCTrack *webrtc.TrackLocalStaticRTP
cb func(data, context.Context, chan error) cb func(formatprocessor.Data, context.Context, chan error)
} }
func gatherMedias(tracks []*webRTCTrack) media.Medias { func gatherMedias(tracks []*webRTCTrack) media.Medias {
@ -441,7 +442,7 @@ outer:
for _, track := range tracks { for _, track := range tracks {
ctrack := track ctrack := track
res.stream.readerAdd(c, track.media, track.format, func(dat data) { res.stream.readerAdd(c, track.media, track.format, func(dat formatprocessor.Data) {
ringBuffer.Push(func() { ringBuffer.Push(func() {
ctrack.cb(dat, ctx, writeError) ctrack.cb(dat, ctx, writeError)
}) })
@ -503,14 +504,14 @@ func (c *webRTCConn) allocateTracks(medias media.Medias) ([]*webRTCTrack, error)
media: vp9Media, media: vp9Media,
format: vp9Format, format: vp9Format,
webRTCTrack: webRTCTrak, webRTCTrack: webRTCTrak,
cb: func(dat data, ctx context.Context, writeError chan error) { cb: func(dat formatprocessor.Data, ctx context.Context, writeError chan error) {
tdata := dat.(*dataVP9) tdata := dat.(*formatprocessor.DataVP9)
if tdata.frame == nil { if tdata.Frame == nil {
return return
} }
packets, err := encoder.Encode(tdata.frame, tdata.pts) packets, err := encoder.Encode(tdata.Frame, tdata.PTS)
if err != nil { if err != nil {
return return
} }
@ -550,14 +551,14 @@ func (c *webRTCConn) allocateTracks(medias media.Medias) ([]*webRTCTrack, error)
media: vp8Media, media: vp8Media,
format: vp8Format, format: vp8Format,
webRTCTrack: webRTCTrak, webRTCTrack: webRTCTrak,
cb: func(dat data, ctx context.Context, writeError chan error) { cb: func(dat formatprocessor.Data, ctx context.Context, writeError chan error) {
tdata := dat.(*dataVP8) tdata := dat.(*formatprocessor.DataVP8)
if tdata.frame == nil { if tdata.Frame == nil {
return return
} }
packets, err := encoder.Encode(tdata.frame, tdata.pts) packets, err := encoder.Encode(tdata.Frame, tdata.PTS)
if err != nil { if err != nil {
return return
} }
@ -600,32 +601,32 @@ func (c *webRTCConn) allocateTracks(medias media.Medias) ([]*webRTCTrack, error)
media: h264Media, media: h264Media,
format: h264Format, format: h264Format,
webRTCTrack: webRTCTrak, webRTCTrack: webRTCTrak,
cb: func(dat data, ctx context.Context, writeError chan error) { cb: func(dat formatprocessor.Data, ctx context.Context, writeError chan error) {
tdata := dat.(*dataH264) tdata := dat.(*formatprocessor.DataH264)
if tdata.au == nil { if tdata.AU == nil {
return return
} }
if !firstNALUReceived { if !firstNALUReceived {
if !h264.IDRPresent(tdata.au) { if !h264.IDRPresent(tdata.AU) {
return return
} }
firstNALUReceived = true firstNALUReceived = true
lastPTS = tdata.pts lastPTS = tdata.PTS
} else { } else {
if tdata.pts < lastPTS { if tdata.PTS < lastPTS {
select { select {
case writeError <- fmt.Errorf("WebRTC doesn't support H264 streams with B-frames"): case writeError <- fmt.Errorf("WebRTC doesn't support H264 streams with B-frames"):
case <-ctx.Done(): case <-ctx.Done():
} }
return return
} }
lastPTS = tdata.pts lastPTS = tdata.PTS
} }
packets, err := encoder.Encode(tdata.au, tdata.pts) packets, err := encoder.Encode(tdata.AU, tdata.PTS)
if err != nil { if err != nil {
return return
} }
@ -658,8 +659,8 @@ func (c *webRTCConn) allocateTracks(medias media.Medias) ([]*webRTCTrack, error)
media: opusMedia, media: opusMedia,
format: opusFormat, format: opusFormat,
webRTCTrack: webRTCTrak, webRTCTrack: webRTCTrak,
cb: func(dat data, ctx context.Context, writeError chan error) { cb: func(dat formatprocessor.Data, ctx context.Context, writeError chan error) {
for _, pkt := range dat.getRTPPackets() { for _, pkt := range dat.GetRTPPackets() {
webRTCTrak.WriteRTP(pkt) webRTCTrak.WriteRTP(pkt)
} }
}, },
@ -688,8 +689,8 @@ func (c *webRTCConn) allocateTracks(medias media.Medias) ([]*webRTCTrack, error)
media: g722Media, media: g722Media,
format: g722Format, format: g722Format,
webRTCTrack: webRTCTrak, webRTCTrack: webRTCTrak,
cb: func(dat data, ctx context.Context, writeError chan error) { cb: func(dat formatprocessor.Data, ctx context.Context, writeError chan error) {
for _, pkt := range dat.getRTPPackets() { for _, pkt := range dat.GetRTPPackets() {
webRTCTrak.WriteRTP(pkt) webRTCTrak.WriteRTP(pkt)
} }
}, },
@ -726,8 +727,8 @@ func (c *webRTCConn) allocateTracks(medias media.Medias) ([]*webRTCTrack, error)
media: g711Media, media: g711Media,
format: g711Format, format: g711Format,
webRTCTrack: webRTCTrak, webRTCTrack: webRTCTrak,
cb: func(dat data, ctx context.Context, writeError chan error) { cb: func(dat formatprocessor.Data, ctx context.Context, writeError chan error) {
for _, pkt := range dat.getRTPPackets() { for _, pkt := range dat.GetRTPPackets() {
webRTCTrak.WriteRTP(pkt) webRTCTrak.WriteRTP(pkt)
} }
}, },

27
internal/core/formatprocessor_generic.go → internal/formatprocessor/generic.go
Unescape View File

@ -1,4 +1,4 @@
package core package formatprocessor
import ( import (
"fmt" "fmt"
@ -13,22 +13,25 @@ const (
maxPacketSize = 1472 maxPacketSize = 1472
) )
type dataGeneric struct { // DataGeneric is a generic data unit.
rtpPackets []*rtp.Packet type DataGeneric struct {
ntp time.Time RTPPackets []*rtp.Packet
NTP time.Time
} }
func (d *dataGeneric) getRTPPackets() []*rtp.Packet { // GetRTPPackets implements Data.
return d.rtpPackets func (d *DataGeneric) GetRTPPackets() []*rtp.Packet {
return d.RTPPackets
} }
func (d *dataGeneric) getNTP() time.Time { // GetNTP implements Data.
return d.ntp func (d *DataGeneric) GetNTP() time.Time {
return d.NTP
} }
type formatProcessorGeneric struct{} type formatProcessorGeneric struct{}
func newFormatProcessorGeneric(forma format.Format, generateRTPPackets bool) (*formatProcessorGeneric, error) { func newGeneric(forma format.Format, generateRTPPackets bool) (*formatProcessorGeneric, error) {
if generateRTPPackets { if generateRTPPackets {
return nil, fmt.Errorf("we don't know how to generate RTP packets of format %+v", forma) return nil, fmt.Errorf("we don't know how to generate RTP packets of format %+v", forma)
} }
@ -36,10 +39,10 @@ func newFormatProcessorGeneric(forma format.Format, generateRTPPackets bool) (*f
return &formatProcessorGeneric{}, nil return &formatProcessorGeneric{}, nil
} }
func (t *formatProcessorGeneric) process(dat data, hasNonRTSPReaders bool) error { func (t *formatProcessorGeneric) Process(dat Data, hasNonRTSPReaders bool) error {
tdata := dat.(*dataGeneric) tdata := dat.(*DataGeneric)
pkt := tdata.rtpPackets[0] pkt := tdata.RTPPackets[0]
// remove padding // remove padding
pkt.Header.Padding = false pkt.Header.Padding = false

51
internal/formatprocessor/generic_test.go
Unescape View File

@ -0,0 +1,51 @@
package formatprocessor
import (
"testing"
"github.com/aler9/gortsplib/v2/pkg/format"
"github.com/pion/rtp"
"github.com/stretchr/testify/require"
)
func TestGenericRemovePadding(t *testing.T) {
forma := &format.Generic{
PayloadTyp: 96,
RTPMap: "private/90000",
}
forma.Init()
p, err := New(forma, false)
require.NoError(t, err)
pkt := &rtp.Packet{
Header: rtp.Header{
Version: 2,
Marker: true,
PayloadType: 96,
SequenceNumber: 123,
Timestamp: 45343,
SSRC: 563423,
Padding: true,
},
Payload: []byte{0x01, 0x02, 0x03, 0x04},
PaddingSize: 20,
}
err = p.Process(&DataGeneric{
RTPPackets: []*rtp.Packet{pkt},
}, false)
require.NoError(t, err)
require.Equal(t, &rtp.Packet{
Header: rtp.Header{
Version: 2,
Marker: true,
PayloadType: 96,
SequenceNumber: 123,
Timestamp: 45343,
SSRC: 563423,
},
Payload: []byte{0x01, 0x02, 0x03, 0x04},
}, pkt)
}

51
internal/core/formatprocessor_h264.go → internal/formatprocessor/h264.go
Unescape View File

@ -1,4 +1,4 @@
package core package formatprocessor
import ( import (
"bytes" "bytes"
@ -67,19 +67,22 @@ func rtpH264ExtractSPSPPS(pkt *rtp.Packet) ([]byte, []byte) {
} }
} }
type dataH264 struct { // DataH264 is a H264 data unit.
rtpPackets []*rtp.Packet type DataH264 struct {
ntp time.Time RTPPackets []*rtp.Packet
pts time.Duration NTP time.Time
au [][]byte PTS time.Duration
AU [][]byte
} }
func (d *dataH264) getRTPPackets() []*rtp.Packet { // GetRTPPackets implements Data.
return d.rtpPackets func (d *DataH264) GetRTPPackets() []*rtp.Packet {
return d.RTPPackets
} }
func (d *dataH264) getNTP() time.Time { // GetNTP implements Data.
return d.ntp func (d *DataH264) GetNTP() time.Time {
return d.NTP
} }
type formatProcessorH264 struct { type formatProcessorH264 struct {
@ -89,7 +92,7 @@ type formatProcessorH264 struct {
decoder *rtph264.Decoder decoder *rtph264.Decoder
} }
func newFormatProcessorH264( func newH264(
forma *format.H264, forma *format.H264,
allocateEncoder bool, allocateEncoder bool,
) (*formatProcessorH264, error) { ) (*formatProcessorH264, error) {
@ -188,11 +191,11 @@ func (t *formatProcessorH264) remuxAccessUnit(nalus [][]byte) [][]byte {
return filteredNALUs return filteredNALUs
} }
func (t *formatProcessorH264) process(dat data, hasNonRTSPReaders bool) error { //nolint:dupl func (t *formatProcessorH264) Process(dat Data, hasNonRTSPReaders bool) error { //nolint:dupl
tdata := dat.(*dataH264) tdata := dat.(*DataH264)
if tdata.rtpPackets != nil { if tdata.RTPPackets != nil {
pkt := tdata.rtpPackets[0] pkt := tdata.RTPPackets[0]
t.updateTrackParametersFromRTPPacket(pkt) t.updateTrackParametersFromRTPPacket(pkt)
if t.encoder == nil { if t.encoder == nil {
@ -223,11 +226,11 @@ func (t *formatProcessorH264) process(dat data, hasNonRTSPReaders bool) error {
} }
if t.encoder != nil { if t.encoder != nil {
tdata.rtpPackets = nil tdata.RTPPackets = nil
} }
// DecodeUntilMarker() is necessary, otherwise Encode() generates partial groups // DecodeUntilMarker() is necessary, otherwise Encode() generates partial groups
au, pts, err := t.decoder.DecodeUntilMarker(pkt) au, PTS, err := t.decoder.DecodeUntilMarker(pkt)
if err != nil { if err != nil {
if err == rtph264.ErrNonStartingPacketAndNoPrevious || err == rtph264.ErrMorePacketsNeeded { if err == rtph264.ErrNonStartingPacketAndNoPrevious || err == rtph264.ErrMorePacketsNeeded {
return nil return nil
@ -235,9 +238,9 @@ func (t *formatProcessorH264) process(dat data, hasNonRTSPReaders bool) error {
return err return err
} }
tdata.au = au tdata.AU = au
tdata.pts = pts tdata.PTS = PTS
tdata.au = t.remuxAccessUnit(tdata.au) tdata.AU = t.remuxAccessUnit(tdata.AU)
} }
// route packet as is // route packet as is
@ -245,15 +248,15 @@ func (t *formatProcessorH264) process(dat data, hasNonRTSPReaders bool) error {
return nil return nil
} }
} else { } else {
t.updateTrackParametersFromNALUs(tdata.au) t.updateTrackParametersFromNALUs(tdata.AU)
tdata.au = t.remuxAccessUnit(tdata.au) tdata.AU = t.remuxAccessUnit(tdata.AU)
} }
pkts, err := t.encoder.Encode(tdata.au, tdata.pts) pkts, err := t.encoder.Encode(tdata.AU, tdata.PTS)
if err != nil { if err != nil {
return err return err
} }
tdata.rtpPackets = pkts tdata.RTPPackets = pkts
return nil return nil
} }

140
internal/formatprocessor/h264_test.go
Unescape View File

@ -0,0 +1,140 @@
package formatprocessor
import (
"bytes"
"testing"
"github.com/aler9/gortsplib/v2/pkg/format"
"github.com/pion/rtp"
"github.com/stretchr/testify/require"
)
func TestH264DynamicParams(t *testing.T) {
forma := &format.H264{
PayloadTyp: 96,
PacketizationMode: 1,
}
p, err := New(forma, false)
require.NoError(t, err)
enc := forma.CreateEncoder()
pkts, err := enc.Encode([][]byte{{7, 1, 2, 3}}, 0) // SPS
require.NoError(t, err)
p.Process(&DataH264{RTPPackets: []*rtp.Packet{pkts[0]}}, false)
pkts, err = enc.Encode([][]byte{{8}}, 0) // PPS
require.NoError(t, err)
p.Process(&DataH264{RTPPackets: []*rtp.Packet{pkts[0]}}, false)
pkts, err = enc.Encode([][]byte{{7, 4, 5, 6}}, 0) // SPS
require.NoError(t, err)
p.Process(&DataH264{RTPPackets: []*rtp.Packet{pkts[0]}}, false)
pkts, err = enc.Encode([][]byte{{8, 1}}, 0) // PPS
require.NoError(t, err)
p.Process(&DataH264{RTPPackets: []*rtp.Packet{pkts[0]}}, false)
require.Equal(t, []byte{7, 4, 5, 6}, forma.SPS)
require.Equal(t, []byte{8, 1}, forma.PPS)
}
func TestH264OversizedPackets(t *testing.T) {
forma := &format.H264{
PayloadTyp: 96,
SPS: []byte{0x01, 0x02, 0x03, 0x04},
PPS: []byte{0x01, 0x02, 0x03, 0x04},
PacketizationMode: 1,
}
p, err := New(forma, false)
require.NoError(t, err)
var out []*rtp.Packet
for _, pkt := range []*rtp.Packet{
{
Header: rtp.Header{
Version: 2,
Marker: true,
PayloadType: 96,
SequenceNumber: 123,
Timestamp: 45343,
SSRC: 563423,
Padding: true,
},
Payload: []byte{0x01, 0x02, 0x03, 0x04},
},
{
Header: rtp.Header{
Version: 2,
Marker: false,
PayloadType: 96,
SequenceNumber: 124,
Timestamp: 45343,
SSRC: 563423,
Padding: true,
},
Payload: append([]byte{0x1c, 0b10000000}, bytes.Repeat([]byte{0x01, 0x02, 0x03, 0x04}, 2000/4)...),
},
{
Header: rtp.Header{
Version: 2,
Marker: true,
PayloadType: 96,
SequenceNumber: 125,
Timestamp: 45343,
SSRC: 563423,
Padding: true,
},
Payload: []byte{0x1c, 0b01000000, 0x01, 0x02, 0x03, 0x04},
},
} {
data := &DataH264{RTPPackets: []*rtp.Packet{pkt}}
p.Process(data, false)
out = append(out, data.RTPPackets...)
}
require.Equal(t, []*rtp.Packet{
{
Header: rtp.Header{
Version: 2,
Marker: true,
PayloadType: 96,
SequenceNumber: 123,
Timestamp: 45343,
SSRC: 563423,
},
Payload: []byte{0x01, 0x02, 0x03, 0x04},
},
{
Header: rtp.Header{
Version: 2,
Marker: false,
PayloadType: 96,
SequenceNumber: 124,
Timestamp: 45343,
SSRC: 563423,
},
Payload: append(
append([]byte{0x1c, 0x80}, bytes.Repeat([]byte{0x01, 0x02, 0x03, 0x04}, 364)...),
[]byte{0x01, 0x02}...,
),
},
{
Header: rtp.Header{
Version: 2,
Marker: true,
PayloadType: 96,
SequenceNumber: 125,
Timestamp: 45343,
SSRC: 563423,
},
Payload: append(
[]byte{0x1c, 0x40, 0x03, 0x04},
bytes.Repeat([]byte{0x01, 0x02, 0x03, 0x04}, 136)...,
),
},
}, out)
}

51
internal/core/formatprocessor_h265.go → internal/formatprocessor/h265.go
Unescape View File

@ -1,4 +1,4 @@
package core package formatprocessor
import ( import (
"bytes" "bytes"
@ -74,19 +74,22 @@ func rtpH265ExtractVPSSPSPPS(pkt *rtp.Packet) ([]byte, []byte, []byte) {
} }
} }
type dataH265 struct { // DataH265 is a H265 data unit.
rtpPackets []*rtp.Packet type DataH265 struct {
ntp time.Time RTPPackets []*rtp.Packet
pts time.Duration NTP time.Time
au [][]byte PTS time.Duration
AU [][]byte
} }
func (d *dataH265) getRTPPackets() []*rtp.Packet { // GetRTPPackets implements Data.
return d.rtpPackets func (d *DataH265) GetRTPPackets() []*rtp.Packet {
return d.RTPPackets
} }
func (d *dataH265) getNTP() time.Time { // GetNTP implements Data.
return d.ntp func (d *DataH265) GetNTP() time.Time {
return d.NTP
} }
type formatProcessorH265 struct { type formatProcessorH265 struct {
@ -96,7 +99,7 @@ type formatProcessorH265 struct {
decoder *rtph265.Decoder decoder *rtph265.Decoder
} }
func newFormatProcessorH265( func newH265(
forma *format.H265, forma *format.H265,
allocateEncoder bool, allocateEncoder bool,
) (*formatProcessorH265, error) { ) (*formatProcessorH265, error) {
@ -206,11 +209,11 @@ func (t *formatProcessorH265) remuxAccessUnit(nalus [][]byte) [][]byte {
return filteredNALUs return filteredNALUs
} }
func (t *formatProcessorH265) process(dat data, hasNonRTSPReaders bool) error { //nolint:dupl func (t *formatProcessorH265) Process(dat Data, hasNonRTSPReaders bool) error { //nolint:dupl
tdata := dat.(*dataH265) tdata := dat.(*DataH265)
if tdata.rtpPackets != nil { if tdata.RTPPackets != nil {
pkt := tdata.rtpPackets[0] pkt := tdata.RTPPackets[0]
t.updateTrackParametersFromRTPPacket(pkt) t.updateTrackParametersFromRTPPacket(pkt)
if t.encoder == nil { if t.encoder == nil {
@ -241,11 +244,11 @@ func (t *formatProcessorH265) process(dat data, hasNonRTSPReaders bool) error {
} }
if t.encoder != nil { if t.encoder != nil {
tdata.rtpPackets = nil tdata.RTPPackets = nil
} }
// DecodeUntilMarker() is necessary, otherwise Encode() generates partial groups // DecodeUntilMarker() is necessary, otherwise Encode() generates partial groups
au, pts, err := t.decoder.DecodeUntilMarker(pkt) au, PTS, err := t.decoder.DecodeUntilMarker(pkt)
if err != nil { if err != nil {
if err == rtph265.ErrNonStartingPacketAndNoPrevious || err == rtph265.ErrMorePacketsNeeded { if err == rtph265.ErrNonStartingPacketAndNoPrevious || err == rtph265.ErrMorePacketsNeeded {
return nil return nil
@ -253,9 +256,9 @@ func (t *formatProcessorH265) process(dat data, hasNonRTSPReaders bool) error {
return err return err
} }
tdata.au = au tdata.AU = au
tdata.pts = pts tdata.PTS = PTS
tdata.au = t.remuxAccessUnit(tdata.au) tdata.AU = t.remuxAccessUnit(tdata.AU)
} }
// route packet as is // route packet as is
@ -263,15 +266,15 @@ func (t *formatProcessorH265) process(dat data, hasNonRTSPReaders bool) error {
return nil return nil
} }
} else { } else {
t.updateTrackParametersFromNALUs(tdata.au) t.updateTrackParametersFromNALUs(tdata.AU)
tdata.au = t.remuxAccessUnit(tdata.au) tdata.AU = t.remuxAccessUnit(tdata.AU)
} }
pkts, err := t.encoder.Encode(tdata.au, tdata.pts) pkts, err := t.encoder.Encode(tdata.AU, tdata.PTS)
if err != nil { if err != nil {
return err return err
} }
tdata.rtpPackets = pkts tdata.RTPPackets = pkts
return nil return nil
} }

125
internal/formatprocessor/h265_test.go
Unescape View File

@ -0,0 +1,125 @@
package formatprocessor
import (
"bytes"
"testing"
"github.com/aler9/gortsplib/v2/pkg/codecs/h265"
"github.com/aler9/gortsplib/v2/pkg/format"
"github.com/pion/rtp"
"github.com/stretchr/testify/require"
)
func TestH265DynamicParams(t *testing.T) {
forma := &format.H265{
PayloadTyp: 96,
}
p, err := New(forma, false)
require.NoError(t, err)
enc := forma.CreateEncoder()
pkts, err := enc.Encode([][]byte{{byte(h265.NALUType_VPS_NUT) << 1, 1, 2, 3}}, 0)
require.NoError(t, err)
p.Process(&DataH265{RTPPackets: []*rtp.Packet{pkts[0]}}, false)
pkts, err = enc.Encode([][]byte{{byte(h265.NALUType_SPS_NUT) << 1, 4, 5, 6}}, 0)
require.NoError(t, err)
p.Process(&DataH265{RTPPackets: []*rtp.Packet{pkts[0]}}, false)
pkts, err = enc.Encode([][]byte{{byte(h265.NALUType_PPS_NUT) << 1, 7, 8, 9}}, 0)
require.NoError(t, err)
p.Process(&DataH265{RTPPackets: []*rtp.Packet{pkts[0]}}, false)
require.Equal(t, []byte{byte(h265.NALUType_VPS_NUT) << 1, 1, 2, 3}, forma.VPS)
require.Equal(t, []byte{byte(h265.NALUType_SPS_NUT) << 1, 4, 5, 6}, forma.SPS)
require.Equal(t, []byte{byte(h265.NALUType_PPS_NUT) << 1, 7, 8, 9}, forma.PPS)
}
func TestH265OversizedPackets(t *testing.T) {
forma := &format.H265{
PayloadTyp: 96,
VPS: []byte{byte(h265.NALUType_VPS_NUT) << 1, 10, 11, 12},
SPS: []byte{byte(h265.NALUType_SPS_NUT) << 1, 13, 14, 15},
PPS: []byte{byte(h265.NALUType_PPS_NUT) << 1, 16, 17, 18},
}
p, err := New(forma, false)
require.NoError(t, err)
var out []*rtp.Packet
for _, pkt := range []*rtp.Packet{
{
Header: rtp.Header{
Version: 2,
Marker: true,
PayloadType: 96,
SequenceNumber: 123,
Timestamp: 45343,
SSRC: 563423,
Padding: true,
},
Payload: []byte{0x01, 0x02, 0x03, 0x04},
},
{
Header: rtp.Header{
Version: 2,
Marker: true,
PayloadType: 96,
SequenceNumber: 124,
Timestamp: 45343,
SSRC: 563423,
Padding: true,
},
Payload: bytes.Repeat([]byte{0x01, 0x02, 0x03, 0x04}, 2000/4),
},
} {
data := &DataH265{RTPPackets: []*rtp.Packet{pkt}}
p.Process(data, false)
out = append(out, data.RTPPackets...)
}
require.Equal(t, []*rtp.Packet{
{
Header: rtp.Header{
Version: 2,
Marker: true,
PayloadType: 96,
SequenceNumber: 123,
Timestamp: 45343,
SSRC: 563423,
},
Payload: []byte{0x01, 0x02, 0x03, 0x04},
},
{
Header: rtp.Header{
Version: 2,
Marker: false,
PayloadType: 96,
SequenceNumber: 124,
Timestamp: 45343,
SSRC: 563423,
},
Payload: append(
append([]byte{0x63, 0x02, 0x80, 0x03, 0x04}, bytes.Repeat([]byte{0x01, 0x02, 0x03, 0x04}, 363)...),
[]byte{0x01, 0x02, 0x03}...,
),
},
{
Header: rtp.Header{
Version: 2,
Marker: true,
PayloadType: 96,
SequenceNumber: 125,
Timestamp: 45343,
SSRC: 563423,
},
Payload: append(
[]byte{0x63, 0x02, 0x40, 0x04},
bytes.Repeat([]byte{0x01, 0x02, 0x03, 0x04}, 135)...,
),
},
}, out)
}

43
internal/core/formatprocessor_mpeg4audio.go → internal/formatprocessor/mpeg4audio.go
Unescape View File

@ -1,4 +1,4 @@
package core package formatprocessor
import ( import (
"fmt" "fmt"
@ -9,19 +9,22 @@ import (
"github.com/pion/rtp" "github.com/pion/rtp"
) )
type dataMPEG4Audio struct { // DataMPEG4Audio is a MPEG4-audio data unit.
rtpPackets []*rtp.Packet type DataMPEG4Audio struct {
ntp time.Time RTPPackets []*rtp.Packet
pts time.Duration NTP time.Time
aus [][]byte PTS time.Duration
AUs [][]byte
} }
func (d *dataMPEG4Audio) getRTPPackets() []*rtp.Packet { // GetRTPPackets implements Data.
return d.rtpPackets func (d *DataMPEG4Audio) GetRTPPackets() []*rtp.Packet {
return d.RTPPackets
} }
func (d *dataMPEG4Audio) getNTP() time.Time { // GetNTP implements Data.
return d.ntp func (d *DataMPEG4Audio) GetNTP() time.Time {
return d.NTP
} }
type formatProcessorMPEG4Audio struct { type formatProcessorMPEG4Audio struct {
@ -30,7 +33,7 @@ type formatProcessorMPEG4Audio struct {
decoder *rtpmpeg4audio.Decoder decoder *rtpmpeg4audio.Decoder
} }
func newFormatProcessorMPEG4Audio( func newMPEG4Audio(
forma *format.MPEG4Audio, forma *format.MPEG4Audio,
allocateEncoder bool, allocateEncoder bool,
) (*formatProcessorMPEG4Audio, error) { ) (*formatProcessorMPEG4Audio, error) {
@ -45,11 +48,11 @@ func newFormatProcessorMPEG4Audio(
return t, nil return t, nil
} }
func (t *formatProcessorMPEG4Audio) process(dat data, hasNonRTSPReaders bool) error { //nolint:dupl func (t *formatProcessorMPEG4Audio) Process(dat Data, hasNonRTSPReaders bool) error { //nolint:dupl
tdata := dat.(*dataMPEG4Audio) tdata := dat.(*DataMPEG4Audio)
if tdata.rtpPackets != nil { if tdata.RTPPackets != nil {
pkt := tdata.rtpPackets[0] pkt := tdata.RTPPackets[0]
// remove padding // remove padding
pkt.Header.Padding = false pkt.Header.Padding = false
@ -66,7 +69,7 @@ func (t *formatProcessorMPEG4Audio) process(dat data, hasNonRTSPReaders bool) er
t.decoder = t.format.CreateDecoder() t.decoder = t.format.CreateDecoder()
} }
aus, pts, err := t.decoder.Decode(pkt) aus, PTS, err := t.decoder.Decode(pkt)
if err != nil { if err != nil {
if err == rtpmpeg4audio.ErrMorePacketsNeeded { if err == rtpmpeg4audio.ErrMorePacketsNeeded {
return nil return nil
@ -74,19 +77,19 @@ func (t *formatProcessorMPEG4Audio) process(dat data, hasNonRTSPReaders bool) er
return err return err
} }
tdata.aus = aus tdata.AUs = aus
tdata.pts = pts tdata.PTS = PTS
} }
// route packet as is // route packet as is
return nil return nil
} }
pkts, err := t.encoder.Encode(tdata.aus, tdata.pts) pkts, err := t.encoder.Encode(tdata.AUs, tdata.PTS)
if err != nil { if err != nil {
return err return err
} }
tdata.rtpPackets = pkts tdata.RTPPackets = pkts
return nil return nil
} }

43
internal/core/formatprocessor_opus.go → internal/formatprocessor/opus.go
Unescape View File

@ -1,4 +1,4 @@
package core package formatprocessor
import ( import (
"fmt" "fmt"
@ -9,19 +9,22 @@ import (
"github.com/pion/rtp" "github.com/pion/rtp"
) )
type dataOpus struct { // DataOpus is a Opus data unit.
rtpPackets []*rtp.Packet type DataOpus struct {
ntp time.Time RTPPackets []*rtp.Packet
pts time.Duration NTP time.Time
frame []byte PTS time.Duration
Frame []byte
} }
func (d *dataOpus) getRTPPackets() []*rtp.Packet { // GetRTPPackets implements Data.
return d.rtpPackets func (d *DataOpus) GetRTPPackets() []*rtp.Packet {
return d.RTPPackets
} }
func (d *dataOpus) getNTP() time.Time { // GetNTP implements Data.
return d.ntp func (d *DataOpus) GetNTP() time.Time {
return d.NTP
} }
type formatProcessorOpus struct { type formatProcessorOpus struct {
@ -30,7 +33,7 @@ type formatProcessorOpus struct {
decoder *rtpsimpleaudio.Decoder decoder *rtpsimpleaudio.Decoder
} }
func newFormatProcessorOpus( func newOpus(
forma *format.Opus, forma *format.Opus,
allocateEncoder bool, allocateEncoder bool,
) (*formatProcessorOpus, error) { ) (*formatProcessorOpus, error) {
@ -45,11 +48,11 @@ func newFormatProcessorOpus(
return t, nil return t, nil
} }
func (t *formatProcessorOpus) process(dat data, hasNonRTSPReaders bool) error { //nolint:dupl func (t *formatProcessorOpus) Process(dat Data, hasNonRTSPReaders bool) error { //nolint:dupl
tdata := dat.(*dataOpus) tdata := dat.(*DataOpus)
if tdata.rtpPackets != nil { if tdata.RTPPackets != nil {
pkt := tdata.rtpPackets[0] pkt := tdata.RTPPackets[0]
// remove padding // remove padding
pkt.Header.Padding = false pkt.Header.Padding = false
@ -66,24 +69,24 @@ func (t *formatProcessorOpus) process(dat data, hasNonRTSPReaders bool) error {
t.decoder = t.format.CreateDecoder() t.decoder = t.format.CreateDecoder()
} }
frame, pts, err := t.decoder.Decode(pkt) frame, PTS, err := t.decoder.Decode(pkt)
if err != nil { if err != nil {
return err return err
} }
tdata.frame = frame tdata.Frame = frame
tdata.pts = pts tdata.PTS = PTS
} }
// route packet as is // route packet as is
return nil return nil
} }
pkt, err := t.encoder.Encode(tdata.frame, tdata.pts) pkt, err := t.encoder.Encode(tdata.Frame, tdata.PTS)
if err != nil { if err != nil {
return err return err
} }
tdata.rtpPackets = []*rtp.Packet{pkt} tdata.RTPPackets = []*rtp.Packet{pkt}
return nil return nil
} }

48
internal/formatprocessor/processor.go
Unescape View File

@ -0,0 +1,48 @@
// Package formatprocessor contains code to cleanup and normalize streams.
package formatprocessor
import (
"time"
"github.com/pion/rtp"
"github.com/aler9/gortsplib/v2/pkg/format"
)
// Data is the elementary data unit routed across the server.
type Data interface {
GetRTPPackets() []*rtp.Packet
GetNTP() time.Time
}
// Processor allows to cleanup and normalize streams.
type Processor interface {
// cleanups and normalizes a data unit.
Process(Data, bool) error
}
// New allocates a Processor.
func New(forma format.Format, generateRTPPackets bool) (Processor, error) {
switch forma := forma.(type) {
case *format.H264:
return newH264(forma, generateRTPPackets)
case *format.H265:
return newH265(forma, generateRTPPackets)
case *format.VP8:
return newVP8(forma, generateRTPPackets)
case *format.VP9:
return newVP9(forma, generateRTPPackets)
case *format.MPEG4Audio:
return newMPEG4Audio(forma, generateRTPPackets)
case *format.Opus:
return newOpus(forma, generateRTPPackets)
default:
return newGeneric(forma, generateRTPPackets)
}
}

43
internal/core/formatprocessor_vp8.go → internal/formatprocessor/vp8.go
Unescape View File

@ -1,4 +1,4 @@
package core //nolint:dupl package formatprocessor //nolint:dupl
import ( import (
"fmt" "fmt"
@ -9,19 +9,22 @@ import (
"github.com/pion/rtp" "github.com/pion/rtp"
) )
type dataVP8 struct { // DataVP8 is a VP8 data unit.
rtpPackets []*rtp.Packet type DataVP8 struct {
ntp time.Time RTPPackets []*rtp.Packet
pts time.Duration NTP time.Time
frame []byte PTS time.Duration
Frame []byte
} }
func (d *dataVP8) getRTPPackets() []*rtp.Packet { // GetRTPPackets implements Data.
return d.rtpPackets func (d *DataVP8) GetRTPPackets() []*rtp.Packet {
return d.RTPPackets
} }
func (d *dataVP8) getNTP() time.Time { // GetNTP implements Data.
return d.ntp func (d *DataVP8) GetNTP() time.Time {
return d.NTP
} }
type formatProcessorVP8 struct { type formatProcessorVP8 struct {
@ -30,7 +33,7 @@ type formatProcessorVP8 struct {
decoder *rtpvp8.Decoder decoder *rtpvp8.Decoder
} }
func newFormatProcessorVP8( func newVP8(
forma *format.VP8, forma *format.VP8,
allocateEncoder bool, allocateEncoder bool,
) (*formatProcessorVP8, error) { ) (*formatProcessorVP8, error) {
@ -45,11 +48,11 @@ func newFormatProcessorVP8(
return t, nil return t, nil
} }
func (t *formatProcessorVP8) process(dat data, hasNonRTSPReaders bool) error { //nolint:dupl func (t *formatProcessorVP8) Process(dat Data, hasNonRTSPReaders bool) error { //nolint:dupl
tdata := dat.(*dataVP8) tdata := dat.(*DataVP8)
if tdata.rtpPackets != nil { if tdata.RTPPackets != nil {
pkt := tdata.rtpPackets[0] pkt := tdata.RTPPackets[0]
// remove padding // remove padding
pkt.Header.Padding = false pkt.Header.Padding = false
@ -66,7 +69,7 @@ func (t *formatProcessorVP8) process(dat data, hasNonRTSPReaders bool) error { /
t.decoder = t.format.CreateDecoder() t.decoder = t.format.CreateDecoder()
} }
frame, pts, err := t.decoder.Decode(pkt) frame, PTS, err := t.decoder.Decode(pkt)
if err != nil { if err != nil {
if err == rtpvp8.ErrMorePacketsNeeded { if err == rtpvp8.ErrMorePacketsNeeded {
return nil return nil
@ -74,19 +77,19 @@ func (t *formatProcessorVP8) process(dat data, hasNonRTSPReaders bool) error { /
return err return err
} }
tdata.frame = frame tdata.Frame = frame
tdata.pts = pts tdata.PTS = PTS
} }
// route packet as is // route packet as is
return nil return nil
} }
pkts, err := t.encoder.Encode(tdata.frame, tdata.pts) pkts, err := t.encoder.Encode(tdata.Frame, tdata.PTS)
if err != nil { if err != nil {
return err return err
} }
tdata.rtpPackets = pkts tdata.RTPPackets = pkts
return nil return nil
} }

43
internal/core/formatprocessor_vp9.go → internal/formatprocessor/vp9.go
Unescape View File

@ -1,4 +1,4 @@
package core //nolint:dupl package formatprocessor //nolint:dupl
import ( import (
"fmt" "fmt"
@ -9,19 +9,22 @@ import (
"github.com/pion/rtp" "github.com/pion/rtp"
) )
type dataVP9 struct { // DataVP9 is a VP9 data unit.
rtpPackets []*rtp.Packet type DataVP9 struct {
ntp time.Time RTPPackets []*rtp.Packet
pts time.Duration NTP time.Time
frame []byte PTS time.Duration
Frame []byte
} }
func (d *dataVP9) getRTPPackets() []*rtp.Packet { // GetRTPPackets implements Data.
return d.rtpPackets func (d *DataVP9) GetRTPPackets() []*rtp.Packet {
return d.RTPPackets
} }
func (d *dataVP9) getNTP() time.Time { // GetNTP implements Data.
return d.ntp func (d *DataVP9) GetNTP() time.Time {
return d.NTP
} }
type formatProcessorVP9 struct { type formatProcessorVP9 struct {
@ -30,7 +33,7 @@ type formatProcessorVP9 struct {
decoder *rtpvp9.Decoder decoder *rtpvp9.Decoder
} }
func newFormatProcessorVP9( func newVP9(
forma *format.VP9, forma *format.VP9,
allocateEncoder bool, allocateEncoder bool,
) (*formatProcessorVP9, error) { ) (*formatProcessorVP9, error) {
@ -45,11 +48,11 @@ func newFormatProcessorVP9(
return t, nil return t, nil
} }
func (t *formatProcessorVP9) process(dat data, hasNonRTSPReaders bool) error { //nolint:dupl func (t *formatProcessorVP9) Process(dat Data, hasNonRTSPReaders bool) error { //nolint:dupl
tdata := dat.(*dataVP9) tdata := dat.(*DataVP9)
if tdata.rtpPackets != nil { if tdata.RTPPackets != nil {
pkt := tdata.rtpPackets[0] pkt := tdata.RTPPackets[0]
// remove padding // remove padding
pkt.Header.Padding = false pkt.Header.Padding = false
@ -66,7 +69,7 @@ func (t *formatProcessorVP9) process(dat data, hasNonRTSPReaders bool) error { /
t.decoder = t.format.CreateDecoder() t.decoder = t.format.CreateDecoder()
} }
frame, pts, err := t.decoder.Decode(pkt) frame, PTS, err := t.decoder.Decode(pkt)
if err != nil { if err != nil {
if err == rtpvp9.ErrMorePacketsNeeded { if err == rtpvp9.ErrMorePacketsNeeded {
return nil return nil
@ -74,19 +77,19 @@ func (t *formatProcessorVP9) process(dat data, hasNonRTSPReaders bool) error { /
return err return err
} }
tdata.frame = frame tdata.Frame = frame
tdata.pts = pts tdata.PTS = PTS
} }
// route packet as is // route packet as is
return nil return nil
} }
pkts, err := t.encoder.Encode(tdata.frame, tdata.pts) pkts, err := t.encoder.Encode(tdata.Frame, tdata.PTS)
if err != nil { if err != nil {
return err return err
} }
tdata.rtpPackets = pkts tdata.RTPPackets = pkts
return nil return nil
} }
Write Preview
Loading…
Cancel
Save