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

rtmp: add primitives

pull/956/head
aler9 4 years ago
parent
13fb24da39
commit
046a3f248d
9 changed files with 535 additions and 420 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. 27
      internal/rtmp/base/base.go
  2. 68
      internal/rtmp/base/chunk0.go
  3. 35
      internal/rtmp/base/chunk1.go
  4. 29
      internal/rtmp/base/chunk3.go
  5. 14
      internal/rtmp/base/handshakec0.go
  6. 45
      internal/rtmp/base/handshakec1.go
  7. 48
      internal/rtmp/base/handshakec2.go
  8. 24
      internal/rtmp/base/handshakes0.go
  9. 667
      internal/rtmp/conn_test.go

27
internal/rtmp/base/base.go
Unescape View File

@ -0,0 +1,27 @@
package base
const (
rtmpVersion = 0x03
)
var (
hsClientFullKey = []byte{
'G', 'e', 'n', 'u', 'i', 'n', 'e', ' ', 'A', 'd', 'o', 'b', 'e', ' ',
'F', 'l', 'a', 's', 'h', ' ', 'P', 'l', 'a', 'y', 'e', 'r', ' ',
'0', '0', '1',
0xF0, 0xEE, 0xC2, 0x4A, 0x80, 0x68, 0xBE, 0xE8, 0x2E, 0x00, 0xD0, 0xD1,
0x02, 0x9E, 0x7E, 0x57, 0x6E, 0xEC, 0x5D, 0x2D, 0x29, 0x80, 0x6F, 0xAB,
0x93, 0xB8, 0xE6, 0x36, 0xCF, 0xEB, 0x31, 0xAE,
}
hsServerFullKey = []byte{
'G', 'e', 'n', 'u', 'i', 'n', 'e', ' ', 'A', 'd', 'o', 'b', 'e', ' ',
'F', 'l', 'a', 's', 'h', ' ', 'M', 'e', 'd', 'i', 'a', ' ',
'S', 'e', 'r', 'v', 'e', 'r', ' ',
'0', '0', '1',
0xF0, 0xEE, 0xC2, 0x4A, 0x80, 0x68, 0xBE, 0xE8, 0x2E, 0x00, 0xD0, 0xD1,
0x02, 0x9E, 0x7E, 0x57, 0x6E, 0xEC, 0x5D, 0x2D, 0x29, 0x80, 0x6F, 0xAB,
0x93, 0xB8, 0xE6, 0x36, 0xCF, 0xEB, 0x31, 0xAE,
}
hsClientPartialKey = hsClientFullKey[:30]
hsServerPartialKey = hsServerFullKey[:36]
)

68
internal/rtmp/base/chunk0.go
Unescape View File

@ -0,0 +1,68 @@
package base
import (
"fmt"
"io"
)
// Chunk0 is a type 0 chunk.
// This type MUST be used at
// the start of a chunk stream, and whenever the stream timestamp goes
// backward (e.g., because of a backward seek).
type Chunk0 struct {
ChunkStreamID byte
Timestamp uint32
Typ byte
StreamID uint32
BodyLen uint32
Body []byte
}
// Read reads the chunk.
func (m *Chunk0) Read(r io.Reader, chunkMaxBodyLen int) error {
header := make([]byte, 12)
_, err := r.Read(header)
if err != nil {
return err
}
if header[0]>>6 != 0 {
return fmt.Errorf("wrong chunk header type")
}
m.ChunkStreamID = header[0] & 0x3F
m.Timestamp = uint32(header[3])<<16 | uint32(header[2])<<8 | uint32(header[1])
m.BodyLen = uint32(header[4])<<16 | uint32(header[5])<<8 | uint32(header[6])
m.Typ = header[7]
m.StreamID = uint32(header[8])<<24 | uint32(header[9])<<16 | uint32(header[10])<<8 | uint32(header[11])
chunkBodyLen := int(m.BodyLen)
if chunkBodyLen > chunkMaxBodyLen {
chunkBodyLen = chunkMaxBodyLen
}
m.Body = make([]byte, chunkBodyLen)
_, err = r.Read(m.Body)
return err
}
// Write writes the chunk.
func (m Chunk0) Write(w io.Writer) error {
header := make([]byte, 12)
header[0] = m.ChunkStreamID
header[1] = byte(m.Timestamp >> 16)
header[2] = byte(m.Timestamp >> 8)
header[3] = byte(m.Timestamp)
header[4] = byte(m.BodyLen >> 16)
header[5] = byte(m.BodyLen >> 8)
header[6] = byte(m.BodyLen)
header[7] = m.Typ
header[8] = byte(m.StreamID)
_, err := w.Write(header)
if err != nil {
return err
}
_, err = w.Write(m.Body)
return err
}

35
internal/rtmp/base/chunk1.go
Unescape View File

@ -0,0 +1,35 @@
package base
import (
"io"
)
// Chunk1 is a type 1 chunk.
// The message stream ID is not
// included; this chunk takes the same stream ID as the preceding chunk.
// Streams with variable-sized messages (for example, many video
// formats) SHOULD use this format for the first chunk of each new
// message after the first.
type Chunk1 struct {
ChunkStreamID byte
Typ byte
Body []byte
}
// Write writes the chunk.
func (m Chunk1) Write(w io.Writer) error {
header := make([]byte, 8)
header[0] = 1<<6 | m.ChunkStreamID
l := uint32(len(m.Body))
header[4] = byte(l >> 16)
header[5] = byte(l >> 8)
header[6] = byte(l)
header[7] = m.Typ
_, err := w.Write(header)
if err != nil {
return err
}
_, err = w.Write(m.Body)
return err
}

29
internal/rtmp/base/chunk3.go
Unescape View File

