From a85d8f13916084e883989d1463453a6c85204f3a Mon Sep 17 00:00:00 2001 From: Alessandro Ros Date: Sat, 10 May 2025 15:26:48 +0200 Subject: [PATCH] rtmp: fix connect command when reading (#4512) when reading, the "connect" command should not contain fpad, capabilities, audioCodecs, videoCodecs, videoFunction. --- internal/core/api_test.go | 28 +++++- internal/core/metrics_test.go | 28 +++++- internal/core/path_test.go | 42 +++++++-- internal/protocols/rtmp/conn.go | 103 +++++++++++---------- internal/protocols/rtmp/conn_test.go | 82 +++++++++++----- internal/protocols/rtmp/from_stream.go | 8 +- internal/protocols/rtmp/reader.go | 19 ++-- internal/protocols/rtmp/reader_test.go | 15 ++- internal/protocols/rtmp/writer.go | 40 ++++---- internal/protocols/rtmp/writer_test.go | 14 ++- internal/servers/rtmp/conn.go | 24 +++-- internal/servers/rtmp/server_test.go | 28 +++++- internal/staticsources/rtmp/source.go | 13 ++- internal/staticsources/rtmp/source_test.go | 12 ++- 14 files changed, 313 insertions(+), 143 deletions(-) diff --git a/internal/core/api_test.go b/internal/core/api_test.go index 277d13ec..6c2eaae6 100644 --- a/internal/core/api_test.go +++ b/internal/core/api_test.go @@ -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}}) diff --git a/internal/core/metrics_test.go b/internal/core/metrics_test.go index 5541aa67..2a6191c6 100644 --- a/internal/core/metrics_test.go +++ b/internal/core/metrics_test.go @@ -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}}) diff --git a/internal/core/path_test.go b/internal/core/path_test.go index 651b9a1a..7d1cf5dd 100644 --- a/internal/core/path_test.go +++ b/internal/core/path_test.go @@ -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": diff --git a/internal/protocols/rtmp/conn.go b/internal/protocols/rtmp/conn.go index 8a0222c3..5b34292d 100644 --- a/internal/protocols/rtmp/conn.go +++ b/internal/protocols/rtmp/conn.go @@ -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() diff --git a/internal/protocols/rtmp/conn_test.go b/internal/protocols/rtmp/conn_test.go index 3b632c82..fa1373ec 100644 --- a/internal/protocols/rtmp/conn_test.go +++ b/internal/protocols/rtmp/conn_test.go @@ -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 diff --git a/internal/protocols/rtmp/from_stream.go b/internal/protocols/rtmp/from_stream.go index b1213e44..b15ba6ed 100644 --- a/internal/protocols/rtmp/from_stream.go +++ b/internal/protocols/rtmp/from_stream.go @@ -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 } diff --git a/internal/protocols/rtmp/reader.go b/internal/protocols/rtmp/reader.go index 69dddae1..d8ff52c9 100644 --- a/internal/protocols/rtmp/reader.go +++ b/internal/protocols/rtmp/reader.go @@ -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 } diff --git a/internal/protocols/rtmp/reader_test.go b/internal/protocols/rtmp/reader_test.go index 886b1488..f53ffb2d 100644 --- a/internal/protocols/rtmp/reader_test.go +++ b/internal/protocols/rtmp/reader_test.go @@ -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) }) diff --git a/internal/protocols/rtmp/writer.go b/internal/protocols/rtmp/writer.go index e36bed6b..ca1dd99a 100644 --- a/internal/protocols/rtmp/writer.go +++ b/internal/protocols/rtmp/writer.go @@ -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, diff --git a/internal/protocols/rtmp/writer_test.go b/internal/protocols/rtmp/writer_test.go index e687e919..45bb3aca 100644 --- a/internal/protocols/rtmp/writer_test.go +++ b/internal/protocols/rtmp/writer_test.go @@ -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) diff --git a/internal/servers/rtmp/conn.go b/internal/servers/rtmp/conn.go index 4fff5b0d..c8520249 100644 --- a/internal/servers/rtmp/conn.go +++ b/internal/servers/rtmp/conn.go @@ -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 } diff --git a/internal/servers/rtmp/server_test.go b/internal/servers/rtmp/server_test.go index 25ce0c71..e107773a 100644 --- a/internal/servers/rtmp/server_test.go +++ b/internal/servers/rtmp/server_test.go @@ -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() diff --git a/internal/staticsources/rtmp/source.go b/internal/staticsources/rtmp/source.go index 2a1de34c..36e62a28 100644 --- a/internal/staticsources/rtmp/source.go +++ b/internal/staticsources/rtmp/source.go @@ -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 } diff --git a/internal/staticsources/rtmp/source_test.go b/internal/staticsources/rtmp/source_test.go index 15ccbcb8..9dbb4fa2 100644 --- a/internal/staticsources/rtmp/source_test.go +++ b/internal/staticsources/rtmp/source_test.go @@ -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}})