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

rtmp: support Extended Metadata Frames (#4006) (#4014) (#4018)

Browse Source
This commit is contained in:
Alessandro Ros
2024-12-04 23:29:16 +01:00
committed by GitHub
parent 82409e2ced
commit 770d72b638
12 changed files with 305 additions and 237 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/protocols/rtmp/amf0/unmarshal.go → internal/protocols/rtmp/amf0/data.go
+213 -2
View File
@@ -1,3 +1,4 @@
// Package amf0 contains an AMF0 decoder and encoder.
package amf0
import (
@@ -28,9 +29,15 @@ const (
var errBufferTooShort = errors.New("buffer is too short")
// StrictArray is an AMF0 Strict Array.
type StrictArray []interface{}
// Data is a list of ActionScript object graphs.
type Data []interface{}
// Unmarshal decodes AMF0 data.
func Unmarshal(buf []byte) ([]interface{}, error) {
var out []interface{}
func Unmarshal(buf []byte) (Data, error) {
var out Data
for len(buf) != 0 {
var item interface{}
@@ -210,3 +217,207 @@ func unmarshal(buf []byte) (interface{}, []byte, error) {
return nil, nil, fmt.Errorf("unsupported marker 0x%.2x", marker)
}
}
// Marshal encodes AMF0 data.
func (data Data) Marshal() ([]byte, error) {
n, err := data.MarshalSize()
if err != nil {
return nil, err
}
buf := make([]byte, n)
_, err = data.MarshalTo(buf)
if err != nil {
return nil, err
}
return buf, nil
}
// MarshalTo encodes AMF0 data into an existing buffer.
func (data Data) MarshalTo(buf []byte) (int, error) {
n := 0
for _, item := range data {
n += marshalItem(item, buf[n:])
}
return n, nil
}
// MarshalSize returns the size needed to encode data in AMF0.
func (data Data) MarshalSize() (int, error) {
n := 0
for _, item := range data {
in, err := marshalSizeItem(item)
if err != nil {
return 0, err
}
n += in
}
return n, nil
}
func marshalSizeItem(item interface{}) (int, error) {
switch item := item.(type) {
case float64:
return 9, nil
case bool:
return 2, nil
case string:
return 3 + len(item), nil
case ECMAArray:
n := 5
for _, entry := range item {
en, err := marshalSizeItem(entry.Value)
if err != nil {
return 0, err
}
n += 2 + len(entry.Key) + en
}
n += 3
return n, nil
case Object:
n := 1
for _, entry := range item {
en, err := marshalSizeItem(entry.Value)
if err != nil {
return 0, err
}
n += 2 + len(entry.Key) + en
}
n += 3
return n, nil
case StrictArray:
n := 5
for _, entry := range item {
en, err := marshalSizeItem(entry)
if err != nil {
return 0, err
}
n += en
}
return n, nil
case nil:
return 1, nil
default:
return 0, fmt.Errorf("unsupported data type: %T", item)
}
}
func marshalItem(item interface{}, buf []byte) int {
switch item := item.(type) {
case float64:
v := math.Float64bits(item)
buf[0] = markerNumber
buf[1] = byte(v >> 56)
buf[2] = byte(v >> 48)
buf[3] = byte(v >> 40)
buf[4] = byte(v >> 32)
buf[5] = byte(v >> 24)
buf[6] = byte(v >> 16)
buf[7] = byte(v >> 8)
buf[8] = byte(v)
return 9
case bool:
buf[0] = markerBoolean
if item {
buf[1] = 1
}
return 2
case string:
le := len(item)
buf[0] = markerString
buf[1] = byte(le >> 8)
buf[2] = byte(le)
copy(buf[3:], item)
return 3 + le
case ECMAArray:
le := len(item)
buf[0] = markerECMAArray
buf[1] = byte(le >> 24)
buf[2] = byte(le >> 16)
buf[3] = byte(le >> 8)
buf[4] = byte(le)
n := 5
for _, entry := range item {
le := len(entry.Key)
buf[n] = byte(le >> 8)
buf[n+1] = byte(le)
copy(buf[n+2:], entry.Key)
n += 2 + le
n += marshalItem(entry.Value, buf[n:])
}
buf[n] = 0
buf[n+1] = 0
buf[n+2] = markerObjectEnd
return n + 3
case Object:
buf[0] = markerObject
n := 1
for _, entry := range item {
le := len(entry.Key)
buf[n] = byte(le >> 8)
buf[n+1] = byte(le)
copy(buf[n+2:], entry.Key)
n += 2 + le
n += marshalItem(entry.Value, buf[n:])
}
buf[n] = 0
buf[n+1] = 0
buf[n+2] = markerObjectEnd
return n + 3
case StrictArray:
le := len(item)
buf[0] = markerStrictArray
buf[1] = byte(le >> 24)
buf[2] = byte(le >> 16)
buf[3] = byte(le >> 8)
buf[4] = byte(le)
n := 5
for _, entry := range item {
n += marshalItem(entry, buf[n:])
}
return n
default:
buf[0] = markerNull
return 1
}
}
internal/protocols/rtmp/amf0/unmarshal_test.go → internal/protocols/rtmp/amf0/data_test.go
+12 -2
View File
@@ -9,7 +9,7 @@ import (
var cases = []struct {
name string
enc []byte
dec []interface{}
dec Data
}{
{
"on metadata",
@@ -316,6 +316,16 @@ func TestUnmarshal(t *testing.T) {
}
}
func TestMarshal(t *testing.T) {
for _, ca := range cases {
t.Run(ca.name, func(t *testing.T) {
enc, err := ca.dec.Marshal()
require.NoError(t, err)
require.Equal(t, ca.enc, enc)
})
}
}
func FuzzUnmarshal(f *testing.F) {
for _, ca := range cases {
f.Add(ca.enc)
@@ -324,7 +334,7 @@ func FuzzUnmarshal(f *testing.F) {
f.Fuzz(func(_ *testing.T, b []byte) {
what, err := Unmarshal(b)
if err == nil {
Marshal(what) //nolint:errcheck
what.Marshal() //nolint:errcheck
}
})
}
internal/protocols/rtmp/amf0/marshal.go
-200
View File
@@ -1,200 +0,0 @@
// Package amf0 contains an AMF0 marshaler and unmarshaler.
package amf0
import (
"fmt"
"math"
)
// Marshal encodes AMF0 data.
func Marshal(data []interface{}) ([]byte, error) {
n, err := marshalSize(data)
if err != nil {
return nil, err
}
buf := make([]byte, n)
n = 0
for _, item := range data {
n += marshalItem(item, buf[n:])
}
return buf, nil
}
func marshalSize(data []interface{}) (int, error) {
n := 0
for _, item := range data {
in, err := marshalSizeItem(item)
if err != nil {
return 0, err
}
n += in
}
return n, nil
}
func marshalSizeItem(item interface{}) (int, error) {
switch item := item.(type) {
case float64:
return 9, nil
case bool:
return 2, nil
case string:
return 3 + len(item), nil
case ECMAArray:
n := 5
for _, entry := range item {
en, err := marshalSizeItem(entry.Value)
if err != nil {
return 0, err
}
n += 2 + len(entry.Key) + en
}
n += 3
return n, nil
case Object:
n := 1
for _, entry := range item {
en, err := marshalSizeItem(entry.Value)
if err != nil {
return 0, err
}
n += 2 + len(entry.Key) + en
}
n += 3
return n, nil
case StrictArray:
n := 5
for _, entry := range item {
en, err := marshalSizeItem(entry)
if err != nil {
return 0, err
}
n += en
}
return n, nil
case nil:
return 1, nil
default:
return 0, fmt.Errorf("unsupported data type: %T", item)
}
}
func marshalItem(item interface{}, buf []byte) int {
switch item := item.(type) {
case float64:
v := math.Float64bits(item)
buf[0] = markerNumber
buf[1] = byte(v >> 56)
buf[2] = byte(v >> 48)
buf[3] = byte(v >> 40)
buf[4] = byte(v >> 32)
buf[5] = byte(v >> 24)
buf[6] = byte(v >> 16)
buf[7] = byte(v >> 8)
buf[8] = byte(v)
return 9
case bool:
buf[0] = markerBoolean
if item {
buf[1] = 1
}
return 2
case string:
le := len(item)
buf[0] = markerString
buf[1] = byte(le >> 8)
buf[2] = byte(le)
copy(buf[3:], item)
return 3 + le
case ECMAArray:
le := len(item)
buf[0] = markerECMAArray
buf[1] = byte(le >> 24)
buf[2] = byte(le >> 16)
buf[3] = byte(le >> 8)
buf[4] = byte(le)
n := 5
for _, entry := range item {
le := len(entry.Key)
buf[n] = byte(le >> 8)
buf[n+1] = byte(le)
copy(buf[n+2:], entry.Key)
n += 2 + le
n += marshalItem(entry.Value, buf[n:])
}
buf[n] = 0
buf[n+1] = 0
buf[n+2] = markerObjectEnd
return n + 3
case Object:
buf[0] = markerObject
n := 1
for _, entry := range item {
le := len(entry.Key)
buf[n] = byte(le >> 8)
buf[n+1] = byte(le)
copy(buf[n+2:], entry.Key)
n += 2 + le
n += marshalItem(entry.Value, buf[n:])
}
buf[n] = 0
buf[n+1] = 0
buf[n+2] = markerObjectEnd
return n + 3
case StrictArray:
le := len(item)
buf[0] = markerStrictArray
buf[1] = byte(le >> 24)
buf[2] = byte(le >> 16)
buf[3] = byte(le >> 8)
buf[4] = byte(le)
n := 5
for _, entry := range item {
n += marshalItem(entry, buf[n:])
}
return n
default:
buf[0] = markerNull
return 1
}
}
internal/protocols/rtmp/amf0/marshal_test.go
-17
View File
@@ -1,17 +0,0 @@
package amf0
import (
"testing"
"github.com/stretchr/testify/require"
)
func TestMarshal(t *testing.T) {
for _, ca := range cases {
t.Run(ca.name, func(t *testing.T) {
enc, err := Marshal(ca.dec)
require.NoError(t, err)
require.Equal(t, ca.enc, enc)
})
}
}
internal/protocols/rtmp/amf0/object.go
-3
View File
@@ -12,9 +12,6 @@ type Object []ObjectEntry
// ECMAArray is an AMF0 ECMA Array.
type ECMAArray Object
// StrictArray is an AMF0 Strict Array.
type StrictArray []interface{}
// Get returns the value corresponding to key.
func (o Object) Get(key string) (interface{}, bool) {
for _, item := range o {
internal/protocols/rtmp/message/command_amf0.go
+3 -3
View File
@@ -13,7 +13,7 @@ type CommandAMF0 struct {
MessageStreamID uint32
Name string
CommandID int
Arguments []interface{}
Arguments amf0.Data
}
func (m *CommandAMF0) unmarshal(raw *rawmessage.Message) error {
@@ -47,12 +47,12 @@ func (m *CommandAMF0) unmarshal(raw *rawmessage.Message) error {
}
func (m CommandAMF0) marshal() (*rawmessage.Message, error) {
data := append([]interface{}{
data := append(amf0.Data{
m.Name,
float64(m.CommandID),
}, m.Arguments...)
body, err := amf0.Marshal(data)
body, err := data.Marshal()
if err != nil {
return nil, err
}
internal/protocols/rtmp/message/data_amf0.go
+4 -4
View File
@@ -9,24 +9,24 @@ import (
type DataAMF0 struct {
ChunkStreamID byte
MessageStreamID uint32
Payload []interface{}
Payload amf0.Data
}
func (m *DataAMF0) unmarshal(raw *rawmessage.Message) error {
m.ChunkStreamID = raw.ChunkStreamID
m.MessageStreamID = raw.MessageStreamID
payload, err := amf0.Unmarshal(raw.Body)
var err error
m.Payload, err = amf0.Unmarshal(raw.Body)
if err != nil {
return err
}
m.Payload = payload
return nil
}
func (m DataAMF0) marshal() (*rawmessage.Message, error) {
body, err := amf0.Marshal(m.Payload)
body, err := m.Payload.Marshal()
if err != nil {
return nil, err
}
internal/protocols/rtmp/message/extended_metadata.go
+50 -4
View File
@@ -2,25 +2,71 @@ package message
import (
"fmt"
"time"
"github.com/bluenviron/mediamtx/internal/protocols/rtmp/amf0"
"github.com/bluenviron/mediamtx/internal/protocols/rtmp/rawmessage"
)
// ExtendedMetadata is a metadata extended message.
type ExtendedMetadata struct {
FourCC FourCC
ChunkStreamID byte
DTS time.Duration
MessageStreamID uint32
FourCC FourCC
Payload amf0.Data
}
func (m *ExtendedMetadata) unmarshal(raw *rawmessage.Message) error {
if len(raw.Body) != 5 {
if len(raw.Body) < 6 {
return fmt.Errorf("invalid body size")
}
m.ChunkStreamID = raw.ChunkStreamID
m.DTS = raw.Timestamp
m.MessageStreamID = raw.MessageStreamID
m.FourCC = FourCC(raw.Body[1])<<24 | FourCC(raw.Body[2])<<16 | FourCC(raw.Body[3])<<8 | FourCC(raw.Body[4])
return fmt.Errorf("ExtendedMetadata is not implemented yet")
var err error
m.Payload, err = amf0.Unmarshal(raw.Body[5:])
if err != nil {
return err
}
return nil
}
func (m ExtendedMetadata) marshalBodySize() (int, error) {
ms, err := m.Payload.MarshalSize()
if err != nil {
return 0, err
}
return 5 + ms, nil
}
func (m ExtendedMetadata) marshal() (*rawmessage.Message, error) {
return nil, fmt.Errorf("TODO")
mbs, err := m.marshalBodySize()
if err != nil {
return nil, err
}
body := make([]byte, mbs)
body[0] = 0b10000000 | byte(ExtendedTypeMetadata)
body[1] = uint8(m.FourCC >> 24)
body[2] = uint8(m.FourCC >> 16)
body[3] = uint8(m.FourCC >> 8)
body[4] = uint8(m.FourCC)
_, err = m.Payload.MarshalTo(body[5:])
if err != nil {
return nil, err
}
return &rawmessage.Message{
ChunkStreamID: m.ChunkStreamID,
Timestamp: m.DTS,
Type: uint8(TypeVideo),
MessageStreamID: m.MessageStreamID,
Body: body,
}, nil
}
internal/protocols/rtmp/message/extended_mpeg2ts_sequence_start.go
+1 -1
View File
@@ -12,7 +12,7 @@ type ExtendedMPEG2TSSequenceStart struct {
}
func (m *ExtendedMPEG2TSSequenceStart) unmarshal(raw *rawmessage.Message) error {
if len(raw.Body) != 5 {
if len(raw.Body) < 5 {
return fmt.Errorf("invalid body size")
}
internal/protocols/rtmp/message/extended_sequence_end.go
+1 -1
View File
@@ -12,7 +12,7 @@ type ExtendedSequenceEnd struct {
}
func (m *ExtendedSequenceEnd) unmarshal(raw *rawmessage.Message) error {
if len(raw.Body) != 5 {
if len(raw.Body) < 5 {
return fmt.Errorf("invalid body size")
}
internal/protocols/rtmp/message/reader_test.go
+19
View File
@@ -277,6 +277,25 @@ var readWriterCases = []struct {
0x31, 0x01, 0x02, 0x03,
},
},
{
"extended metadata",
&ExtendedMetadata{
ChunkStreamID: 0x6,
DTS: 0,
MessageStreamID: 0x1000000,
FourCC: 0x68766331,
Payload: []interface{}{"colorInfo", amf0.Object{amf0.ObjectEntry{Key: "colorConfig", Value: amf0.Object{}}}},
},
[]byte{
0x06, 0x00, 0x00, 0x00, 0x00, 0x00, 0x26, 0x09,
0x01, 0x00, 0x00, 0x00, 0x84, 0x68, 0x76, 0x63,
0x31, 0x02, 0x00, 0x09, 0x63, 0x6f, 0x6c, 0x6f,
0x72, 0x49, 0x6e, 0x66, 0x6f, 0x03, 0x00, 0x0b,
0x63, 0x6f, 0x6c, 0x6f, 0x72, 0x43, 0x6f, 0x6e,
0x66, 0x69, 0x67, 0x03, 0x00, 0x00, 0x09, 0x00,
0x00, 0x09,
},
},
}
func TestReader(t *testing.T) {
internal/protocols/rtmp/message/testdata/fuzz/FuzzReader/105635bd94dfe048
Vendored
+2
View File
@@ -0,0 +1,2 @@
go test fuzz v1
[]byte("0000\x00\x00\b\t0000\x94hvc1000")