moq: support draft-17 (#6040)

api/openapi.yaml

@@ -98,6 +98,7 @@ components:
     MoQVersion:
       type: string
       enum:
+        - moqt-17
         - moqt-18
         - moqt-19
 

docs/1-kickoff/2-install.md

@@ -34,6 +34,7 @@ docker run --rm -it \
 -p 8890:8890/udp \
 -p 8189:8189/udp \
 -p 8892:8892/udp \
+-p 8893:8893/udp \
 bluenviron/mediamtx:1
 ```
 

docs/3-publish/01-moq-clients.md

@@ -9,7 +9,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.
+- The server supports `draft-19`, `draft-18` and `draft-17` of the [main specification](https://datatracker.ietf.org/doc/html/draft-ietf-moq-transport-19), and prefers `draft-19` when multiple versions are offered during negotiation.
 - 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).

internal/defs/api_moq.go

@@ -28,6 +28,7 @@ type APIMoQVersion string
 
 // protocol versions.
 const (
+	APIMoQVersionDraft17 APIMoQVersion = "moqt-17"
 	APIMoQVersionDraft18 APIMoQVersion = "moqt-18"
 	APIMoQVersionDraft19 APIMoQVersion = "moqt-19"
 )

internal/protocols/moq/controlmessage/message.go

@@ -50,6 +50,8 @@ func Read(r io.Reader) (Message, error) {
 		m = &RequestError{}
 	case typePublish:
 		m = &Publish{}
+	case typePublishOk:
+		m = &PublishOk{}
 	case typeRequestOk:
 		m = &RequestOk{}
 	default:

internal/protocols/moq/controlmessage/message_test.go

@@ -124,6 +124,15 @@ var cases = []struct {
 			Reason: "foo",
 		},
 	},
+	{
+		name: "publish_ok",
+		enc: []byte{
+			0x1E,       // type 0x1E
+			0x00, 0x01, // length = 1
+			0x00, // Number of Parameters = 0
+		},
+		dec: &controlmessage.PublishOk{},
+	},
 	{
 		name: "request_ok",
 		enc: []byte{

internal/protocols/moq/controlmessage/publish_ok.go

@@ -0,0 +1,64 @@
+package controlmessage //nolint:dupl
+
+import (
+	"github.com/bluenviron/mediamtx/internal/protocols/moq/parameter"
+	"github.com/bluenviron/mediamtx/internal/protocols/moq/property"
+	"github.com/bluenviron/mediamtx/internal/protocols/moq/varint"
+)
+
+const typePublishOk varint.Varint = 0x1E
+
+// PublishOk is the PUBLISH_OK control message.
+// spec: draft-17, section 9.12
+// spec: draft-18/19, section 10.5 (alias of REQUEST_OK)
+type PublishOk struct {
+	Parameters      parameter.Parameters
+	TrackProperties property.Properties
+}
+
+func (*PublishOk) isMessage() {}
+
+func (m *PublishOk) unmarshal(buf []byte) error {
+	var numParams varint.Varint
+	n, err := numParams.Unmarshal(buf)
+	if err != nil {
+		return err
+	}
+	buf = buf[n:]
+
+	consumed, err := m.Parameters.Unmarshal(int(numParams), buf)
+	if err != nil {
+		return err
+	}
+	buf = buf[consumed:]
+
+	return m.TrackProperties.Unmarshal(buf)
+}
+
+func (m PublishOk) marshalSize() int {
+	payloadSize := varint.Varint(len(m.Parameters)).MarshalSize() +
+		m.Parameters.MarshalSize() +
+		m.TrackProperties.MarshalSize()
+	return typePublishOk.MarshalSize() + 2 + payloadSize
+}
+
+func (m PublishOk) marshalTo(buf []byte) int {
+	payloadSize := varint.Varint(len(m.Parameters)).MarshalSize() +
+		m.Parameters.MarshalSize() +
+		m.TrackProperties.MarshalSize()
+	n := typePublishOk.MarshalTo(buf)
+	buf[n] = byte(payloadSize >> 8)
+	buf[n+1] = byte(payloadSize)
+	n += 2
+	n += varint.Varint(len(m.Parameters)).MarshalTo(buf[n:])
+	n += m.Parameters.MarshalTo(buf[n:])
+	n += m.TrackProperties.MarshalTo(buf[n:])
+	return n
+}
+
+// Marshal implements Message.
+func (m PublishOk) Marshal() []byte {
+	buf := make([]byte, m.marshalSize())
+	m.marshalTo(buf)
+	return buf
+}

internal/protocols/moq/controlmessage/request_ok.go

@@ -1,4 +1,4 @@
-package controlmessage
+package controlmessage //nolint:dupl
 
 import (
 	"github.com/bluenviron/mediamtx/internal/protocols/moq/parameter"

internal/servers/moq/http_server.go

@@ -45,6 +45,7 @@ const (
 var supportedMoqtVersions = []defs.APIMoQVersion{
 	defs.APIMoQVersionDraft19,
 	defs.APIMoQVersionDraft18,
+	defs.APIMoQVersionDraft17,
 }
 
 type ginUnwrapper interface {

internal/servers/moq/native_listener.go

@@ -17,6 +17,7 @@ import (
 var supportedMoqtALPNs = []string{
 	string(defs.APIMoQVersionDraft19),
 	string(defs.APIMoQVersionDraft18),
+	string(defs.APIMoQVersionDraft17),
 }
 
 type nativeListenerParent interface {
@@ -96,6 +97,9 @@ func alpnToVersion(alpn string) defs.APIMoQVersion {
 
 	case string(defs.APIMoQVersionDraft18):
 		return defs.APIMoQVersionDraft18
+
+	case string(defs.APIMoQVersionDraft17):
+		return defs.APIMoQVersionDraft17
 	}
 
 	return ""

internal/servers/moq/server_test.go

@@ -271,6 +271,11 @@ func TestServer(t *testing.T) {
 		clientProtocols []string
 		expectedVersion defs.APIMoQVersion
 	}{
+		{
+			name:            "draft-17",
+			clientProtocols: []string{"moqt-17"},
+			expectedVersion: defs.APIMoQVersionDraft17,
+		},
 		{
 			name:            "draft-18",
 			clientProtocols: []string{"moqt-18"},
@@ -286,6 +291,11 @@ func TestServer(t *testing.T) {
 			clientProtocols: []string{"moqt-19", "moqt-18"},
 			expectedVersion: defs.APIMoQVersionDraft19,
 		},
+		{
+			name:            "highest-preferred",
+			clientProtocols: []string{"moqt-17", "moqt-18", "moqt-19"},
+			expectedVersion: defs.APIMoQVersionDraft19,
+		},
 	} {
 		t.Run(ca.name, func(t *testing.T) {
 			desc := &description.Session{Medias: []*description.Media{test.UniqueMediaH264()}}

internal/servers/moq/session.go

@@ -620,7 +620,14 @@ func (s *session) onPublishCatalog(wstream io.ReadWriteCloser, m *controlmessage
 		return fmt.Errorf("terminated")
 	}
 
-	_, err := wstream.Write(controlmessage.RequestOk{}.Marshal())
+	var ackPayload []byte
+	if s.version == defs.APIMoQVersionDraft17 {
+		ackPayload = controlmessage.PublishOk{}.Marshal()
+	} else {
+		ackPayload = controlmessage.RequestOk{}.Marshal()
+	}
+
+	_, err := wstream.Write(ackPayload)
 	if err != nil {
 		return err
 	}
@@ -637,7 +644,14 @@ func (s *session) onPublishTrack(wstream io.ReadWriteCloser) error {
 	}
 	s.mutex.Unlock()
 
-	_, err := wstream.Write(controlmessage.RequestOk{}.Marshal())
+	var ackPayload []byte
+	if s.version == defs.APIMoQVersionDraft17 {
+		ackPayload = controlmessage.PublishOk{}.Marshal()
+	} else {
+		ackPayload = controlmessage.RequestOk{}.Marshal()
+	}
+
+	_, err := wstream.Write(ackPayload)
 	if err != nil {
 		return err
 	}