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

moq: support publishing and reading through native QUIC (#6039)

Browse Source
This commit is contained in:
Alessandro Ros
2026-08-04 23:47:59 +02:00
committed by GitHub
parent dcd4b38838
commit e0ed9885fc
37 changed files with 734 additions and 137 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/openapi.yaml
+4
View File
@@ -447,6 +447,8 @@ components:
type: string
nullable: true
deprecated: true
moqQUICAddress:
type: string
moqServerCert:
type: string
moqServerKey:
@@ -825,6 +827,8 @@ components:
type: string
state:
$ref: "#/components/schemas/MoQSessionState"
transport:
type: string
userAgent:
type: string
version:
docs/3-publish/01-moq-clients.md
+2 -2
View File
@@ -10,7 +10,7 @@ Media-over-QUIC is a streaming protocol built upon cutting edge protocols (QUIC,
Media-over-QUIC has a wide range of features and variants, most of them in active development. We currently support the following:
- The server supports `draft-19` and `draft-18` of the [main specification](https://datatracker.ietf.org/doc/html/draft-ietf-moq-transport-19), and prefers `draft-19` when both are offered during negotiation.
- We only support using Media-over-QUIC through browsers and in particular through the WebTransport API. We do not support using QUIC directly.
- We support using Media-over-QUIC through browsers with the WebTransport API and through native QUIC clients.
- We support the `PUBLISH` and `SUBSCRIBE` messages only, which are the ones meant to be used with a routing solution like _MediaMTX_.
- We use the MOQT Streaming Format (MSF) to advertise tracks, described in [this specification](https://datatracker.ietf.org/doc/html/draft-ietf-moq-msf-00).
- We use the Low Overhead Media Container (LOC) to ship frames, described in [this specification](https://datatracker.ietf.org/doc/draft-ietf-moq-loc/).
@@ -34,4 +34,4 @@ You can publish a stream with Media-over-QUIC and a web browser by visiting:
https://localhost:8892/mystream/publish
```
The only clients that can currently publish with Media-over-QUIC are [Web browsers](16-web-browsers.md).
You can also publish with native QUIC clients by connecting directly to `moqQUICAddress` (default is `:8893`).
docs/4-read/01-moq.md
+1 -1
View File
@@ -15,4 +15,4 @@ You can read a stream with Media-over-QUIC and a web browser by visiting:
https://localhost:8892/mystream
```
The only clients that can currently read with Media-over-QUIC are [Web browsers](07-web-browsers.md).
You can also read with native QUIC clients by connecting directly to `moqQUICAddress` (default is `:8893`).
internal/conf/conf.go
+6
View File
@@ -400,6 +400,7 @@ type Conf struct {
MoQ bool `json:"moq"`
MoQHTTP2Address string `json:"moqHTTP2Address"`
MoQHTTP3Address string `json:"moqHTTP3Address"`
MoQQUICAddress string `json:"moqQUICAddress"`
MoQServerKey string `json:"moqServerKey"`
MoQServerCert string `json:"moqServerCert"`
MoQAllowOrigins []string `json:"moqAllowOrigins"`
@@ -539,6 +540,7 @@ func (conf *Conf) setDefaults() {
conf.MoQ = true
conf.MoQHTTP2Address = ":8892"
conf.MoQHTTP3Address = ":8892"
conf.MoQQUICAddress = ":8893"
conf.MoQServerKey = "auto.key"
conf.MoQServerCert = "auto.crt"
conf.MoQAllowOrigins = []string{"*"}
@@ -1056,6 +1058,10 @@ func (conf *Conf) Validate(l logger.Writer) error {
conf.MoQHTTP3Address = *conf.MoQHTTPS3Address
}
if conf.MoQ && conf.MoQQUICAddress == "" {
return fmt.Errorf("'moqQUICAddress' must be set when MoQ is enabled")
}
// Record (deprecated)
if conf.Record != nil {
internal/core/core.go
+2
View File
@@ -706,6 +706,7 @@ func (p *Core) createResources(initial bool) error {
i := &moq.Server{
HTTP2Address: p.conf.MoQHTTP2Address,
HTTP3Address: p.conf.MoQHTTP3Address,
QUICAddress: p.conf.MoQQUICAddress,
ServerKey: p.conf.MoQServerKey,
ServerCert: p.conf.MoQServerCert,
AllowOrigins: p.conf.MoQAllowOrigins,
@@ -1013,6 +1014,7 @@ func (p *Core) closeResources(newConf *conf.Conf, calledByAPI bool) {
newConf.MoQ != p.conf.MoQ ||
newConf.MoQHTTP2Address != p.conf.MoQHTTP2Address ||
newConf.MoQHTTP3Address != p.conf.MoQHTTP3Address ||
newConf.MoQQUICAddress != p.conf.MoQQUICAddress ||
newConf.MoQServerKey != p.conf.MoQServerKey ||
newConf.MoQServerCert != p.conf.MoQServerCert ||
!slices.Equal(newConf.MoQAllowOrigins, p.conf.MoQAllowOrigins) ||
internal/defs/api_moq.go
+10
View File
@@ -32,6 +32,15 @@ const (
APIMoQVersionDraft19 APIMoQVersion = "moqt-19"
)
// APIMoQSessionTransport is the underlying transport of a MoQ session.
type APIMoQSessionTransport string
// transports.
const (
APIMoQSessionTransportWebTransport APIMoQSessionTransport = "webtransport"
APIMoQSessionTransportQUIC APIMoQSessionTransport = "quic"
)
// APIMoQSessionList is a list of MoQ sessions.
type APIMoQSessionList struct {
ItemCount int `json:"itemCount"`
@@ -48,6 +57,7 @@ type APIMoQSession struct {
Path string `json:"path"`
Query string `json:"query"`
UserAgent string `json:"userAgent"`
Transport APIMoQSessionTransport `json:"transport"`
Version APIMoQVersion `json:"version"`
InboundBytes uint64 `json:"inboundBytes"`
OutboundBytes uint64 `json:"outboundBytes"`
internal/protocols/moq/controlmessage/message_test.go
+43 -12
View File
@@ -1,9 +1,10 @@
package controlmessage
package controlmessage_test
import (
"bytes"
"testing"
"github.com/bluenviron/mediamtx/internal/protocols/moq/controlmessage"
"github.com/bluenviron/mediamtx/internal/protocols/moq/parameter"
"github.com/stretchr/testify/require"
)
@@ -11,7 +12,7 @@ import (
var cases = []struct {
name string
enc []byte
dec Message
dec controlmessage.Message
}{
{
name: "setup",
@@ -19,7 +20,37 @@ var cases = []struct {
0xAF, 0x00, // type 0x2F00 (2-byte varint)
0x00, 0x00, // length = 0
},
dec: &Setup{},
dec: &controlmessage.Setup{},
},
{
name: "setup with path",
enc: []byte{
0xAF, 0x00, // type 0x2F00 (2-byte varint)
0x00, 0x06, // length = 6
0x01, // delta type = PATH (0x01)
0x04, // option len
0x2F, 0x66, 0x6F, 0x6F, // "/foo"
},
dec: &controlmessage.Setup{
Path: "/foo",
},
},
{
name: "setup with path and authority",
enc: []byte{
0xAF, 0x00, // type 0x2F00 (2-byte varint)
0x00, 0x11, // length = 17
0x01, // delta type = PATH (0x01 - 0 = 0x01)
0x04, // option len
0x2F, 0x66, 0x6F, 0x6F, // "/foo"
0x04, // delta type = AUTHORITY (0x05 - 0x01 = 0x04)
0x09, // option len
0x6C, 0x6F, 0x63, 0x61, 0x6C, 0x68, 0x6F, 0x73, 0x74, // "localhost"
},
dec: &controlmessage.Setup{
Path: "/foo",
Authority: "localhost",
},
},
{
name: "subscribe",
@@ -32,7 +63,7 @@ var cases = []struct {
0x03, 0x62, 0x61, 0x72, // track name = "bar"
0x00, // parameters count = 0
},
dec: &Subscribe{
dec: &controlmessage.Subscribe{
RequestID: 1,
Namespace: []string{"foo"},
TrackName: "bar",
@@ -54,7 +85,7 @@ var cases = []struct {
0x01, // token type = 1
0x73, 0x65, 0x63, 0x72, 0x65, 0x74, // token value = "secret"
},
dec: &Subscribe{
dec: &controlmessage.Subscribe{
RequestID: 1,
Namespace: []string{"foo"},
TrackName: "bar",
@@ -75,7 +106,7 @@ var cases = []struct {
0x01, // TrackAlias = 1
0x00, // Number of Parameters = 0
},
dec: &SubscribeOk{
dec: &controlmessage.SubscribeOk{
TrackAlias: 1,
},
},
@@ -88,7 +119,7 @@ var cases = []struct {
0x00, // retryInterval = 0 (ignored)
0x03, 0x66, 0x6F, 0x6F, // Reason = "foo"
},
dec: &RequestError{
dec: &controlmessage.RequestError{
Code: 1,
Reason: "foo",
},
@@ -100,7 +131,7 @@ var cases = []struct {
0x00, 0x01, // length = 1
0x00, // Number of Parameters = 0
},
dec: &RequestOk{},
dec: &controlmessage.RequestOk{},
},
{
name: "publish",
@@ -114,7 +145,7 @@ var cases = []struct {
0x02, // TrackAlias = 2
0x00, // parameters count = 0
},
dec: &Publish{
dec: &controlmessage.Publish{
RequestID: 1,
Namespace: []string{"foo"},
TrackName: "bar",
@@ -138,7 +169,7 @@ var cases = []struct {
0x01, // token type = 1
0x73, 0x65, 0x63, 0x72, 0x65, 0x74, // token value = "secret"
},
dec: &Publish{
dec: &controlmessage.Publish{
RequestID: 1,
Namespace: []string{"foo"},
TrackName: "bar",
@@ -157,7 +188,7 @@ var cases = []struct {
func TestUnmarshal(t *testing.T) {
for _, ca := range cases {
t.Run(ca.name, func(t *testing.T) {
m, err := Read(bytes.NewReader(ca.enc))
m, err := controlmessage.Read(bytes.NewReader(ca.enc))
require.NoError(t, err)
require.Equal(t, ca.dec, m)
})
@@ -179,7 +210,7 @@ func FuzzUnmarshal(f *testing.F) {
}
f.Fuzz(func(_ *testing.T, buf []byte) {
m, err := Read(bytes.NewReader(buf))
m, err := controlmessage.Read(bytes.NewReader(buf))
if err != nil {
return
}
internal/protocols/moq/controlmessage/publish.go
+1 -1
View File
@@ -15,7 +15,7 @@ const (
)
// Publish is the PUBLISH control message.
// spec: draft-18, section 10.10
// spec: draft-18/19, section 10.10
type Publish struct {
RequestID uint64
Namespace []string
internal/protocols/moq/controlmessage/request_error.go
+2 -2
View File
@@ -11,7 +11,7 @@ const typeRequestError varint.Varint = 0x05
// RequestErrorCode is a code of REQUEST_ERROR.
type RequestErrorCode uint64
// spec: draft-18, section 15.10.2
// spec: draft-18, section 15.10.2 / draft-19, section 15.11.2
const (
RequestErrorCodeUnauthorized RequestErrorCode = 0x01
RequestErrorCodeNotSupported RequestErrorCode = 0x03
@@ -20,7 +20,7 @@ const (
)
// RequestError is the REQUEST_ERROR control message.
// spec: draft-18, section 10.6.2
// spec: draft-18/19, section 10.6.2
type RequestError struct {
Code RequestErrorCode
Reason string
internal/protocols/moq/controlmessage/request_ok.go
+1 -1
View File
@@ -9,7 +9,7 @@ import (
const typeRequestOk varint.Varint = 0x07
// RequestOk is the REQUEST_OK control message.
// spec: draft-18, section 10.5
// spec: draft-18/19, section 10.5
type RequestOk struct {
Parameters parameter.Parameters
TrackProperties property.Properties
internal/protocols/moq/controlmessage/setup.go
+112 -13
View File
@@ -1,31 +1,130 @@
package controlmessage
import "github.com/bluenviron/mediamtx/internal/protocols/moq/varint"
import (
"fmt"
"github.com/bluenviron/mediamtx/internal/protocols/moq/varint"
)
const typeSetup varint.Varint = 0x2F00
const (
setupOptionPath varint.Varint = 0x01
setupOptionAuthority varint.Varint = 0x05
)
// Setup is the SETUP control message.
// spec: draft-18, section 10.3
type Setup struct{}
// spec: draft-18/19, section 10.3
type Setup struct {
Path string
Authority string
}
func (*Setup) isMessage() {}
func (*Setup) unmarshal(_ []byte) error { return nil }
func (m *Setup) unmarshal(buf []byte) error {
var previousType uint64
for len(buf) > 0 {
var deltaType varint.Varint
n, err := deltaType.Unmarshal(buf)
if err != nil {
return err
}
buf = buf[n:]
func (Setup) marshalSize() int {
return typeSetup.MarshalSize() + 2
currentType := previousType + uint64(deltaType)
previousType = currentType
if currentType%2 == 0 { // even type: single varint value
var v varint.Varint
n, err = v.Unmarshal(buf)
if err != nil {
return err
}
buf = buf[n:]
continue
}
// odd type: length-prefixed byte field
var l varint.Varint
n, err = l.Unmarshal(buf)
if err != nil {
return err
}
buf = buf[n:]
if uint64(len(buf)) < uint64(l) {
return fmt.Errorf("not enough bytes for setup option")
}
value := string(buf[:int(l)])
buf = buf[int(l):]
switch varint.Varint(currentType) {
case setupOptionPath:
m.Path = value
case setupOptionAuthority:
m.Authority = value
}
}
return nil
}
func (Setup) marshalTo(buf []byte) int {
func (m Setup) marshalSize() int {
payloadSize := 0
var previousType varint.Varint
if m.Path != "" {
delta := setupOptionPath - previousType
payloadSize += delta.MarshalSize() +
varint.Varint(len(m.Path)).MarshalSize() +
len(m.Path)
previousType = setupOptionPath
}
if m.Authority != "" {
delta := setupOptionAuthority - previousType
payloadSize += delta.MarshalSize() +
varint.Varint(len(m.Authority)).MarshalSize() +
len(m.Authority)
}
return typeSetup.MarshalSize() + 2 + payloadSize
}
func (m Setup) marshalTo(buf []byte) int {
payloadSize := m.marshalSize() - typeSetup.MarshalSize() - 2
pos := typeSetup.MarshalTo(buf)
buf[pos] = 0x00
buf[pos+1] = 0x00
return pos + 2
buf[pos] = byte(payloadSize >> 8)
buf[pos+1] = byte(payloadSize)
pos += 2
var previousType varint.Varint
if m.Path != "" {
delta := setupOptionPath - previousType
pos += delta.MarshalTo(buf[pos:])
pos += varint.Varint(len(m.Path)).MarshalTo(buf[pos:])
pos += copy(buf[pos:], m.Path)
previousType = setupOptionPath
}
if m.Authority != "" {
delta := setupOptionAuthority - previousType
pos += delta.MarshalTo(buf[pos:])
pos += varint.Varint(len(m.Authority)).MarshalTo(buf[pos:])
pos += copy(buf[pos:], m.Authority)
}
return pos
}
// Marshal implements Message.
func (Setup) Marshal() []byte {
buf := make([]byte, Setup{}.marshalSize())
Setup{}.marshalTo(buf)
func (m Setup) Marshal() []byte {
buf := make([]byte, m.marshalSize())
m.marshalTo(buf)
return buf
}
internal/protocols/moq/controlmessage/subscribe.go
+1 -1
View File
@@ -10,7 +10,7 @@ import (
const typeSubscribe varint.Varint = 0x03
// Subscribe is the SUBSCRIBE control message.
// spec: draft-18, section 10.7
// spec: draft-18/19, section 10.7
type Subscribe struct {
RequestID uint64
Namespace []string
internal/protocols/moq/controlmessage/subscribe_ok.go
+1 -1
View File
@@ -9,7 +9,7 @@ import (
const typeSubscribeOk varint.Varint = 0x04
// SubscribeOk is the SUBSCRIBE_OK control message.
// spec: draft-18, section 10.8
// spec: draft-18/19, section 10.8
type SubscribeOk struct {
TrackAlias uint64
Parameters parameter.Parameters
internal/protocols/moq/controlmessage/testdata/fuzz/FuzzUnmarshal/1cf7bedaacbce494
Vendored
+2
View File
@@ -0,0 +1,2 @@
go test fuzz v1
[]byte("\xaf\x00\x00\x0200")
internal/protocols/moq/controlmessage/testdata/fuzz/FuzzUnmarshal/3105c332986970bb
Vendored
+2
View File
@@ -0,0 +1,2 @@
go test fuzz v1
[]byte("\xaf\x00\x00\b\xa10\xa100\x0100")
internal/protocols/moq/controlmessage/testdata/fuzz/FuzzUnmarshal/d2fdaaf5c2595f2c
Vendored
+2
View File
@@ -0,0 +1,2 @@
go test fuzz v1
[]byte("\xaf\x00\x00\b\xff0000000")
internal/protocols/moq/controlmessage/testdata/fuzz/FuzzUnmarshal/efb79f37f6577948
Vendored
+2
View File
@@ -0,0 +1,2 @@
go test fuzz v1
[]byte("\xaf\x00\x00\x010")
internal/protocols/moq/parameter/authorization_token.go
+3 -3
View File
@@ -9,16 +9,16 @@ import (
const typeAuthorizationToken = 0x03
// AuthorizationTokenAliasType is a value of Alias Type.
// spec: draft-18, section 10.2.2
// spec: draft-18/19, section 10.2.2
type AuthorizationTokenAliasType uint64
// spec: draft-18, section 10.2.2
// spec: draft-18/19, section 10.2.2
const (
AuthorizationTokenAliasTypeUseValue AuthorizationTokenAliasType = 0x03
)
// AuthorizationToken is the AUTHORIZATION_TOKEN parameter.
// spec: draft-18, section 10.2.2
// spec: draft-18/19, section 10.2.2
type AuthorizationToken struct {
AliasType AuthorizationTokenAliasType
TokenType uint64
internal/protocols/moq/parameter/parameter.go
+1 -1
View File
@@ -8,7 +8,7 @@ import (
)
// Parameter is a parameter of a control message.
// spec: draft-18, section 10.2
// spec: draft-18/19, section 10.2
type Parameter interface {
isParameter()
paramType() uint64
internal/protocols/moq/parameter/parameter_test.go
+8 -7
View File
@@ -1,8 +1,9 @@
package parameter
package parameter_test
import (
"testing"
"github.com/bluenviron/mediamtx/internal/protocols/moq/parameter"
"github.com/stretchr/testify/require"
)
@@ -10,7 +11,7 @@ var cases = []struct {
name string
count int
enc []byte
dec Parameters
dec parameter.Parameters
}{
{
name: "no parameters",
@@ -28,9 +29,9 @@ var cases = []struct {
0x01, // token type = 1
0x73, 0x65, 0x63, 0x72, 0x65, 0x74, // token value = "secret"
},
dec: Parameters{
&AuthorizationToken{
AliasType: AuthorizationTokenAliasTypeUseValue,
dec: parameter.Parameters{
&parameter.AuthorizationToken{
AliasType: parameter.AuthorizationTokenAliasTypeUseValue,
TokenType: 1,
TokenValue: []byte("secret"),
},
@@ -41,7 +42,7 @@ var cases = []struct {
func TestUnmarshal(t *testing.T) {
for _, ca := range cases {
t.Run(ca.name, func(t *testing.T) {
var params Parameters
var params parameter.Parameters
_, err := params.Unmarshal(ca.count, ca.enc)
require.NoError(t, err)
require.Equal(t, ca.dec, params)
@@ -65,7 +66,7 @@ func FuzzUnmarshal(f *testing.F) {
}
f.Fuzz(func(_ *testing.T, count int, buf []byte) {
var params Parameters
var params parameter.Parameters
_, err := params.Unmarshal(count, buf)
if err != nil {
return
internal/protocols/moq/property/property.go
+2 -2
View File
@@ -8,7 +8,7 @@ import (
)
// Property is an object property.
// spec: draft-18, section 11.2.1.2
// spec: draft-18/19, section 11.2.1.2
type Property interface {
isProperty()
propType() varint.Varint
@@ -18,7 +18,7 @@ type Property interface {
}
// Properties are object properties.
// spec: draft-18, section 11.2.1.2
// spec: draft-18/19, section 11.2.1.2
type Properties []Property
// Unmarshal decodes properties.
internal/protocols/moq/property/property_test.go
+7 -6
View File
@@ -1,15 +1,16 @@
package property
package property_test
import (
"testing"
"github.com/bluenviron/mediamtx/internal/protocols/moq/property"
"github.com/stretchr/testify/require"
)
var cases = []struct {
name string
enc []byte
dec Properties
dec property.Properties
}{
{
name: "no properties",
@@ -22,8 +23,8 @@ var cases = []struct {
0x06, // type delta = 6 (Timestamp)
0x83, 0xe8, // value = 1000
},
dec: Properties{
new(Timestamp(1000)),
dec: property.Properties{
new(property.Timestamp(1000)),
},
},
}
@@ -31,7 +32,7 @@ var cases = []struct {
func TestUnmarshal(t *testing.T) {
for _, ca := range cases {
t.Run(ca.name, func(t *testing.T) {
var props Properties
var props property.Properties
err := props.Unmarshal(ca.enc)
require.NoError(t, err)
require.Equal(t, ca.dec, props)
@@ -55,7 +56,7 @@ func FuzzUnmarshal(f *testing.F) {
}
f.Fuzz(func(_ *testing.T, buf []byte) {
var props Properties
var props property.Properties
err := props.Unmarshal(buf)
if err != nil {
return
internal/protocols/moq/reorderer/reorderer_test.go
+9 -8
View File
@@ -1,9 +1,10 @@
package reorderer
package reorderer_test
import (
"testing"
"github.com/bluenviron/mediamtx/internal/logger"
"github.com/bluenviron/mediamtx/internal/protocols/moq/reorderer"
"github.com/bluenviron/mediamtx/internal/protocols/moq/subgroup"
"github.com/stretchr/testify/require"
)
@@ -19,7 +20,7 @@ func makeSG(groupID uint64) *subgroup.SubGroup {
}
func TestReordererFirstPush(t *testing.T) {
r := &Reorderer{MaxReordered: 5, Parent: nopLogger{}}
r := &reorderer.Reorderer{MaxReordered: 5, Parent: nopLogger{}}
r.Initialize()
sg0 := makeSG(0)
@@ -29,7 +30,7 @@ func TestReordererFirstPush(t *testing.T) {
}
func TestReordererInOrder(t *testing.T) {
r := &Reorderer{MaxReordered: 5, Parent: nopLogger{}}
r := &reorderer.Reorderer{MaxReordered: 5, Parent: nopLogger{}}
r.Initialize()
sg0 := makeSG(0)
@@ -49,7 +50,7 @@ func TestReordererInOrder(t *testing.T) {
}
func TestReordererStale(t *testing.T) {
r := &Reorderer{MaxReordered: 5, Parent: nopLogger{}}
r := &reorderer.Reorderer{MaxReordered: 5, Parent: nopLogger{}}
r.Initialize()
sg5 := makeSG(5)
@@ -69,7 +70,7 @@ func TestReordererStale(t *testing.T) {
}
func TestReordererOutOfOrderPending(t *testing.T) {
r := &Reorderer{MaxReordered: 5, Parent: nopLogger{}}
r := &reorderer.Reorderer{MaxReordered: 5, Parent: nopLogger{}}
r.Initialize()
sg0 := makeSG(0)
@@ -84,7 +85,7 @@ func TestReordererOutOfOrderPending(t *testing.T) {
}
func TestReordererOutOfOrderFilled(t *testing.T) {
r := &Reorderer{MaxReordered: 5, Parent: nopLogger{}}
r := &reorderer.Reorderer{MaxReordered: 5, Parent: nopLogger{}}
r.Initialize()
sg0 := makeSG(0)
@@ -107,7 +108,7 @@ func TestReordererOutOfOrderFilled(t *testing.T) {
}
func TestReordererDrainAfterFill(t *testing.T) { //nolint:dupl
r := &Reorderer{MaxReordered: 5, Parent: nopLogger{}}
r := &reorderer.Reorderer{MaxReordered: 5, Parent: nopLogger{}}
r.Initialize()
sg0 := makeSG(0)
@@ -134,7 +135,7 @@ func TestReordererDrainAfterFill(t *testing.T) { //nolint:dupl
}
func TestReordererMaxReordered(t *testing.T) { //nolint:dupl
r := &Reorderer{MaxReordered: 2, Parent: nopLogger{}}
r := &reorderer.Reorderer{MaxReordered: 2, Parent: nopLogger{}}
r.Initialize()
sg0 := makeSG(0)
internal/protocols/moq/subgroup/header.go
+1 -1
View File
@@ -8,7 +8,7 @@ import (
)
// Header is the SUBGROUP_HEADER structure.
// spec: draft-18, section 11.4.2
// spec: draft-18/19, section 11.4.2
type Header struct {
Properties bool
FirstObject bool
internal/protocols/moq/subgroup/object.go
+1 -1
View File
@@ -14,7 +14,7 @@ const (
)
// Object is an object of a subgroup stream.
// spec: draft-18, section 11.4.2
// spec: draft-18/19, section 11.4.2
type Object struct {
IDDelta uint64
Properties property.Properties
internal/protocols/moq/subgroup/subgroup_test.go
+11 -10
View File
@@ -1,17 +1,18 @@
package subgroup
package subgroup_test
import (
"bytes"
"testing"
"github.com/bluenviron/mediamtx/internal/protocols/moq/property"
"github.com/bluenviron/mediamtx/internal/protocols/moq/subgroup"
"github.com/stretchr/testify/require"
)
var cases = []struct {
name string
enc []byte
dec SubGroup
dec subgroup.SubGroup
}{
{
name: "stream without properties",
@@ -26,14 +27,14 @@ var cases = []struct {
0x00, // payload length = 0 (end-of-stream)
0x03, // status = EndOfGroup
},
dec: SubGroup{
Header: Header{
dec: subgroup.SubGroup{
Header: subgroup.Header{
Properties: false,
FirstObject: false,
TrackAlias: 1,
GroupID: 0,
},
Objects: []Object{{
Objects: []subgroup.Object{{
Payload: []byte("hello"),
}},
},
@@ -55,14 +56,14 @@ var cases = []struct {
0x00, // payload length = 0 (end-of-stream)
0x03, // status = EndOfGroup
},
dec: SubGroup{
Header: Header{
dec: subgroup.SubGroup{
Header: subgroup.Header{
Properties: true,
FirstObject: false,
TrackAlias: 1,
GroupID: 0,
},
Objects: []Object{{
Objects: []subgroup.Object{{
Properties: property.Properties{
new(property.Timestamp(1000)),
},
@@ -75,7 +76,7 @@ var cases = []struct {
func TestUnmarshal(t *testing.T) {
for _, ca := range cases {
t.Run(ca.name, func(t *testing.T) {
var s SubGroup
var s subgroup.SubGroup
err := s.Read(bytes.NewReader(ca.enc))
require.NoError(t, err)
require.Equal(t, ca.dec, s)
@@ -97,7 +98,7 @@ func FuzzUnmarshal(f *testing.F) {
}
f.Fuzz(func(_ *testing.T, buf []byte) {
var s SubGroup
var s subgroup.SubGroup
err := s.Read(bytes.NewReader(buf))
if err != nil {
return
internal/protocols/moq/varint/varint.go
+1 -1
View File
@@ -7,7 +7,7 @@ import (
)
// Varint is a variable-length integer.
// spec: draft-18, section 1.4.1
// spec: draft-18/19, section 1.4.1
type Varint uint64
// Read reads a Varint from a Reader.
internal/protocols/moq/varint/varint_test.go
+7 -6
View File
@@ -1,16 +1,17 @@
package varint
package varint_test
import (
"bytes"
"testing"
"github.com/bluenviron/mediamtx/internal/protocols/moq/varint"
"github.com/stretchr/testify/require"
)
var cases = []struct {
name string
enc []byte
dec Varint
dec varint.Varint
}{
{
name: "1 byte",
@@ -62,7 +63,7 @@ var cases = []struct {
func TestRead(t *testing.T) {
for _, ca := range cases {
t.Run(ca.name, func(t *testing.T) {
var v Varint
var v varint.Varint
err := v.Read(bytes.NewReader(ca.enc))
require.NoError(t, err)
require.Equal(t, ca.dec, v)
@@ -73,7 +74,7 @@ func TestRead(t *testing.T) {
func TestUnmarshal(t *testing.T) {
for _, ca := range cases {
t.Run(ca.name, func(t *testing.T) {
var v Varint
var v varint.Varint
n, err := v.Unmarshal(ca.enc)
require.NoError(t, err)
require.Equal(t, len(ca.enc), n)
@@ -96,7 +97,7 @@ func FuzzUnmarshal(f *testing.F) {
}
f.Fuzz(func(_ *testing.T, buf []byte) {
var v Varint
var v varint.Varint
_, err := v.Unmarshal(buf)
if err != nil {
return
@@ -112,7 +113,7 @@ func FuzzRead(f *testing.F) {
}
f.Fuzz(func(_ *testing.T, buf []byte) {
var v Varint
var v varint.Varint
err := v.Read(bytes.NewReader(buf))
if err != nil {
return
internal/servers/moq/conn.go
+19
View File
@@ -0,0 +1,19 @@
package moq
import (
"context"
"io"
"net"
"github.com/bluenviron/mediamtx/internal/defs"
)
type conn interface {
RemoteAddr() net.Addr
OpenUniStreamSync(ctx context.Context) (io.WriteCloser, error)
AcceptUniStream(ctx context.Context) (io.Reader, error)
OpenStreamSync(ctx context.Context) (io.ReadWriteCloser, error)
AcceptStream(ctx context.Context) (io.ReadWriteCloser, error)
CloseWithError(code uint64, msg string) error
Transport() defs.APIMoQSessionTransport
}
internal/servers/moq/conn_quic.go
+42
View File
@@ -0,0 +1,42 @@
package moq //nolint:dupl
import (
"context"
"io"
"net"
"github.com/bluenviron/mediamtx/internal/defs"
"github.com/quic-go/quic-go"
)
type connQUIC struct {
conn *quic.Conn
}
func (c *connQUIC) RemoteAddr() net.Addr {
return c.conn.RemoteAddr()
}
func (c *connQUIC) OpenUniStreamSync(ctx context.Context) (io.WriteCloser, error) {
return c.conn.OpenUniStreamSync(ctx)
}
func (c *connQUIC) AcceptUniStream(ctx context.Context) (io.Reader, error) {
return c.conn.AcceptUniStream(ctx)
}
func (c *connQUIC) OpenStreamSync(ctx context.Context) (io.ReadWriteCloser, error) {
return c.conn.OpenStreamSync(ctx)
}
func (c *connQUIC) AcceptStream(ctx context.Context) (io.ReadWriteCloser, error) {
return c.conn.AcceptStream(ctx)
}
func (c *connQUIC) CloseWithError(code uint64, msg string) error {
return c.conn.CloseWithError(quic.ApplicationErrorCode(code), msg)
}
func (*connQUIC) Transport() defs.APIMoQSessionTransport {
return defs.APIMoQSessionTransportQUIC
}
internal/servers/moq/conn_webtransport.go
+42
View File
@@ -0,0 +1,42 @@
package moq //nolint:dupl
import (
"context"
"io"
"net"
"github.com/bluenviron/mediamtx/internal/defs"
"github.com/quic-go/webtransport-go"
)
type connWebTransport struct {
session *webtransport.Session
}
func (c *connWebTransport) RemoteAddr() net.Addr {
return c.session.RemoteAddr()
}
func (c *connWebTransport) OpenUniStreamSync(ctx context.Context) (io.WriteCloser, error) {
return c.session.OpenUniStreamSync(ctx)
}
func (c *connWebTransport) AcceptUniStream(ctx context.Context) (io.Reader, error) {
return c.session.AcceptUniStream(ctx)
}
func (c *connWebTransport) OpenStreamSync(ctx context.Context) (io.ReadWriteCloser, error) {
return c.session.OpenStreamSync(ctx)
}
func (c *connWebTransport) AcceptStream(ctx context.Context) (io.ReadWriteCloser, error) {
return c.session.AcceptStream(ctx)
}
func (c *connWebTransport) CloseWithError(code uint64, msg string) error {
return c.session.CloseWithError(webtransport.SessionErrorCode(code), msg)
}
func (*connWebTransport) Transport() defs.APIMoQSessionTransport {
return defs.APIMoQSessionTransportWebTransport
}
internal/servers/moq/http_server.go
+1 -1
View File
@@ -350,7 +350,7 @@ func (s *httpServer) onRequestHTTPS3(ctx *gin.Context) {
query: ctx.Request.URL.RawQuery,
userAgent: ctx.Request.Header.Get("User-Agent"),
version: version,
wt: wt,
conn: &connWebTransport{session: wt},
})
if res.err != nil {
wt.CloseWithError(0, res.err.Error()) //nolint:errcheck
internal/servers/moq/native_listener.go
+144
View File
@@ -0,0 +1,144 @@
package moq
import (
"context"
"crypto/tls"
"fmt"
"net"
"os"
"strings"
"github.com/bluenviron/gortsplib/v5/pkg/readbuffer"
"github.com/bluenviron/mediamtx/internal/defs"
"github.com/bluenviron/mediamtx/internal/logger"
"github.com/quic-go/quic-go"
)
var supportedMoqtALPNs = []string{
string(defs.APIMoQVersionDraft19),
string(defs.APIMoQVersionDraft18),
}
type nativeListenerParent interface {
newSession(req newSessionReq) newSessionRes
Log(level logger.Level, format string, args ...any)
}
type nativeListener struct {
address string
serverKey string
serverCert string
udpReadBufferSize uint
parent nativeListenerParent
ctx context.Context
ctxCancel context.CancelFunc
ln net.PacketConn
listener *quic.Listener
done chan struct{}
}
func (s *nativeListener) initialize() error {
ctx, ctxCancel := context.WithCancel(context.Background())
s.ctx = ctx
s.ctxCancel = ctxCancel
s.done = make(chan struct{})
os.Setenv("QUIC_GO_DISABLE_RECEIVE_BUFFER_WARNING", "true") //nolint:errcheck
ln, err := net.ListenPacket("udp", s.address)
if err != nil {
ctxCancel()
return err
}
s.ln = ln
if s.udpReadBufferSize != 0 {
err = readbuffer.SetReadBuffer(s.ln.(*net.UDPConn), int(s.udpReadBufferSize))
if err != nil {
s.ln.Close()
ctxCancel()
return err
}
}
cert, err := tls.LoadX509KeyPair(s.serverCert, s.serverKey)
if err != nil {
s.ln.Close()
ctxCancel()
return fmt.Errorf("unable to load TLS keypair for native MoQ QUIC listener: %w", err)
}
tlsConfig := &tls.Config{
Certificates: []tls.Certificate{cert},
NextProtos: supportedMoqtALPNs,
}
listener, err := quic.Listen(s.ln, tlsConfig, &quic.Config{
EnableDatagrams: true,
})
if err != nil {
s.ln.Close()
ctxCancel()
return err
}
s.listener = listener
go s.run()
return nil
}
func alpnToVersion(alpn string) defs.APIMoQVersion {
switch alpn {
case string(defs.APIMoQVersionDraft19):
return defs.APIMoQVersionDraft19
case string(defs.APIMoQVersionDraft18):
return defs.APIMoQVersionDraft18
}
return ""
}
func (s *nativeListener) run() {
defer close(s.done)
for {
conn, err := s.listener.Accept(s.ctx)
if err != nil {
if s.ctx.Err() != nil || strings.Contains(err.Error(), "closed") {
return
}
s.parent.Log(logger.Warn, "[MoQ] native QUIC accept error: %v", err)
continue
}
version := alpnToVersion(conn.ConnectionState().TLS.NegotiatedProtocol)
if version == "" {
conn.CloseWithError(0, "unsupported ALPN") //nolint:errcheck
continue
}
res := s.parent.newSession(newSessionReq{
version: version,
conn: &connQUIC{
conn: conn,
},
})
if res.err != nil {
conn.CloseWithError(0, res.err.Error()) //nolint:errcheck
}
}
}
func (s *nativeListener) close() {
s.ctxCancel()
if s.listener != nil {
s.listener.Close() //nolint:errcheck
}
if s.ln != nil {
s.ln.Close()
}
<-s.done
}
internal/servers/moq/server.go
+29 -5
View File
@@ -12,7 +12,6 @@ import (
"github.com/bluenviron/mediamtx/internal/defs"
"github.com/bluenviron/mediamtx/internal/logger"
"github.com/google/uuid"
"github.com/quic-go/webtransport-go"
)
// ErrSessionNotFound is returned when a session is not found.
@@ -38,7 +37,7 @@ type newSessionReq struct {
query string
userAgent string
version defs.APIMoQVersion
wt *webtransport.Session
conn conn
res chan newSessionRes
}
@@ -81,6 +80,7 @@ type serverMetrics interface {
type Server struct {
HTTP2Address string
HTTP3Address string
QUICAddress string
ServerKey string
ServerCert string
AllowOrigins []string
@@ -95,6 +95,7 @@ type Server struct {
ctx context.Context
ctxCancel context.CancelFunc
httpServer *httpServer
nativeListener *nativeListener
sessions map[*session]struct{}
chNewSession chan newSessionReq
@@ -138,7 +139,25 @@ func (s *Server) Initialize() error {
return err
}
s.Log(logger.Info, "started with listeners on %s (TCP/HTTP2), %s (UDP/HTTP3)", s.HTTP2Address, s.HTTP3Address)
s.nativeListener = &nativeListener{
address: s.QUICAddress,
serverKey: s.ServerKey,
serverCert: s.ServerCert,
udpReadBufferSize: s.UDPReadBufferSize,
parent: s,
}
err = s.nativeListener.initialize()
if err != nil {
s.httpServer.close()
ctxCancel()
return err
}
s.Log(logger.Info,
"started with listeners on %s (TCP/HTTP2), %s (UDP/HTTP3), %s (UDP/QUIC)",
s.HTTP2Address,
s.HTTP3Address,
s.QUICAddress)
go s.run()
@@ -176,11 +195,12 @@ outer:
select {
case req := <-s.chNewSession:
sx := &session{
wt: req.wt,
conn: req.conn,
wg: &wg,
pathName: req.pathName,
query: req.query,
userAgent: req.userAgent,
transport: req.conn.Transport(),
version: req.version,
pathManager: s.PathManager,
parent: s,
@@ -236,11 +256,15 @@ outer:
}
}
// close sessions before closing UDP packet listener
// close sessions before closing packet listeners
for sx := range s.sessions {
sx.Close()
}
if s.nativeListener != nil {
s.nativeListener.close()
}
s.httpServer.close()
wg.Wait()
internal/servers/moq/server_test.go
+121 -7
View File
@@ -1,4 +1,4 @@
package moq
package moq_test
import (
"context"
@@ -19,6 +19,7 @@ import (
"github.com/bluenviron/mediamtx/internal/protocols/moq/controlmessage"
"github.com/bluenviron/mediamtx/internal/protocols/moq/parameter"
"github.com/bluenviron/mediamtx/internal/protocols/moq/subgroup"
"github.com/bluenviron/mediamtx/internal/servers/moq"
"github.com/bluenviron/mediamtx/internal/stream"
"github.com/bluenviron/mediamtx/internal/test"
"github.com/bluenviron/mediamtx/internal/unit"
@@ -57,9 +58,10 @@ func TestAuthError(t *testing.T) {
return nil, &auth.Error{Wrapped: fmt.Errorf("auth error")}
},
}
s := &Server{
s := &moq.Server{
HTTP2Address: "127.0.0.1:19895",
HTTP3Address: "127.0.0.1:19896",
QUICAddress: "127.0.0.1:19897",
ServerCert: serverCertFile,
ServerKey: serverKeyFile,
AllowOrigins: []string{"*"},
@@ -128,9 +130,10 @@ func TestAuthError(t *testing.T) {
}
}
s := &Server{
s := &moq.Server{
HTTP2Address: "127.0.0.1:19895",
HTTP3Address: "127.0.0.1:19896",
QUICAddress: "127.0.0.1:19897",
ServerCert: serverCertFile,
ServerKey: serverKeyFile,
AllowOrigins: []string{"*"},
@@ -307,8 +310,7 @@ func TestServer(t *testing.T) {
FindPathConfImpl: func(_ defs.PathFindPathConfReq) (*defs.PathFindPathConfRes, error) {
return &defs.PathFindPathConfRes{Conf: &conf.Path{}}, nil
},
AddReaderImpl: func(req defs.PathAddReaderReq) (*defs.PathAddReaderRes, error) {
require.Equal(t, ca.expectedVersion, req.Author.(*session).version)
AddReaderImpl: func(_ defs.PathAddReaderReq) (*defs.PathAddReaderRes, error) {
return &defs.PathAddReaderRes{Path: &serverDummyPath{}, Stream: strm}, nil
},
}
@@ -316,9 +318,10 @@ func TestServer(t *testing.T) {
serverCertFile := test.CreateTempFile(t, test.TLSCertPub)
serverKeyFile := test.CreateTempFile(t, test.TLSCertKey)
s := &Server{
s := &moq.Server{
HTTP2Address: "127.0.0.1:19895",
HTTP3Address: "127.0.0.1:19896",
QUICAddress: "127.0.0.1:19897",
ServerCert: serverCertFile,
ServerKey: serverKeyFile,
AllowOrigins: []string{"*"},
@@ -380,6 +383,11 @@ func TestServer(t *testing.T) {
require.NoError(t, err)
require.Equal(t, &controlmessage.SubscribeOk{TrackAlias: 1}, catalogOkMsg)
sessions, err := s.APISessionsList()
require.NoError(t, err)
require.Equal(t, 1, len(sessions.Items))
require.Equal(t, ca.expectedVersion, sessions.Items[0].Version)
catalogDataStream, err := sx.AcceptUniStream(ctx)
require.NoError(t, err)
@@ -486,9 +494,10 @@ func TestServerUnsupportedVersion(t *testing.T) {
serverCertFile := test.CreateTempFile(t, test.TLSCertPub)
serverKeyFile := test.CreateTempFile(t, test.TLSCertKey)
s := &Server{
s := &moq.Server{
HTTP2Address: "127.0.0.1:19895",
HTTP3Address: "127.0.0.1:19896",
QUICAddress: "127.0.0.1:19897",
ServerCert: serverCertFile,
ServerKey: serverKeyFile,
AllowOrigins: []string{"*"},
@@ -519,3 +528,108 @@ func TestServerUnsupportedVersion(t *testing.T) {
require.Error(t, err)
defer res.Body.Close()
}
func TestServerNativeQUICSubscribe(t *testing.T) {
desc := &description.Session{Medias: []*description.Media{test.UniqueMediaH264()}}
strm := &stream.Stream{
OrigDesc: desc,
WriteQueueSize: 512,
RTPMaxPayloadSize: 1450,
Parent: test.NilLogger,
}
err := strm.Initialize()
require.NoError(t, err)
defer strm.Close()
pm := &test.PathManager{
FindPathConfImpl: func(_ defs.PathFindPathConfReq) (*defs.PathFindPathConfRes, error) {
return &defs.PathFindPathConfRes{Conf: &conf.Path{}}, nil
},
AddReaderImpl: func(_ defs.PathAddReaderReq) (*defs.PathAddReaderRes, error) {
return &defs.PathAddReaderRes{Path: &serverDummyPath{}, Stream: strm}, nil
},
}
serverCertFile := test.CreateTempFile(t, test.TLSCertPub)
serverKeyFile := test.CreateTempFile(t, test.TLSCertKey)
s := &moq.Server{
HTTP2Address: "127.0.0.1:19895",
HTTP3Address: "127.0.0.1:19896",
QUICAddress: "127.0.0.1:19897",
ServerCert: serverCertFile,
ServerKey: serverKeyFile,
AllowOrigins: []string{"*"},
TrustedProxies: conf.IPNetworks{},
ReadTimeout: conf.Duration(10 * time.Second),
WriteTimeout: conf.Duration(10 * time.Second),
PathManager: pm,
Parent: test.NilLogger,
}
err = s.Initialize()
require.NoError(t, err)
defer s.Close()
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
conn, err := quic.DialAddr(ctx, "127.0.0.1:19897", &tls.Config{ //nolint:gosec
InsecureSkipVerify: true,
NextProtos: []string{string(defs.APIMoQVersionDraft19)},
}, &quic.Config{EnableDatagrams: true})
require.NoError(t, err)
defer conn.CloseWithError(0, "") //nolint:errcheck
setupStream, err := conn.AcceptUniStream(ctx)
require.NoError(t, err)
setupMsg, err := controlmessage.Read(setupStream)
require.NoError(t, err)
require.Equal(t, &controlmessage.Setup{}, setupMsg)
clientSetup, err := conn.OpenUniStreamSync(ctx)
require.NoError(t, err)
_, err = clientSetup.Write(controlmessage.Setup{Path: "/teststream"}.Marshal())
require.NoError(t, err)
catalogBidi, err := conn.OpenStreamSync(ctx)
require.NoError(t, err)
_, err = catalogBidi.Write(controlmessage.Subscribe{
RequestID: 1,
TrackName: ".catalog",
}.Marshal())
require.NoError(t, err)
catalogOkMsg, err := controlmessage.Read(catalogBidi)
require.NoError(t, err)
require.Equal(t, &controlmessage.SubscribeOk{TrackAlias: 1}, catalogOkMsg)
sessions, err := s.APISessionsList()
require.NoError(t, err)
require.Equal(t, 1, len(sessions.Items))
require.Equal(t, defs.APIMoQVersionDraft19, sessions.Items[0].Version)
require.Equal(t, defs.APIMoQSessionTransportQUIC, sessions.Items[0].Transport)
catalogDataStream, err := conn.AcceptUniStream(ctx)
require.NoError(t, err)
var catalogSG subgroup.SubGroup
err = catalogSG.Read(catalogDataStream)
require.NoError(t, err)
var cat catalog.Catalog
err = json.Unmarshal(catalogSG.Objects[0].Payload, &cat)
require.NoError(t, err)
require.Equal(t, catalog.Catalog{
Version: 1,
Tracks: []catalog.Track{{
Name: "0",
Packaging: "loc",
IsLive: true,
Codec: "avc3.640028",
}},
}, cat)
}
internal/servers/moq/session.go
+74 -29
View File
@@ -10,6 +10,7 @@ import (
"fmt"
"io"
"net"
"net/url"
"strconv"
"strings"
"sync"
@@ -28,7 +29,6 @@ import (
"github.com/bluenviron/mediamtx/internal/protocols/moq/subgroup"
"github.com/bluenviron/mediamtx/internal/stream"
"github.com/google/uuid"
"github.com/quic-go/webtransport-go"
"golang.org/x/sync/errgroup"
)
@@ -79,12 +79,14 @@ type sessionParent interface {
closeSession(sx *session)
logger.Writer
}
type session struct {
wt *webtransport.Session
conn conn
wg *sync.WaitGroup
pathName string
query string
userAgent string
transport defs.APIMoQSessionTransport
version defs.APIMoQVersion
pathManager serverPathManager
parent sessionParent
@@ -121,7 +123,7 @@ func (s *session) initialize() {
s.setupReceived = make(chan struct{})
s.done = make(chan struct{})
s.Log(logger.Info, "created by %s", s.wt.RemoteAddr())
s.Log(logger.Info, "created by %s", s.conn.RemoteAddr())
s.wg.Add(1)
go s.run()
@@ -178,19 +180,19 @@ func (s *session) runInner() error {
select {
case <-s.ctx.Done():
s.wt.CloseWithError(0, "") //nolint:errcheck
s.conn.CloseWithError(0, "") //nolint:errcheck
errGroup.Wait() //nolint:errcheck
return fmt.Errorf("terminated")
case <-errGroupCtx.Done():
s.ctxCancel()
s.wt.CloseWithError(0, "") //nolint:errcheck
s.conn.CloseWithError(0, "") //nolint:errcheck
return errGroup.Wait()
}
}
func (s *session) runSetupWriter() error {
wstream, err := s.wt.OpenUniStreamSync(context.Background())
wstream, err := s.conn.OpenUniStreamSync(context.Background())
if err != nil {
return err
}
@@ -205,7 +207,7 @@ func (s *session) runSetupWriter() error {
func (s *session) runUniStreamAcceptor(errGroup *errgroup.Group) error {
for {
stream, err := s.wt.AcceptUniStream(context.Background())
stream, err := s.conn.AcceptUniStream(context.Background())
if err != nil {
return fmt.Errorf("AcceptUniStream returned: %w", err)
}
@@ -218,7 +220,7 @@ func (s *session) runUniStreamAcceptor(errGroup *errgroup.Group) error {
func (s *session) runBidiStreamAcceptor(errGroup *errgroup.Group) error {
for {
stream, err := s.wt.AcceptStream(context.Background())
stream, err := s.conn.AcceptStream(context.Background())
if err != nil {
return fmt.Errorf("AcceptStream returned: %w", err)
}
@@ -229,7 +231,7 @@ func (s *session) runBidiStreamAcceptor(errGroup *errgroup.Group) error {
}
}
func (s *session) runUniStream(wstream *webtransport.ReceiveStream) error {
func (s *session) runUniStream(wstream io.Reader) error {
br := bufio.NewReader(wstream)
firstByte, err := br.Peek(1)
if err != nil {
@@ -249,12 +251,41 @@ func (s *session) onUniMessage(r io.Reader) error {
return err
}
switch msg.(type) {
switch m := msg.(type) {
case *controlmessage.Setup:
err = func() error {
s.mutex.Lock()
defer s.mutex.Unlock()
if s.transport == defs.APIMoQSessionTransportWebTransport {
if m.Path != "" {
return fmt.Errorf("received PATH setup option over WebTransport")
}
if m.Authority != "" {
return fmt.Errorf("received AUTHORITY setup option over WebTransport")
}
}
if s.transport == defs.APIMoQSessionTransportQUIC && s.pathName == "" {
pathWithQuery := m.Path
if pathWithQuery == "" {
return fmt.Errorf("missing PATH setup option")
}
u, err2 := url.ParseRequestURI(pathWithQuery)
if err2 != nil {
return fmt.Errorf("invalid PATH setup option: %w", err2)
}
pathName := strings.Trim(u.Path, "/")
if pathName == "" {
return fmt.Errorf("invalid PATH setup option: empty path")
}
s.pathName = pathName
s.query = u.RawQuery
}
select {
case <-s.setupReceived:
return fmt.Errorf("SETUP stream is already present")
@@ -267,15 +298,15 @@ func (s *session) onUniMessage(r io.Reader) error {
return err
}
io.Copy(io.Discard, r)
return fmt.Errorf("SETUP stream closed")
_, err = io.Copy(io.Discard, r)
return err
default:
return fmt.Errorf("unsupported stream type: %T", msg)
}
}
func (s *session) runBidiStream(wstream *webtransport.Stream) error {
func (s *session) runBidiStream(wstream io.ReadWriteCloser) error {
select {
case <-s.setupReceived:
case <-s.ctx.Done():
@@ -311,7 +342,14 @@ func (s *session) runBidiStream(wstream *webtransport.Stream) error {
}
}
func (s *session) onSubscribeCatalog(wstream *webtransport.Stream, m *controlmessage.Subscribe) error {
func (s *session) getPathNameAndQuery() (string, string) {
s.mutex.Lock()
defer s.mutex.Unlock()
return s.pathName, s.query
}
func (s *session) onSubscribeCatalog(wstream io.ReadWriteCloser, m *controlmessage.Subscribe) error {
s.mutex.Lock()
if s.state != defs.APIMoQSessionStateIdle {
s.mutex.Unlock()
@@ -320,12 +358,14 @@ func (s *session) onSubscribeCatalog(wstream *webtransport.Stream, m *controlmes
s.state = defs.APIMoQSessionStateRead
s.mutex.Unlock()
remoteHost, _, _ := net.SplitHostPort(s.wt.RemoteAddr().String())
pathName, query := s.getPathNameAndQuery()
remoteHost, _, _ := net.SplitHostPort(s.conn.RemoteAddr().String())
addRes, err := s.pathManager.AddReader(defs.PathAddReaderReq{
Author: s,
AccessRequest: defs.PathAccessRequest{
Name: s.pathName,
Query: s.query,
Name: pathName,
Query: query,
Proto: auth.ProtocolMoQ,
ID: &s.uuid,
Credentials: credentialsFromAuthorizationToken(findAuthorizationToken(m.Parameters)),
@@ -367,7 +407,7 @@ func (s *session) onSubscribeCatalog(wstream *webtransport.Stream, m *controlmes
s.setupTracks = setupTracks
s.mutex.Unlock()
s.Log(logger.Info, "is reading from path %s", s.pathName)
s.Log(logger.Info, "is reading from path %s", pathName)
_, err = wstream.Write(controlmessage.SubscribeOk{TrackAlias: m.RequestID}.Marshal())
if err != nil {
@@ -379,7 +419,7 @@ func (s *session) onSubscribeCatalog(wstream *webtransport.Stream, m *controlmes
return err
}
dataWStream, err := s.wt.OpenUniStreamSync(context.Background())
dataWStream, err := s.conn.OpenUniStreamSync(context.Background())
if err != nil {
return err
}
@@ -407,7 +447,7 @@ func (s *session) onSubscribeCatalog(wstream *webtransport.Stream, m *controlmes
return fmt.Errorf("SUBSCRIBE catalog stream closed")
}
func (s *session) onSubscribeTrack(wstream *webtransport.Stream, m *controlmessage.Subscribe) error {
func (s *session) onSubscribeTrack(wstream io.ReadWriteCloser, m *controlmessage.Subscribe) error {
trackID, err := strconv.Atoi(m.TrackName)
if err != nil || trackID < 0 {
return fmt.Errorf("invalid track name: %s", m.TrackName)
@@ -440,7 +480,7 @@ func (s *session) onSubscribeTrack(wstream *webtransport.Stream, m *controlmessa
groupID := uint64(0)
writeData := func(payload []byte, pts int64) error {
wstream, err2 := s.wt.OpenUniStreamSync(context.Background())
wstream, err2 := s.conn.OpenUniStreamSync(context.Background())
if err2 != nil {
return err2
}
@@ -496,7 +536,7 @@ func (s *session) onSubscribeTrack(wstream *webtransport.Stream, m *controlmessa
}
}
func (s *session) onPublishCatalog(wstream *webtransport.Stream, m *controlmessage.Publish) error {
func (s *session) onPublishCatalog(wstream io.ReadWriteCloser, m *controlmessage.Publish) error {
s.mutex.Lock()
if s.state != defs.APIMoQSessionStateIdle {
s.mutex.Unlock()
@@ -505,6 +545,8 @@ func (s *session) onPublishCatalog(wstream *webtransport.Stream, m *controlmessa
s.state = defs.APIMoQSessionStatePublish
s.mutex.Unlock()
pathName, query := s.getPathNameAndQuery()
select {
case cat := <-s.catalogReceived:
var subStream *stream.SubStream
@@ -526,15 +568,15 @@ func (s *session) onPublishCatalog(wstream *webtransport.Stream, m *controlmessa
s.inboundTracks[trackAlias] = tr
}
remoteHost, _, _ := net.SplitHostPort(s.wt.RemoteAddr().String())
remoteHost, _, _ := net.SplitHostPort(s.conn.RemoteAddr().String())
addRes, err := s.pathManager.AddPublisher(defs.PathAddPublisherReq{
Author: s,
Desc: &description.Session{Medias: medias},
UseRTPPackets: false,
ReplaceNTP: true,
AccessRequest: defs.PathAccessRequest{
Name: s.pathName,
Query: s.query,
Name: pathName,
Query: query,
Publish: true,
Proto: auth.ProtocolMoQ,
ID: &s.uuid,
@@ -587,7 +629,7 @@ func (s *session) onPublishCatalog(wstream *webtransport.Stream, m *controlmessa
return fmt.Errorf("PUBLISH catalog stream closed")
}
func (s *session) onPublishTrack(wstream *webtransport.Stream) error {
func (s *session) onPublishTrack(wstream io.ReadWriteCloser) error {
s.mutex.Lock()
if s.state != defs.APIMoQSessionStatePublish {
s.mutex.Unlock()
@@ -663,16 +705,19 @@ func (s *session) onDataTrack(r io.Reader, sg *subgroup.SubGroup) error {
func (s *session) apiItem() defs.APIMoQSession {
s.mutex.Lock()
state := s.state
pathName := s.pathName
query := s.query
s.mutex.Unlock()
return defs.APIMoQSession{
ID: s.uuid,
Created: s.created,
RemoteAddr: s.wt.RemoteAddr().String(),
RemoteAddr: s.conn.RemoteAddr().String(),
State: state,
Path: s.pathName,
Query: s.query,
Path: pathName,
Query: query,
UserAgent: s.userAgent,
Transport: s.transport,
Version: s.version,
InboundBytes: s.inboundBytes.Load(),
OutboundBytes: s.outboundBytes.Load(),
mediamtx.yml
+2
View File
@@ -449,6 +449,8 @@ moqHTTP2Address: :8892
# Address of the UDP/HTTP3 listener.
# This hosts the WebTransport endpoint.
moqHTTP3Address: :8892
# Address of the UDP/QUIC listener used for native MoQ-over-QUIC.
moqQUICAddress: :8893
# Path to the server key.
# This can be generated with:
# openssl genrsa -out server.key 2048