OPC/mediamtx
2
0
Fork 0
Code Issues 3 Pull Requests Actions Packages Projects Releases Wiki Activity

rtmp: fix connect command when reading (#4512)

Browse Source 
when reading, the "connect" command should not contain fpad,
capabilities, audioCodecs, videoCodecs, videoFunction.
This commit is contained in:
Alessandro Ros
2025-05-10 15:26:48 +02:00
committed by GitHub
parent fd0f08e94a
commit a85d8f1391
14 changed files with 313 additions and 143 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/core/api_test.go
+24 -4
View File
@@ -430,10 +430,20 @@ func TestAPIProtocolListGet(t *testing.T) {
require.NoError(t, err)
defer nconn.Close()
conn, err := rtmp.NewClientConn(nconn, u, true)
conn := &rtmp.Conn{
RW: nconn,
Client: true,
URL: u,
Publish: true,
}
err = conn.Initialize()
require.NoError(t, err)
w, err := rtmp.NewWriter(conn, test.FormatH264, nil)
w := &rtmp.Writer{
Conn: conn,
VideoTrack: test.FormatH264,
}
err = w.Initialize()
require.NoError(t, err)
err = w.WriteH264(2*time.Second, 2*time.Second, [][]byte{{5, 2, 3, 4}})
@@ -1007,10 +1017,20 @@ func TestAPIProtocolKick(t *testing.T) {
require.NoError(t, err)
defer nconn.Close()
conn, err := rtmp.NewClientConn(nconn, u, true)
conn := &rtmp.Conn{
RW: nconn,
Client: true,
URL: u,
Publish: true,
}
err = conn.Initialize()
require.NoError(t, err)
w, err := rtmp.NewWriter(conn, test.FormatH264, nil)
w := &rtmp.Writer{
Conn: conn,
VideoTrack: test.FormatH264,
}
err = w.Initialize()
require.NoError(t, err)
err = w.WriteH264(2*time.Second, 2*time.Second, [][]byte{{5, 2, 3, 4}})
internal/core/metrics_test.go
+24 -4
View File
@@ -203,10 +203,20 @@ webrtc_sessions_bytes_sent 0
require.NoError(t, err)
defer nconn.Close()
conn, err := rtmp.NewClientConn(nconn, u, true)
conn := &rtmp.Conn{
RW: nconn,
Client: true,
URL: u,
Publish: true,
}
err = conn.Initialize()
require.NoError(t, err)
w, err := rtmp.NewWriter(conn, test.FormatH264, nil)
w := &rtmp.Writer{
Conn: conn,
VideoTrack: test.FormatH264,
}
err = w.Initialize()
require.NoError(t, err)
err = w.WriteH264(2*time.Second, 2*time.Second, [][]byte{{5, 2, 3, 4}})
@@ -224,10 +234,20 @@ webrtc_sessions_bytes_sent 0
require.NoError(t, err)
defer nconn.Close() //nolint:errcheck
conn, err := rtmp.NewClientConn(nconn, u, true)
conn := &rtmp.Conn{
RW: nconn,
Client: true,
URL: u,
Publish: true,
}
err = conn.Initialize()
require.NoError(t, err)
w, err := rtmp.NewWriter(conn, test.FormatH264, nil)
w := &rtmp.Writer{
Conn: conn,
VideoTrack: test.FormatH264,
}
err = w.Initialize()
require.NoError(t, err)
err = w.WriteH264(2*time.Second, 2*time.Second, [][]byte{{5, 2, 3, 4}})
internal/core/path_test.go
+36 -6
View File
@@ -217,7 +217,13 @@ func TestPathRunOnConnect(t *testing.T) {
require.NoError(t, err)
defer nconn.Close()
_, err = rtmp.NewClientConn(nconn, u, true)
conn := &rtmp.Conn{
RW: nconn,
Client: true,
URL: u,
Publish: true,
}
err = conn.Initialize()
require.NoError(t, err)
case "rtmps":
@@ -230,7 +236,13 @@ func TestPathRunOnConnect(t *testing.T) {
require.NoError(t, err)
defer nconn.Close() //nolint:errcheck
_, err = rtmp.NewClientConn(nconn, u, true)
conn := &rtmp.Conn{
RW: nconn,
Client: true,
URL: u,
Publish: true,
}
err = conn.Initialize()
require.NoError(t, err)
case "srt":
@@ -440,10 +452,19 @@ func TestPathRunOnRead(t *testing.T) {
require.NoError(t, err)
defer nconn.Close()
conn, err := rtmp.NewClientConn(nconn, u, false)
conn := &rtmp.Conn{
RW: nconn,
Client: true,
URL: u,
Publish: false,
}
err = conn.Initialize()
require.NoError(t, err)
_, err = rtmp.NewReader(conn)
r := &rtmp.Reader{
Conn: conn,
}
err = r.Initialize()
require.NoError(t, err)
case "rtmps":
@@ -454,7 +475,13 @@ func TestPathRunOnRead(t *testing.T) {
require.NoError(t, err)
defer nconn.Close() //nolint:errcheck
conn, err := rtmp.NewClientConn(nconn, u, false)
conn := &rtmp.Conn{
RW: nconn,
Client: true,
URL: u,
Publish: false,
}
err = conn.Initialize()
require.NoError(t, err)
go func() {
@@ -474,7 +501,10 @@ func TestPathRunOnRead(t *testing.T) {
}
}()
_, err = rtmp.NewReader(conn)
r := &rtmp.Reader{
Conn: conn,
}
err = r.Initialize()
require.NoError(t, err)
case "srt":
internal/protocols/rtmp/conn.go
+54 -49
View File
@@ -141,26 +141,50 @@ func readCommandResult(
// Conn is a RTMP connection.
type Conn struct {
RW io.ReadWriter
Client bool
URL *url.URL
Publish bool
skipHandshake bool
bc *bytecounter.ReadWriter
mrw *message.ReadWriter
}
// NewClientConn initializes a client-side connection.
func NewClientConn(rw io.ReadWriter, u *url.URL, publish bool) (*Conn, error) {
c := &Conn{
bc: bytecounter.NewReadWriter(rw),
// Initialize initializes Conn.
func (c *Conn) Initialize() error {
c.bc = bytecounter.NewReadWriter(c.RW)
if !c.skipHandshake {
if c.Client {
if c.URL == nil {
return fmt.Errorf("URL must be specified in client mode")
}
err := c.initializeClient()
if err != nil {
return err
}
} else {
if c.URL != nil {
return fmt.Errorf("URL must be empty in server mode")
}
var err error
c.URL, c.Publish, err = c.initializeServer()
if err != nil {
return err
}
}
} else {
c.mrw = message.NewReadWriter(c.bc, c.bc, false)
}
err := c.initializeClient(u, publish)
if err != nil {
return nil, err
}
return c, nil
return nil
}
func (c *Conn) initializeClient(u *url.URL, publish bool) error {
connectpath, actionpath := splitPath(u)
func (c *Conn) initializeClient() error {
connectpath, actionpath := splitPath(c.URL)
_, _, err := handshake.DoClient(c.bc, false, false)
if err != nil {
@@ -191,22 +215,27 @@ func (c *Conn) initializeClient(u *url.URL, publish bool) error {
return err
}
connectArg := amf0.Object{
{Key: "app", Value: connectpath},
{Key: "flashVer", Value: "LNX 9,0,124,2"},
{Key: "tcUrl", Value: getTcURL(c.URL)},
}
if !c.Publish {
connectArg = append(connectArg,
amf0.ObjectEntry{Key: "fpad", Value: false},
amf0.ObjectEntry{Key: "capabilities", Value: float64(15)},
amf0.ObjectEntry{Key: "audioCodecs", Value: float64(4071)},
amf0.ObjectEntry{Key: "videoCodecs", Value: float64(252)},
amf0.ObjectEntry{Key: "videoFunction", Value: float64(1)},
)
}
err = c.mrw.Write(&message.CommandAMF0{
ChunkStreamID: 3,
Name: "connect",
CommandID: 1,
Arguments: []interface{}{
amf0.Object{
{Key: "app", Value: connectpath},
{Key: "flashVer", Value: "LNX 9,0,124,2"},
{Key: "tcUrl", Value: getTcURL(u)},
{Key: "fpad", Value: false},
{Key: "capabilities", Value: float64(15)},
{Key: "audioCodecs", Value: float64(4071)},
{Key: "videoCodecs", Value: float64(252)},
{Key: "videoFunction", Value: float64(1)},
},
},
Arguments: []interface{}{connectArg},
})
if err != nil {
return err
@@ -217,7 +246,7 @@ func (c *Conn) initializeClient(u *url.URL, publish bool) error {
return err
}
if !publish {
if !c.Publish {
err = c.mrw.Write(&message.CommandAMF0{
ChunkStreamID: 3,
Name: "createStream",
@@ -320,20 +349,6 @@ func (c *Conn) initializeClient(u *url.URL, publish bool) error {
return readCommandResult(c.mrw, 5, "onStatus", resultIsOK1)
}
// NewServerConn initializes a server-side connection.
func NewServerConn(rw io.ReadWriter) (*Conn, *url.URL, bool, error) {
c := &Conn{
bc: bytecounter.NewReadWriter(rw),
}
u, publish, err := c.initializeServer()
if err != nil {
return nil, nil, false, err
}
return c, u, publish, nil
}
func (c *Conn) initializeServer() (*url.URL, bool, error) {
keyIn, keyOut, err := handshake.DoServer(c.bc, false)
if err != nil {
@@ -599,16 +614,6 @@ func (c *Conn) initializeServer() (*url.URL, bool, error) {
}
}
func newNoHandshakeConn(rw io.ReadWriter) *Conn {
c := &Conn{
bc: bytecounter.NewReadWriter(rw),
}
c.mrw = message.NewReadWriter(c.bc, c.bc, false)
return c
}
// BytesReceived returns the number of bytes received.
func (c *Conn) BytesReceived() uint64 {
return c.bc.Reader.Count()
internal/protocols/rtmp/conn_test.go
+58 -24
View File
@@ -57,25 +57,42 @@ func TestNewClientConn(t *testing.T) {
Value: 65536,
}, msg)
msg, err2 = mrw.Read()
require.NoError(t, err2)
require.Equal(t, &message.CommandAMF0{
ChunkStreamID: 3,
Name: "connect",
CommandID: 1,
Arguments: []interface{}{
amf0.Object{
{Key: "app", Value: "stream"},
{Key: "flashVer", Value: "LNX 9,0,124,2"},
{Key: "tcUrl", Value: "rtmp://127.0.0.1:9121/stream"},
{Key: "fpad", Value: false},
{Key: "capabilities", Value: float64(15)},
{Key: "audioCodecs", Value: float64(4071)},
{Key: "videoCodecs", Value: float64(252)},
{Key: "videoFunction", Value: float64(1)},
if ca != "publish" {
msg, err2 = mrw.Read()
require.NoError(t, err2)
require.Equal(t, &message.CommandAMF0{
ChunkStreamID: 3,
Name: "connect",
CommandID: 1,
Arguments: []interface{}{
amf0.Object{
{Key: "app", Value: "stream"},
{Key: "flashVer", Value: "LNX 9,0,124,2"},
{Key: "tcUrl", Value: "rtmp://127.0.0.1:9121/stream"},
{Key: "fpad", Value: false},
{Key: "capabilities", Value: float64(15)},
{Key: "audioCodecs", Value: float64(4071)},
{Key: "videoCodecs", Value: float64(252)},
{Key: "videoFunction", Value: float64(1)},
},
},
},
}, msg)
}, msg)
} else {
msg, err2 = mrw.Read()
require.NoError(t, err2)
require.Equal(t, &message.CommandAMF0{
ChunkStreamID: 3,
Name: "connect",
CommandID: 1,
Arguments: []interface{}{
amf0.Object{
{Key: "app", Value: "stream"},
{Key: "flashVer", Value: "LNX 9,0,124,2"},
{Key: "tcUrl", Value: "rtmp://127.0.0.1:9121/stream"},
},
},
}, msg)
}
err2 = mrw.Write(&message.CommandAMF0{
ChunkStreamID: 3,
@@ -248,7 +265,13 @@ func TestNewClientConn(t *testing.T) {
require.NoError(t, err)
defer nconn.Close()
conn, err := NewClientConn(nconn, u, ca == "publish")
conn := &Conn{
RW: nconn,
Client: true,
URL: u,
Publish: ca == "publish",
}
err = conn.Initialize()
require.NoError(t, err)
switch ca {
@@ -258,7 +281,7 @@ func TestNewClientConn(t *testing.T) {
case "publish":
require.Equal(t, uint64(3427), conn.BytesReceived())
require.Equal(t, uint64(3466), conn.BytesSent())
require.Equal(t, uint64(0xd27), conn.BytesSent())
}
<-done
@@ -284,15 +307,19 @@ func TestNewServerConn(t *testing.T) {
require.NoError(t, err2)
defer nconn.Close()
_, u, isPublishing, err2 := NewServerConn(nconn)
conn := &Conn{
RW: nconn,
Client: false,
}
err2 = conn.Initialize()
require.NoError(t, err2)
require.Equal(t, &url.URL{
Scheme: "rtmp",
Host: "127.0.0.1:9121",
Path: "//stream/",
}, u)
require.Equal(t, ca == "publish" || ca == "publish neko", isPublishing)
}, conn.URL)
require.Equal(t, ca == "publish" || ca == "publish neko", conn.Publish)
close(done)
}()
@@ -500,7 +527,14 @@ func BenchmarkRead(b *testing.B) {
})
}
conn := newNoHandshakeConn(&buf)
conn := &Conn{
RW: &buf,
skipHandshake: true,
}
err := conn.Initialize()
if err != nil {
panic(err)
}
for n := 0; n < b.N; n++ {
conn.Read() //nolint:errcheck
internal/protocols/rtmp/from_stream.go
+6 -2
View File
@@ -213,8 +213,12 @@ func FromStream(
return errNoSupportedCodecsFrom
}
var err error
w, err = NewWriter(conn, videoFormat, audioFormat)
w = &Writer{
Conn: conn,
VideoTrack: videoFormat,
AudioTrack: audioFormat,
}
err := w.Initialize()
if err != nil {
return err
}
internal/protocols/rtmp/reader.go
+8 -11
View File
@@ -272,29 +272,26 @@ func sortedKeys(m map[uint8]format.Format) []int {
// Reader is a wrapper around Conn that provides utilities to demux incoming data.
type Reader struct {
conn *Conn
Conn *Conn
videoTracks map[uint8]format.Format
audioTracks map[uint8]format.Format
onVideoData map[uint8]func(message.Message) error
onAudioData map[uint8]func(message.Message) error
}
// NewReader allocates a Reader.
func NewReader(conn *Conn) (*Reader, error) {
r := &Reader{
conn: conn,
}
// Initialize initializes a reader.
func (r *Reader) Initialize() error {
var err error
r.videoTracks, r.audioTracks, err = r.readTracks()
if err != nil {
return nil, err
return err
}
r.onVideoData = make(map[uint8]func(message.Message) error)
r.onAudioData = make(map[uint8]func(message.Message) error)
return r, nil
return nil
}
func (r *Reader) readTracks() (map[uint8]format.Format, map[uint8]format.Format, error) {
@@ -369,7 +366,7 @@ func (r *Reader) readTracks() (map[uint8]format.Format, map[uint8]format.Format,
}
for {
msg, err := r.conn.Read()
msg, err := r.Conn.Read()
if err != nil {
return nil, nil, err
}
@@ -762,7 +759,7 @@ func (r *Reader) OnDataLPCM(track *format.LPCM, cb OnDataLPCMFunc) {
// Read reads data.
func (r *Reader) Read() error {
msg, err := r.conn.Read()
msg, err := r.Conn.Read()
if err != nil {
return err
}
internal/protocols/rtmp/reader_test.go
+12 -3
View File
@@ -1630,10 +1630,19 @@ func TestReadTracks(t *testing.T) {
require.NoError(t, err)
}
c := newNoHandshakeConn(&buf)
r, err := NewReader(c)
c := &Conn{
RW: &buf,
skipHandshake: true,
}
err := c.Initialize()
require.NoError(t, err)
r := &Reader{
Conn: c,
}
err = r.Initialize()
require.NoError(t, err)
tracks := r.Tracks()
require.Equal(t, ca.tracks, tracks)
})
internal/protocols/rtmp/writer.go
+19 -21
View File
@@ -45,25 +45,23 @@ func mpeg1AudioChannels(m mpeg1audio.ChannelMode) bool {
// Writer is a wrapper around Conn that provides utilities to mux outgoing data.
type Writer struct {
conn *Conn
Conn *Conn
VideoTrack format.Format
AudioTrack format.Format
}
// NewWriter allocates a Writer.
func NewWriter(conn *Conn, videoTrack format.Format, audioTrack format.Format) (*Writer, error) {
w := &Writer{
conn: conn,
}
err := w.writeTracks(videoTrack, audioTrack)
// Initialize initializes a Writer.
func (w *Writer) Initialize() error {
err := w.writeTracks()
if err != nil {
return nil, err
return err
}
return w, nil
return nil
}
func (w *Writer) writeTracks(videoTrack format.Format, audioTrack format.Format) error {
err := w.conn.Write(&message.DataAMF0{
func (w *Writer) writeTracks() error {
err := w.Conn.Write(&message.DataAMF0{
ChunkStreamID: 4,
MessageStreamID: 0x1000000,
Payload: []interface{}{
@@ -77,7 +75,7 @@ func (w *Writer) writeTracks(videoTrack format.Format, audioTrack format.Format)
{
Key: "videocodecid",
Value: func() float64 {
switch videoTrack.(type) {
switch w.VideoTrack.(type) {
case *format.H264:
return message.CodecH264
@@ -93,7 +91,7 @@ func (w *Writer) writeTracks(videoTrack format.Format, audioTrack format.Format)
{
Key: "audiocodecid",
Value: func() float64 {
switch audioTrack.(type) {
switch w.AudioTrack.(type) {
case *format.MPEG1Audio:
return message.CodecMPEG1Audio
@@ -112,7 +110,7 @@ func (w *Writer) writeTracks(videoTrack format.Format, audioTrack format.Format)
return err
}
if videoTrack, ok := videoTrack.(*format.H264); ok {
if videoTrack, ok := w.VideoTrack.(*format.H264); ok {
// write decoder config only if SPS and PPS are available.
// if they're not available yet, they're sent later.
if sps, pps := videoTrack.SafeParams(); sps != nil && pps != nil {
@@ -121,7 +119,7 @@ func (w *Writer) writeTracks(videoTrack format.Format, audioTrack format.Format)
PPS: pps,
}.Marshal()
err = w.conn.Write(&message.Video{
err = w.Conn.Write(&message.Video{
ChunkStreamID: message.VideoChunkStreamID,
MessageStreamID: 0x1000000,
Codec: message.CodecH264,
@@ -137,7 +135,7 @@ func (w *Writer) writeTracks(videoTrack format.Format, audioTrack format.Format)
var audioConfig *mpeg4audio.AudioSpecificConfig
if track, ok := audioTrack.(*format.MPEG4Audio); ok {
if track, ok := w.AudioTrack.(*format.MPEG4Audio); ok {
audioConfig = track.GetConfig()
}
@@ -147,7 +145,7 @@ func (w *Writer) writeTracks(videoTrack format.Format, audioTrack format.Format)
return err
}
err = w.conn.Write(&message.Audio{
err = w.Conn.Write(&message.Audio{
ChunkStreamID: message.AudioChunkStreamID,
MessageStreamID: 0x1000000,
Codec: message.CodecMPEG4Audio,
@@ -172,7 +170,7 @@ func (w *Writer) WriteH264(pts time.Duration, dts time.Duration, au [][]byte) er
return err
}
return w.conn.Write(&message.Video{
return w.Conn.Write(&message.Video{
ChunkStreamID: message.VideoChunkStreamID,
MessageStreamID: 0x1000000,
Codec: message.CodecH264,
@@ -186,7 +184,7 @@ func (w *Writer) WriteH264(pts time.Duration, dts time.Duration, au [][]byte) er
// WriteMPEG4Audio writes MPEG-4 Audio data.
func (w *Writer) WriteMPEG4Audio(pts time.Duration, au []byte) error {
return w.conn.Write(&message.Audio{
return w.Conn.Write(&message.Audio{
ChunkStreamID: message.AudioChunkStreamID,
MessageStreamID: 0x1000000,
Codec: message.CodecMPEG4Audio,
@@ -201,7 +199,7 @@ func (w *Writer) WriteMPEG4Audio(pts time.Duration, au []byte) error {
// WriteMPEG1Audio writes MPEG-1 Audio data.
func (w *Writer) WriteMPEG1Audio(pts time.Duration, h *mpeg1audio.FrameHeader, frame []byte) error {
return w.conn.Write(&message.Audio{
return w.Conn.Write(&message.Audio{
ChunkStreamID: message.AudioChunkStreamID,
MessageStreamID: 0x1000000,
Codec: message.CodecMPEG1Audio,
internal/protocols/rtmp/writer_test.go
+12 -2
View File
@@ -40,9 +40,19 @@ func TestWriteTracks(t *testing.T) {
}
var buf bytes.Buffer
c := newNoHandshakeConn(&buf)
c := &Conn{
RW: &buf,
skipHandshake: true,
}
err := c.Initialize()
require.NoError(t, err)
_, err := NewWriter(c, videoTrack, audioTrack)
w := &Writer{
Conn: c,
VideoTrack: videoTrack,
AudioTrack: audioTrack,
}
err = w.Initialize()
require.NoError(t, err)
bc := bytecounter.NewReadWriter(&buf)
internal/servers/rtmp/conn.go
+15 -9
View File
@@ -137,7 +137,10 @@ func (c *conn) runInner() error {
func (c *conn) runReader() error {
c.nconn.SetReadDeadline(time.Now().Add(time.Duration(c.readTimeout)))
c.nconn.SetWriteDeadline(time.Now().Add(time.Duration(c.writeTimeout)))
conn, u, publish, err := rtmp.NewServerConn(c.nconn)
conn := &rtmp.Conn{
RW: c.nconn,
}
err := conn.Initialize()
if err != nil {
return err
}
@@ -146,14 +149,14 @@ func (c *conn) runReader() error {
c.rconn = conn
c.mutex.Unlock()
if !publish {
return c.runRead(conn, u)
if !conn.Publish {
return c.runRead(conn)
}
return c.runPublish(conn, u)
return c.runPublish(conn)
}
func (c *conn) runRead(conn *rtmp.Conn, u *url.URL) error {
pathName, query, rawQuery := pathNameAndQuery(u)
func (c *conn) runRead(conn *rtmp.Conn) error {
pathName, query, rawQuery := pathNameAndQuery(conn.URL)
path, stream, err := c.pathManager.AddReader(defs.PathAddReaderReq{
Author: c,
@@ -218,8 +221,8 @@ func (c *conn) runRead(conn *rtmp.Conn, u *url.URL) error {
}
}
func (c *conn) runPublish(conn *rtmp.Conn, u *url.URL) error {
pathName, query, rawQuery := pathNameAndQuery(u)
func (c *conn) runPublish(conn *rtmp.Conn) error {
pathName, query, rawQuery := pathNameAndQuery(conn.URL)
path, err := c.pathManager.AddPublisher(defs.PathAddPublisherReq{
Author: c,
@@ -252,7 +255,10 @@ func (c *conn) runPublish(conn *rtmp.Conn, u *url.URL) error {
c.query = rawQuery
c.mutex.Unlock()
r, err := rtmp.NewReader(conn)
r := &rtmp.Reader{
Conn: conn,
}
err = r.Initialize()
if err != nil {
return err
}
internal/servers/rtmp/server_test.go
+24 -4
View File
@@ -128,10 +128,21 @@ func TestServerPublish(t *testing.T) {
require.NoError(t, err)
defer nconn.Close()
conn, err := rtmp.NewClientConn(nconn, u, true)
conn := &rtmp.Conn{
RW: nconn,
Client: true,
URL: u,
Publish: true,
}
err = conn.Initialize()
require.NoError(t, err)
w, err := rtmp.NewWriter(conn, test.FormatH264, test.FormatMPEG4Audio)
w := &rtmp.Writer{
Conn: conn,
VideoTrack: test.FormatH264,
AudioTrack: test.FormatMPEG4Audio,
}
err = w.Initialize()
require.NoError(t, err)
err = w.WriteH264(
@@ -281,10 +292,19 @@ func TestServerRead(t *testing.T) {
})
}()
conn, err := rtmp.NewClientConn(nconn, u, false)
conn := &rtmp.Conn{
RW: nconn,
Client: true,
URL: u,
Publish: false,
}
err = conn.Initialize()
require.NoError(t, err)
r, err := rtmp.NewReader(conn)
r := &rtmp.Reader{
Conn: conn,
}
err = r.Initialize()
require.NoError(t, err)
tracks := r.Tracks()
internal/staticsources/rtmp/source.go
+11 -2
View File
@@ -90,12 +90,21 @@ func (s *Source) Run(params defs.StaticSourceRunParams) error {
func (s *Source) runReader(u *url.URL, nconn net.Conn) error {
nconn.SetReadDeadline(time.Now().Add(time.Duration(s.ReadTimeout)))
nconn.SetWriteDeadline(time.Now().Add(time.Duration(s.WriteTimeout)))
conn, err := rtmp.NewClientConn(nconn, u, false)
conn := &rtmp.Conn{
RW: nconn,
Client: true,
URL: u,
Publish: false,
}
err := conn.Initialize()
if err != nil {
return err
}
r, err := rtmp.NewReader(conn)
r := &rtmp.Reader{
Conn: conn,
}
err = r.Initialize()
if err != nil {
return err
}
internal/staticsources/rtmp/source_test.go
+10 -2
View File
@@ -48,10 +48,18 @@ func TestSource(t *testing.T) {
require.NoError(t, err)
defer nconn.Close()
conn, _, _, err := rtmp.NewServerConn(nconn)
conn := &rtmp.Conn{
RW: nconn,
}
err = conn.Initialize()
require.NoError(t, err)
w, err := rtmp.NewWriter(conn, test.FormatH264, test.FormatMPEG4Audio)
w := &rtmp.Writer{
Conn: conn,
VideoTrack: test.FormatH264,
AudioTrack: test.FormatMPEG4Audio,
}
err = w.Initialize()
require.NoError(t, err)
err = w.WriteH264(2*time.Second, 2*time.Second, [][]byte{{5, 2, 3, 4}})