@ -0,0 +1,29 @@
package base
import (
"io"
)
// Chunk3 is a type 3 chunk.
// Type 3 chunks have no message header. The stream ID, message length
// and timestamp delta fields are not present; chunks of this type take
// values from the preceding chunk for the same Chunk Stream ID. When a
// single message is split into chunks, all chunks of a message except
// the first one SHOULD use this type.
type Chunk3 struct {
ChunkStreamID byte
Body []byte
}
// Write writes the chunk.
func (m Chunk3) Write(w io.Writer) error {
header := make([]byte, 1)
header[0] = 3<<6 | m.ChunkStreamID
_, err := w.Write(header)
if err != nil {
return err
}
_, err = w.Write(m.Body)
return err
}

14
internal/rtmp/base/handshakec0.go
Unescape View File

@ -0,0 +1,14 @@
package base
import (
"io"
)
// HandshakeC0 is the C0 part of an handshake.
type HandshakeC0 struct{}
// Read reads a HandshakeC0.
func (HandshakeC0) Write(w io.Writer) error {
_, err := w.Write([]byte{rtmpVersion})
return err
}

45
internal/rtmp/base/handshakec1.go
Unescape View File

@ -0,0 +1,45 @@
package base
import (
"crypto/hmac"
"crypto/rand"
"crypto/sha256"
"io"
)
func hsCalcDigestPos(p []byte, base int) (pos int) {
for i := 0; i < 4; i++ {
pos += int(p[base+i])
}
pos = (pos % 728) + base + 4
return
}
func hsMakeDigest(key []byte, src []byte, gap int) (dst []byte) {
h := hmac.New(sha256.New, key)
if gap <= 0 {
h.Write(src)
} else {
h.Write(src[:gap])
h.Write(src[gap+32:])
}
return h.Sum(nil)
}
// HandshakeC1 is the C1 part of an handshake.
type HandshakeC1 struct{}
// Read reads a HandshakeC1.
func (HandshakeC1) Write(w io.Writer) error {
buf := make([]byte, 1536)
copy(buf[0:4], []byte{0x00, 0x00, 0x00, 0x00})
copy(buf[4:8], []byte{0x09, 0x00, 0x7c, 0x02})
rand.Read(buf[8:])
gap := hsCalcDigestPos(buf[0:], 8)
digest := hsMakeDigest(hsClientPartialKey, buf[0:], gap)
copy(buf[gap+0:], digest)
_, err := w.Write(buf[0:])
return err
}

48
internal/rtmp/base/handshakec2.go
Unescape View File

@ -0,0 +1,48 @@
package base
import (
"bytes"
"crypto/rand"
"fmt"
"io"
)
func hsFindDigest(p []byte, key []byte, base int) int {
gap := hsCalcDigestPos(p, base)
digest := hsMakeDigest(key, p, gap)
if !bytes.Equal(p[gap:gap+32], digest) {
return -1
}
return gap
}
func hsParse1(p []byte, peerkey []byte, key []byte) (ok bool, digest []byte) {
var pos int
if pos = hsFindDigest(p, peerkey, 772); pos == -1 {
if pos = hsFindDigest(p, peerkey, 8); pos == -1 {
return
}
}
ok = true
digest = hsMakeDigest(key, p[pos:pos+32], -1)
return
}
// HandshakeC2 is the C2 part of an handshake.
type HandshakeC2 struct{}
// Read reads a HandshakeC2.
func (HandshakeC2) Write(w io.Writer, s1s2 []byte) error {
ok, key := hsParse1(s1s2[:1536], hsServerPartialKey, hsClientFullKey)
if !ok {
return fmt.Errorf("unable to parse S1+S2")
}
buf := make([]byte, 1536)
rand.Read(buf)
gap := len(buf) - 32
digest := hsMakeDigest(key, buf, gap)
copy(buf[gap:], digest)
_, err := w.Write(buf)
return err
}

24
internal/rtmp/base/handshakes0.go
Unescape View File

@ -0,0 +1,24 @@
package base
import (
"fmt"
"io"
)
// HandshakeS0 is the S0 part of an handshake.
type HandshakeS0 struct{}
// Read reads a HandshakeS0.
func (HandshakeS0) Read(r io.Reader) error {
buf := make([]byte, 1)
_, err := io.ReadFull(r, buf)
if err != nil {
return err
}
if buf[0] != rtmpVersion {
return fmt.Errorf("invalid rtmp version (%d)", buf[0])
}
return nil
}

667
internal/rtmp/conn_test.go
Unescape View File

@ -1,12 +1,6 @@
package rtmp package rtmp
import ( import (
"bytes"
"crypto/hmac"
"crypto/rand"
"crypto/sha256"
"fmt"
"io"
"net" "net"
"net/url" "net/url"
"strings" "strings"
@ -17,28 +11,8 @@ import (
nh264 "github.com/notedit/rtmp/codec/h264" nh264 "github.com/notedit/rtmp/codec/h264"
"github.com/notedit/rtmp/format/flv/flvio" "github.com/notedit/rtmp/format/flv/flvio"
"github.com/stretchr/testify/require" "github.com/stretchr/testify/require"
)
var ( "github.com/aler9/rtsp-simple-server/internal/rtmp/base"
hsClientFullKey = []byte{
'G', 'e', 'n', 'u', 'i', 'n', 'e', ' ', 'A', 'd', 'o', 'b', 'e', ' ',
'F', 'l', 'a', 's', 'h', ' ', 'P', 'l', 'a', 'y', 'e', 'r', ' ',
'0', '0', '1',
0xF0, 0xEE, 0xC2, 0x4A, 0x80, 0x68, 0xBE, 0xE8, 0x2E, 0x00, 0xD0, 0xD1,
0x02, 0x9E, 0x7E, 0x57, 0x6E, 0xEC, 0x5D, 0x2D, 0x29, 0x80, 0x6F, 0xAB,
0x93, 0xB8, 0xE6, 0x36, 0xCF, 0xEB, 0x31, 0xAE,
}
hsServerFullKey = []byte{
'G', 'e', 'n', 'u', 'i', 'n', 'e', ' ', 'A', 'd', 'o', 'b', 'e', ' ',
'F', 'l', 'a', 's', 'h', ' ', 'M', 'e', 'd', 'i', 'a', ' ',
'S', 'e', 'r', 'v', 'e', 'r', ' ',
'0', '0', '1',
0xF0, 0xEE, 0xC2, 0x4A, 0x80, 0x68, 0xBE, 0xE8, 0x2E, 0x00, 0xD0, 0xD1,
0x02, 0x9E, 0x7E, 0x57, 0x6E, 0xEC, 0x5D, 0x2D, 0x29, 0x80, 0x6F, 0xAB,
0x93, 0xB8, 0xE6, 0x36, 0xCF, 0xEB, 0x31, 0xAE,
}
hsClientPartialKey = hsClientFullKey[:30]
hsServerPartialKey = hsServerFullKey[:36]
) )
func splitPath(u *url.URL) (app, stream string) { func splitPath(u *url.URL) (app, stream string) {
@ -60,75 +34,6 @@ func splitPath(u *url.URL) (app, stream string) {
return return
} }
func hsMakeDigest(key []byte, src []byte, gap int) (dst []byte) {
h := hmac.New(sha256.New, key)
if gap <= 0 {
h.Write(src)
} else {
h.Write(src[:gap])
h.Write(src[gap+32:])
}
return h.Sum(nil)
}
func hsCalcDigestPos(p []byte, base int) (pos int) {
for i := 0; i < 4; i++ {
pos += int(p[base+i])
}
pos = (pos % 728) + base + 4
return
}
func hsFindDigest(p []byte, key []byte, base int) int {
gap := hsCalcDigestPos(p, base)
digest := hsMakeDigest(key, p, gap)
if !bytes.Equal(p[gap:gap+32], digest) {
return -1
}
return gap
}
func hsParse1(p []byte, peerkey []byte, key []byte) (ok bool, digest []byte) {
var pos int
if pos = hsFindDigest(p, peerkey, 772); pos == -1 {
if pos = hsFindDigest(p, peerkey, 8); pos == -1 {
return
}
}
ok = true
digest = hsMakeDigest(key, p[pos:pos+32], -1)
return
}
func writeHandshakeC2(w io.Writer, s0s1s2 []byte) error {
ok, key := hsParse1(s0s1s2[1:1537], hsServerPartialKey, hsClientFullKey)
if !ok {
panic("unable to parse s0s1s2")
}
buf := make([]byte, 1536)
rand.Read(buf)
gap := len(buf) - 32
digest := hsMakeDigest(key, buf, gap)
copy(buf[gap:], digest)
_, err := w.Write(buf)
return err
}
func writeHandshakeC0C1(w io.Writer) error {
buf := make([]byte, 1537)
buf[0] = 0x03
copy(buf[1:5], []byte{0x00, 0x00, 0x00, 0x00})
copy(buf[5:9], []byte{0x09, 0x00, 0x7c, 0x02})
rand.Read(buf[1+8:])
gap := hsCalcDigestPos(buf[1:], 8)
digest := hsMakeDigest(hsClientPartialKey, buf[1:], gap)
copy(buf[gap+1:], digest)
_, err := w.Write(buf)
return err
}
func getTcURL(u string) string { func getTcURL(u string) string {
ur, err := url.Parse(u) ur, err := url.Parse(u)
if err != nil { if err != nil {
@ -141,102 +46,6 @@ func getTcURL(u string) string {
return nu.String() + app return nu.String() + app
} }
type chunk0 struct {
chunkStreamID byte
timestamp uint32
typ byte
streamID uint32
bodyLen uint32
body []byte
}
func (m *chunk0) read(r io.Reader, chunkMaxBodyLen int) error {
header := make([]byte, 12)
_, err := r.Read(header)
if err != nil {
return err
}
if header[0]>>6 != 0 {
return fmt.Errorf("wrong chunk header type")
}
m.chunkStreamID = header[0] & 0x3F
m.timestamp = uint32(header[3])<<16 | uint32(header[2])<<8 | uint32(header[1])
m.bodyLen = uint32(header[4])<<16 | uint32(header[5])<<8 | uint32(header[6])
m.typ = header[7]
m.streamID = uint32(header[8])<<24 | uint32(header[9])<<16 | uint32(header[10])<<8 | uint32(header[11])
chunkBodyLen := int(m.bodyLen)
if chunkBodyLen > chunkMaxBodyLen {
chunkBodyLen = chunkMaxBodyLen
}
m.body = make([]byte, chunkBodyLen)
_, err = r.Read(m.body)
return err
}
func (m chunk0) write(w io.Writer) error {
header := make([]byte, 12)
header[0] = m.chunkStreamID
header[1] = byte(m.timestamp >> 16)
header[2] = byte(m.timestamp >> 8)
header[3] = byte(m.timestamp)
header[4] = byte(m.bodyLen >> 16)
header[5] = byte(m.bodyLen >> 8)
header[6] = byte(m.bodyLen)
header[7] = m.typ
header[8] = byte(m.streamID)
_, err := w.Write(header)
if err != nil {
return err
}
_, err = w.Write(m.body)
return err
}
type chunk1 struct {
chunkStreamID byte
typ byte
body []byte
}
func (m chunk1) write(w io.Writer) error {
header := make([]byte, 8)
header[0] = 1<<6 | m.chunkStreamID
l := uint32(len(m.body))
header[4] = byte(l >> 16)
header[5] = byte(l >> 8)
header[6] = byte(l)
header[7] = m.typ
_, err := w.Write(header)
if err != nil {
return err
}
_, err = w.Write(m.body)
return err
}
type chunk3 struct {
chunkStreamID byte
body []byte
}
func (m chunk3) write(w io.Writer) error {
header := make([]byte, 1)
header[0] = 3<<6 | m.chunkStreamID
_, err := w.Write(header)
if err != nil {
return err
}
_, err = w.Write(m.body)
return err
}
func TestReadTracks(t *testing.T) { func TestReadTracks(t *testing.T) {
sps := []byte{ sps := []byte{
0x67, 0x64, 0x00, 0x0c, 0xac, 0x3b, 0x50, 0xb0, 0x67, 0x64, 0x00, 0x0c, 0xac, 0x3b, 0x50, 0xb0,
@ -304,17 +113,25 @@ func TestReadTracks(t *testing.T) {
require.NoError(t, err) require.NoError(t, err)
defer conn.Close() defer conn.Close()
// C->S handshake C0+C1 // C->S handshake C0
err = writeHandshakeC0C1(conn) err = base.HandshakeC0{}.Write(conn)
require.NoError(t, err) require.NoError(t, err)
// S->C handshake S0+S1+S2 // C->S handshake C1
s0s1s2 := make([]byte, 1536*2+1) err = base.HandshakeC1{}.Write(conn)
_, err = conn.Read(s0s1s2) require.NoError(t, err)
// S->C handshake S0
err = base.HandshakeS0{}.Read(conn)
require.NoError(t, err)
// S->C handshake S1+S2
s1s2 := make([]byte, 1536*2)
_, err = conn.Read(s1s2)
require.NoError(t, err) require.NoError(t, err)
// C->S handshake C2 // C->S handshake C2
err = writeHandshakeC2(conn, s0s1s2) err = base.HandshakeC2{}.Write(conn, s1s2)
require.NoError(t, err) require.NoError(t, err)
// C->S connect // C->S connect
@ -332,56 +149,56 @@ func TestReadTracks(t *testing.T) {
{K: "videoFunction", V: 1}, {K: "videoFunction", V: 1},
}, },
}) })
err = chunk0{ err = base.Chunk0{
chunkStreamID: 3, ChunkStreamID: 3,
typ: 0x14, Typ: 0x14,
bodyLen: uint32(len(byts)), BodyLen: uint32(len(byts)),
body: byts[:128], Body: byts[:128],
}.write(conn) }.Write(conn)
require.NoError(t, err) require.NoError(t, err)
err = chunk3{ err = base.Chunk3{
chunkStreamID: 3, ChunkStreamID: 3,
body: byts[128:], Body: byts[128:],
}.write(conn) }.Write(conn)
require.NoError(t, err) require.NoError(t, err)
// S->C window acknowledgement size // S->C window acknowledgement size
var c0 chunk0 var c0 base.Chunk0
err = c0.read(conn, 128) err = c0.Read(conn, 128)
require.NoError(t, err) require.NoError(t, err)
require.Equal(t, chunk0{ require.Equal(t, base.Chunk0{
chunkStreamID: 2, ChunkStreamID: 2,
typ: 5, Typ: 5,
bodyLen: 4, BodyLen: 4,
body: []byte{0x00, 38, 37, 160}, Body: []byte{0x00, 38, 37, 160},
}, c0) }, c0)
// S->C set peer bandwidth // S->C set peer bandwidth
err = c0.read(conn, 128) err = c0.Read(conn, 128)
require.NoError(t, err) require.NoError(t, err)
require.Equal(t, chunk0{ require.Equal(t, base.Chunk0{
chunkStreamID: 2, ChunkStreamID: 2,
typ: 6, Typ: 6,
bodyLen: 5, BodyLen: 5,
body: []byte{0x00, 0x26, 0x25, 0xa0, 0x02}, Body: []byte{0x00, 0x26, 0x25, 0xa0, 0x02},
}, c0) }, c0)
// S->C set chunk size // S->C set chunk size
err = c0.read(conn, 128) err = c0.Read(conn, 128)
require.NoError(t, err) require.NoError(t, err)
require.Equal(t, chunk0{ require.Equal(t, base.Chunk0{
chunkStreamID: 2, ChunkStreamID: 2,
typ: 1, Typ: 1,
bodyLen: 4, BodyLen: 4,
body: []byte{0x00, 0x01, 0x00, 0x00}, Body: []byte{0x00, 0x01, 0x00, 0x00},
}, c0) }, c0)
// S->C result // S->C result
err = c0.read(conn, 65536) err = c0.Read(conn, 65536)
require.NoError(t, err) require.NoError(t, err)
require.Equal(t, uint8(3), c0.chunkStreamID) require.Equal(t, uint8(3), c0.ChunkStreamID)
require.Equal(t, uint8(0x14), c0.typ) require.Equal(t, uint8(0x14), c0.Typ)
arr, err := flvio.ParseAMFVals(c0.body, false) arr, err := flvio.ParseAMFVals(c0.Body, false)
require.NoError(t, err) require.NoError(t, err)
require.Equal(t, []interface{}{ require.Equal(t, []interface{}{
"_result", "_result",
@ -399,57 +216,57 @@ func TestReadTracks(t *testing.T) {
}, arr) }, arr)
// C->S set chunk size // C->S set chunk size
err = chunk0{ err = base.Chunk0{
chunkStreamID: 2, ChunkStreamID: 2,
typ: 1, Typ: 1,
bodyLen: 4, BodyLen: 4,
body: []byte{0x00, 0x01, 0x00, 0x00}, Body: []byte{0x00, 0x01, 0x00, 0x00},
}.write(conn) }.Write(conn)
require.NoError(t, err) require.NoError(t, err)
// C->S releaseStream // C->S releaseStream
err = chunk1{ err = base.Chunk1{
chunkStreamID: 3, ChunkStreamID: 3,
typ: 0x14, Typ: 0x14,
body: flvio.FillAMF0ValsMalloc([]interface{}{ Body: flvio.FillAMF0ValsMalloc([]interface{}{
"releaseStream", "releaseStream",
float64(2), float64(2),
nil, nil,
"", "",
}), }),
}.write(conn) }.Write(conn)
require.NoError(t, err) require.NoError(t, err)
// C->S FCPublish // C->S FCPublish
err = chunk1{ err = base.Chunk1{
chunkStreamID: 3, ChunkStreamID: 3,
typ: 0x14, Typ: 0x14,
body: flvio.FillAMF0ValsMalloc([]interface{}{ Body: flvio.FillAMF0ValsMalloc([]interface{}{
"FCPublish", "FCPublish",
float64(3), float64(3),
nil, nil,
"", "",
}), }),
}.write(conn) }.Write(conn)
require.NoError(t, err) require.NoError(t, err)
// C->S createStream // C->S createStream
err = chunk3{ err = base.Chunk3{
chunkStreamID: 3, ChunkStreamID: 3,
body: flvio.FillAMF0ValsMalloc([]interface{}{ Body: flvio.FillAMF0ValsMalloc([]interface{}{
"createStream", "createStream",
float64(4), float64(4),
nil, nil,
}), }),
}.write(conn) }.Write(conn)
require.NoError(t, err) require.NoError(t, err)
// S->C result // S->C result
err = c0.read(conn, 65536) err = c0.Read(conn, 65536)
require.NoError(t, err) require.NoError(t, err)
require.Equal(t, uint8(3), c0.chunkStreamID) require.Equal(t, uint8(3), c0.ChunkStreamID)
require.Equal(t, uint8(0x14), c0.typ) require.Equal(t, uint8(0x14), c0.Typ)
arr, err = flvio.ParseAMFVals(c0.body, false) arr, err = flvio.ParseAMFVals(c0.Body, false)
require.NoError(t, err) require.NoError(t, err)
require.Equal(t, []interface{}{ require.Equal(t, []interface{}{
"_result", "_result",
@ -466,21 +283,21 @@ func TestReadTracks(t *testing.T) {
"", "",
"live", "live",
}) })
err = chunk0{ err = base.Chunk0{
chunkStreamID: 8, ChunkStreamID: 8,
typ: 0x14, Typ: 0x14,
streamID: 1, StreamID: 1,
bodyLen: uint32(len(byts)), BodyLen: uint32(len(byts)),
body: byts, Body: byts,
}.write(conn) }.Write(conn)
require.NoError(t, err) require.NoError(t, err)
// S->C onStatus // S->C onStatus
err = c0.read(conn, 65536) err = c0.Read(conn, 65536)
require.NoError(t, err) require.NoError(t, err)
require.Equal(t, uint8(5), c0.chunkStreamID) require.Equal(t, uint8(5), c0.ChunkStreamID)
require.Equal(t, uint8(0x14), c0.typ) require.Equal(t, uint8(0x14), c0.Typ)
arr, err = flvio.ParseAMFVals(c0.body, false) arr, err = flvio.ParseAMFVals(c0.Body, false)
require.NoError(t, err) require.NoError(t, err)
require.Equal(t, []interface{}{ require.Equal(t, []interface{}{
"onStatus", "onStatus",
@ -518,13 +335,13 @@ func TestReadTracks(t *testing.T) {
}, },
}, },
}) })
err = chunk0{ err = base.Chunk0{
chunkStreamID: 4, ChunkStreamID: 4,
typ: 0x12, Typ: 0x12,
streamID: 1, StreamID: 1,
bodyLen: uint32(len(byts)), BodyLen: uint32(len(byts)),
body: byts, Body: byts,
}.write(conn) }.Write(conn)
require.NoError(t, err) require.NoError(t, err)
// C->S H264 decoder config // C->S H264 decoder config
@ -540,13 +357,13 @@ func TestReadTracks(t *testing.T) {
var n int var n int
codec.ToConfig(b, &n) codec.ToConfig(b, &n)
body := append([]byte{flvio.FRAME_KEY<<4 | flvio.VIDEO_H264, 0, 0, 0, 0}, b[:n]...) body := append([]byte{flvio.FRAME_KEY<<4 | flvio.VIDEO_H264, 0, 0, 0, 0}, b[:n]...)
err = chunk0{ err = base.Chunk0{
chunkStreamID: 6, ChunkStreamID: 6,
typ: flvio.TAG_VIDEO, Typ: flvio.TAG_VIDEO,
streamID: 1, StreamID: 1,
bodyLen: uint32(len(body)), BodyLen: uint32(len(body)),
body: body, Body: body,
}.write(conn) }.Write(conn)
require.NoError(t, err) require.NoError(t, err)
// C->S AAC decoder config // C->S AAC decoder config
@ -556,16 +373,16 @@ func TestReadTracks(t *testing.T) {
ChannelCount: 2, ChannelCount: 2,
}.Encode() }.Encode()
require.NoError(t, err) require.NoError(t, err)
err = chunk0{ err = base.Chunk0{
chunkStreamID: 4, ChunkStreamID: 4,
typ: flvio.TAG_AUDIO, Typ: flvio.TAG_AUDIO,
streamID: 1, StreamID: 1,
bodyLen: uint32(len(enc) + 2), BodyLen: uint32(len(enc) + 2),
body: append([]byte{ Body: append([]byte{
flvio.SOUND_AAC<<4 | flvio.SOUND_44Khz<<2 | flvio.SOUND_16BIT<<1 | flvio.SOUND_STEREO, flvio.SOUND_AAC<<4 | flvio.SOUND_44Khz<<2 | flvio.SOUND_16BIT<<1 | flvio.SOUND_STEREO,
flvio.AAC_SEQHDR, flvio.AAC_SEQHDR,
}, enc...), }, enc...),
}.write(conn) }.Write(conn)
require.NoError(t, err) require.NoError(t, err)
case "metadata without codec id": case "metadata without codec id":
@ -588,13 +405,13 @@ func TestReadTracks(t *testing.T) {
}, },
}, },
}) })
err = chunk0{ err = base.Chunk0{
chunkStreamID: 4, ChunkStreamID: 4,
typ: 0x12, Typ: 0x12,
streamID: 1, StreamID: 1,
bodyLen: uint32(len(byts)), BodyLen: uint32(len(byts)),
body: byts, Body: byts,
}.write(conn) }.Write(conn)
require.NoError(t, err) require.NoError(t, err)
// C->S H264 decoder config // C->S H264 decoder config
@ -610,13 +427,13 @@ func TestReadTracks(t *testing.T) {
var n int var n int
codec.ToConfig(b, &n) codec.ToConfig(b, &n)
body := append([]byte{flvio.FRAME_KEY<<4 | flvio.VIDEO_H264, 0, 0, 0, 0}, b[:n]...) body := append([]byte{flvio.FRAME_KEY<<4 | flvio.VIDEO_H264, 0, 0, 0, 0}, b[:n]...)
err = chunk0{ err = base.Chunk0{
chunkStreamID: 6, ChunkStreamID: 6,
typ: flvio.TAG_VIDEO, Typ: flvio.TAG_VIDEO,
streamID: 1, StreamID: 1,
bodyLen: uint32(len(body)), BodyLen: uint32(len(body)),
body: body, Body: body,
}.write(conn) }.Write(conn)
require.NoError(t, err) require.NoError(t, err)
case "no metadata": case "no metadata":
@ -633,13 +450,13 @@ func TestReadTracks(t *testing.T) {
var n int var n int
codec.ToConfig(b, &n) codec.ToConfig(b, &n)
body := append([]byte{flvio.FRAME_KEY<<4 | flvio.VIDEO_H264, 0, 0, 0, 0}, b[:n]...) body := append([]byte{flvio.FRAME_KEY<<4 | flvio.VIDEO_H264, 0, 0, 0, 0}, b[:n]...)
err = chunk0{ err = base.Chunk0{
chunkStreamID: 6, ChunkStreamID: 6,
typ: flvio.TAG_VIDEO, Typ: flvio.TAG_VIDEO,
streamID: 1, StreamID: 1,
bodyLen: uint32(len(body)), BodyLen: uint32(len(body)),
body: body, Body: body,
}.write(conn) }.Write(conn)
require.NoError(t, err) require.NoError(t, err)
} }
@ -685,17 +502,25 @@ func TestWriteTracks(t *testing.T) {
require.NoError(t, err) require.NoError(t, err)
defer conn.Close() defer conn.Close()
// C->S handshake C0+C1 // C->S handshake C0
err = writeHandshakeC0C1(conn) err = base.HandshakeC0{}.Write(conn)
require.NoError(t, err)
// C-> handshake C1
err = base.HandshakeC1{}.Write(conn)
require.NoError(t, err)
// S->C handshake S0
err = base.HandshakeS0{}.Read(conn)
require.NoError(t, err) require.NoError(t, err)
// S->C handshake S0+S1+S2 // S->C handshake S1+S2
s0s1s2 := make([]byte, 1536*2+1) s1s2 := make([]byte, 1536*2)
_, err = conn.Read(s0s1s2) _, err = conn.Read(s1s2)
require.NoError(t, err) require.NoError(t, err)
// C->S handshake C2 // C->S handshake C2
err = writeHandshakeC2(conn, s0s1s2) err = base.HandshakeC2{}.Write(conn, s1s2)
require.NoError(t, err) require.NoError(t, err)
// C->S connect // C->S connect
@ -713,56 +538,56 @@ func TestWriteTracks(t *testing.T) {
{K: "videoFunction", V: 1}, {K: "videoFunction", V: 1},
}, },
}) })
err = chunk0{ err = base.Chunk0{
chunkStreamID: 3, ChunkStreamID: 3,
typ: 0x14, Typ: 0x14,
bodyLen: uint32(len(byts)), BodyLen: uint32(len(byts)),
body: byts[:128], Body: byts[:128],
}.write(conn) }.Write(conn)
require.NoError(t, err) require.NoError(t, err)
err = chunk3{ err = base.Chunk3{
chunkStreamID: 3, ChunkStreamID: 3,
body: byts[128:], Body: byts[128:],
}.write(conn) }.Write(conn)
require.NoError(t, err) require.NoError(t, err)
// S->C window acknowledgement size // S->C window acknowledgement size
var c0 chunk0 var c0 base.Chunk0
err = c0.read(conn, 128) err = c0.Read(conn, 128)
require.NoError(t, err) require.NoError(t, err)
require.Equal(t, chunk0{ require.Equal(t, base.Chunk0{
chunkStreamID: 2, ChunkStreamID: 2,
typ: 5, Typ: 5,
bodyLen: 4, BodyLen: 4,
body: []byte{0x00, 38, 37, 160}, Body: []byte{0x00, 38, 37, 160},
}, c0) }, c0)
// S->C set peer bandwidth // S->C set peer bandwidth
err = c0.read(conn, 128) err = c0.Read(conn, 128)
require.NoError(t, err) require.NoError(t, err)
require.Equal(t, chunk0{ require.Equal(t, base.Chunk0{
chunkStreamID: 2, ChunkStreamID: 2,
typ: 6, Typ: 6,
bodyLen: 5, BodyLen: 5,
body: []byte{0x00, 0x26, 0x25, 0xa0, 0x02}, Body: []byte{0x00, 0x26, 0x25, 0xa0, 0x02},
}, c0) }, c0)
// S->C set chunk size // S->C set chunk size
err = c0.read(conn, 128) err = c0.Read(conn, 128)
require.NoError(t, err) require.NoError(t, err)
require.Equal(t, chunk0{ require.Equal(t, base.Chunk0{
chunkStreamID: 2, ChunkStreamID: 2,
typ: 1, Typ: 1,
bodyLen: 4, BodyLen: 4,
body: []byte{0x00, 0x01, 0x00, 0x00}, Body: []byte{0x00, 0x01, 0x00, 0x00},
}, c0) }, c0)
// S->C result // S->C result
err = c0.read(conn, 65536) err = c0.Read(conn, 65536)
require.NoError(t, err) require.NoError(t, err)
require.Equal(t, uint8(3), c0.chunkStreamID) require.Equal(t, uint8(3), c0.ChunkStreamID)
require.Equal(t, uint8(0x14), c0.typ) require.Equal(t, uint8(0x14), c0.Typ)
arr, err := flvio.ParseAMFVals(c0.body, false) arr, err := flvio.ParseAMFVals(c0.Body, false)
require.NoError(t, err) require.NoError(t, err)
require.Equal(t, []interface{}{ require.Equal(t, []interface{}{
"_result", "_result",
@ -780,41 +605,41 @@ func TestWriteTracks(t *testing.T) {
}, arr) }, arr)
// C->S window acknowledgement size // C->S window acknowledgement size
err = chunk0{ err = base.Chunk0{
chunkStreamID: 2, ChunkStreamID: 2,
typ: 0x05, Typ: 0x05,
bodyLen: 4, BodyLen: 4,
body: []byte{0x00, 0x26, 0x25, 0xa0}, Body: []byte{0x00, 0x26, 0x25, 0xa0},
}.write(conn) }.Write(conn)
require.NoError(t, err) require.NoError(t, err)
// C->S set chunk size // C->S set chunk size
err = chunk0{ err = base.Chunk0{
chunkStreamID: 2, ChunkStreamID: 2,
typ: 1, Typ: 1,
bodyLen: 4, BodyLen: 4,
body: []byte{0x00, 0x01, 0x00, 0x00}, Body: []byte{0x00, 0x01, 0x00, 0x00},
}.write(conn) }.Write(conn)
require.NoError(t, err) require.NoError(t, err)
// C->S createStream // C->S createStream
err = chunk1{ err = base.Chunk1{
chunkStreamID: 3, ChunkStreamID: 3,
typ: 0x14, Typ: 0x14,
body: flvio.FillAMF0ValsMalloc([]interface{}{ Body: flvio.FillAMF0ValsMalloc([]interface{}{
"createStream", "createStream",
float64(2), float64(2),
nil, nil,
}), }),
}.write(conn) }.Write(conn)
require.NoError(t, err) require.NoError(t, err)
// S->C result // S->C result
err = c0.read(conn, 65536) err = c0.Read(conn, 65536)
require.NoError(t, err) require.NoError(t, err)
require.Equal(t, uint8(3), c0.chunkStreamID) require.Equal(t, uint8(3), c0.ChunkStreamID)
require.Equal(t, uint8(0x14), c0.typ) require.Equal(t, uint8(0x14), c0.Typ)
arr, err = flvio.ParseAMFVals(c0.body, false) arr, err = flvio.ParseAMFVals(c0.Body, false)
require.NoError(t, err) require.NoError(t, err)
require.Equal(t, []interface{}{ require.Equal(t, []interface{}{
"_result", "_result",
@ -830,11 +655,11 @@ func TestWriteTracks(t *testing.T) {
nil, nil,
"", "",
}) })
err = chunk0{ err = base.Chunk0{
chunkStreamID: 8, ChunkStreamID: 8,
bodyLen: uint32(len(byts)), BodyLen: uint32(len(byts)),
body: byts, Body: byts,
}.write(conn) }.Write(conn)
require.NoError(t, err) require.NoError(t, err)
// C->S play // C->S play
@ -845,40 +670,40 @@ func TestWriteTracks(t *testing.T) {
"", "",
float64(-2000), float64(-2000),
}) })
err = chunk0{ err = base.Chunk0{
chunkStreamID: 8, ChunkStreamID: 8,
typ: 0x14, Typ: 0x14,
bodyLen: uint32(len(byts)), BodyLen: uint32(len(byts)),
body: byts, Body: byts,
}.write(conn) }.Write(conn)
require.NoError(t, err) require.NoError(t, err)
// S->C event "stream is recorded" // S->C event "stream is recorded"
err = c0.read(conn, 65536) err = c0.Read(conn, 65536)
require.NoError(t, err) require.NoError(t, err)
require.Equal(t, chunk0{ require.Equal(t, base.Chunk0{
chunkStreamID: 2, ChunkStreamID: 2,
typ: 4, Typ: 4,
bodyLen: 6, BodyLen: 6,
body: []byte{0x00, 0x04, 0x00, 0x00, 0x00, 0x01}, Body: []byte{0x00, 0x04, 0x00, 0x00, 0x00, 0x01},
}, c0) }, c0)
// S->C event "stream begin 1" // S->C event "stream begin 1"
err = c0.read(conn, 65536) err = c0.Read(conn, 65536)
require.NoError(t, err) require.NoError(t, err)
require.Equal(t, chunk0{ require.Equal(t, base.Chunk0{
chunkStreamID: 2, ChunkStreamID: 2,
typ: 4, Typ: 4,
bodyLen: 6, BodyLen: 6,
body: []byte{0x00, 0x00, 0x00, 0x00, 0x00, 0x01}, Body: []byte{0x00, 0x00, 0x00, 0x00, 0x00, 0x01},
}, c0) }, c0)
// S->C onStatus // S->C onStatus
err = c0.read(conn, 65536) err = c0.Read(conn, 65536)
require.NoError(t, err) require.NoError(t, err)
require.Equal(t, uint8(5), c0.chunkStreamID) require.Equal(t, uint8(5), c0.ChunkStreamID)
require.Equal(t, uint8(0x14), c0.typ) require.Equal(t, uint8(0x14), c0.Typ)
arr, err = flvio.ParseAMFVals(c0.body, false) arr, err = flvio.ParseAMFVals(c0.Body, false)
require.NoError(t, err) require.NoError(t, err)
require.Equal(t, []interface{}{ require.Equal(t, []interface{}{
"onStatus", "onStatus",
@ -892,11 +717,11 @@ func TestWriteTracks(t *testing.T) {
}, arr) }, arr)
// S->C onStatus // S->C onStatus
err = c0.read(conn, 65536) err = c0.Read(conn, 65536)
require.NoError(t, err) require.NoError(t, err)
require.Equal(t, uint8(5), c0.chunkStreamID) require.Equal(t, uint8(5), c0.ChunkStreamID)
require.Equal(t, uint8(0x14), c0.typ) require.Equal(t, uint8(0x14), c0.Typ)
arr, err = flvio.ParseAMFVals(c0.body, false) arr, err = flvio.ParseAMFVals(c0.Body, false)
require.NoError(t, err) require.NoError(t, err)
require.Equal(t, []interface{}{ require.Equal(t, []interface{}{
"onStatus", "onStatus",
@ -910,11 +735,11 @@ func TestWriteTracks(t *testing.T) {
}, arr) }, arr)
// S->C onStatus // S->C onStatus
err = c0.read(conn, 65536) err = c0.Read(conn, 65536)
require.NoError(t, err) require.NoError(t, err)
require.Equal(t, uint8(5), c0.chunkStreamID) require.Equal(t, uint8(5), c0.ChunkStreamID)
require.Equal(t, uint8(0x14), c0.typ) require.Equal(t, uint8(0x14), c0.Typ)
arr, err = flvio.ParseAMFVals(c0.body, false) arr, err = flvio.ParseAMFVals(c0.Body, false)
require.NoError(t, err) require.NoError(t, err)
require.Equal(t, []interface{}{ require.Equal(t, []interface{}{
"onStatus", "onStatus",
@ -928,11 +753,11 @@ func TestWriteTracks(t *testing.T) {
}, arr) }, arr)
// S->C onStatus // S->C onStatus
err = c0.read(conn, 65536) err = c0.Read(conn, 65536)
require.NoError(t, err) require.NoError(t, err)
require.Equal(t, uint8(5), c0.chunkStreamID) require.Equal(t, uint8(5), c0.ChunkStreamID)
require.Equal(t, uint8(0x14), c0.typ) require.Equal(t, uint8(0x14), c0.Typ)
arr, err = flvio.ParseAMFVals(c0.body, false) arr, err = flvio.ParseAMFVals(c0.Body, false)
require.NoError(t, err) require.NoError(t, err)
require.Equal(t, []interface{}{ require.Equal(t, []interface{}{
"onStatus", "onStatus",
@ -946,11 +771,11 @@ func TestWriteTracks(t *testing.T) {
}, arr) }, arr)
// S->C onMetadata // S->C onMetadata
err = c0.read(conn, 65536) err = c0.Read(conn, 65536)
require.NoError(t, err) require.NoError(t, err)
require.Equal(t, uint8(4), c0.chunkStreamID) require.Equal(t, uint8(4), c0.ChunkStreamID)
require.Equal(t, uint8(0x12), c0.typ) require.Equal(t, uint8(0x12), c0.Typ)
arr, err = flvio.ParseAMFVals(c0.body, false) arr, err = flvio.ParseAMFVals(c0.Body, false)
require.NoError(t, err) require.NoError(t, err)
require.Equal(t, []interface{}{ require.Equal(t, []interface{}{
"onMetaData", "onMetaData",
@ -963,10 +788,10 @@ func TestWriteTracks(t *testing.T) {
}, arr) }, arr)
// S->C H264 decoder config // S->C H264 decoder config
err = c0.read(conn, 65536) err = c0.Read(conn, 65536)
require.NoError(t, err) require.NoError(t, err)
require.Equal(t, uint8(6), c0.chunkStreamID) require.Equal(t, uint8(6), c0.ChunkStreamID)
require.Equal(t, uint8(0x09), c0.typ) require.Equal(t, uint8(0x09), c0.Typ)
require.Equal(t, []byte{ require.Equal(t, []byte{
0x17, 0x0, 0x0, 0x0, 0x0, 0x1, 0x64, 0x0, 0x17, 0x0, 0x0, 0x0, 0x0, 0x1, 0x64, 0x0,
0xc, 0xff, 0xe1, 0x0, 0x15, 0x67, 0x64, 0x0, 0xc, 0xff, 0xe1, 0x0, 0x15, 0x67, 0x64, 0x0,
@ -974,12 +799,12 @@ func TestWriteTracks(t *testing.T) {
0x0, 0x3, 0x0, 0x2, 0x0, 0x0, 0x3, 0x0, 0x0, 0x3, 0x0, 0x2, 0x0, 0x0, 0x3, 0x0,
0x3d, 0x8, 0x1, 0x0, 0x4, 0x68, 0xee, 0x3c, 0x3d, 0x8, 0x1, 0x0, 0x4, 0x68, 0xee, 0x3c,
0x80, 0x80,
}, c0.body) }, c0.Body)
// S->C AAC decoder config // S->C AAC decoder config
err = c0.read(conn, 65536) err = c0.Read(conn, 65536)
require.NoError(t, err) require.NoError(t, err)
require.Equal(t, uint8(4), c0.chunkStreamID) require.Equal(t, uint8(4), c0.ChunkStreamID)
require.Equal(t, uint8(0x08), c0.typ) require.Equal(t, uint8(0x08), c0.Typ)
require.Equal(t, []byte{0xae, 0x0, 0x12, 0x10}, c0.body) require.Equal(t, []byte{0xae, 0x0, 0x12, 0x10}, c0.Body)
} }

Write Preview
Loading…
Cancel
Save