From 8ac6ba64c8e7e5394b57c819c7c589a7ce690e4b Mon Sep 17 00:00:00 2001 From: Alessandro Ros Date: Tue, 18 Aug 2026 13:23:40 +0200 Subject: [PATCH] moq: support pulling streams from other servers (#6111) --- api/openapi.yaml | 4 + docs/2-features/03-publish.md | 41 +- docs/2-features/11-forward.md | 2 +- docs/2-features/24-srt-specific-features.md | 2 +- .../2-features/25-webrtc-specific-features.md | 2 +- docs/2-features/26-rtsp-specific-features.md | 4 +- docs/2-features/27-rtmp-specific-features.md | 2 +- docs/3-publish/02-moq-servers.md | 16 + .../{02-srt-clients.md => 03-srt-clients.md} | 2 +- ...rvers.md => 04-srt-cameras-and-servers.md} | 0 ...webrtc-clients.md => 05-webrtc-clients.md} | 2 +- ...webrtc-servers.md => 06-webrtc-servers.md} | 0 ...{06-rtsp-clients.md => 07-rtsp-clients.md} | 2 +- ...vers.md => 08-rtsp-cameras-and-servers.md} | 0 ...{08-rtmp-clients.md => 09-rtmp-clients.md} | 2 +- ...vers.md => 10-rtmp-cameras-and-servers.md} | 0 ...rvers.md => 11-hls-cameras-and-servers.md} | 0 .../{11-mpeg-ts.md => 12-mpeg-ts.md} | 2 +- docs/3-publish/{12-rtp.md => 13-rtp.md} | 2 +- ...-cameras.md => 14-raspberry-pi-cameras.md} | 0 ...neric-webcams.md => 15-generic-webcams.md} | 0 ...{15-web-browsers.md => 16-web-browsers.md} | 2 +- docs/3-publish/{16-ffmpeg.md => 17-ffmpeg.md} | 4 +- .../{17-gstreamer.md => 18-gstreamer.md} | 4 +- .../{18-obs-studio.md => 19-obs-studio.md} | 4 +- ...9-python-opencv.md => 20-python-opencv.md} | 2 +- docs/3-publish/{20-golang.md => 21-golang.md} | 0 docs/3-publish/{21-unity.md => 22-unity.md} | 2 +- internal/conf/conf_test.go | 5 + internal/conf/moq_transport.go | 38 ++ internal/conf/path.go | 12 + internal/defs/api_path.go | 1 + internal/staticsources/handler.go | 7 + internal/staticsources/handler_test.go | 7 + internal/staticsources/moq/source.go | 539 ++++++++++++++++ internal/staticsources/moq/source_test.go | 607 ++++++++++++++++++ mediamtx.yml | 7 + 37 files changed, 1285 insertions(+), 41 deletions(-) create mode 100644 docs/3-publish/02-moq-servers.md rename docs/3-publish/{02-srt-clients.md => 03-srt-clients.md} (90%) rename docs/3-publish/{03-srt-cameras-and-servers.md => 04-srt-cameras-and-servers.md} (100%) rename docs/3-publish/{04-webrtc-clients.md => 05-webrtc-clients.md} (91%) rename docs/3-publish/{05-webrtc-servers.md => 06-webrtc-servers.md} (100%) rename docs/3-publish/{06-rtsp-clients.md => 07-rtsp-clients.md} (90%) rename docs/3-publish/{07-rtsp-cameras-and-servers.md => 08-rtsp-cameras-and-servers.md} (100%) rename docs/3-publish/{08-rtmp-clients.md => 09-rtmp-clients.md} (86%) rename docs/3-publish/{09-rtmp-cameras-and-servers.md => 10-rtmp-cameras-and-servers.md} (100%) rename docs/3-publish/{10-hls-cameras-and-servers.md => 11-hls-cameras-and-servers.md} (100%) rename docs/3-publish/{11-mpeg-ts.md => 12-mpeg-ts.md} (96%) rename docs/3-publish/{12-rtp.md => 13-rtp.md} (97%) rename docs/3-publish/{13-raspberry-pi-cameras.md => 14-raspberry-pi-cameras.md} (100%) rename docs/3-publish/{14-generic-webcams.md => 15-generic-webcams.md} (100%) rename docs/3-publish/{15-web-browsers.md => 16-web-browsers.md} (98%) rename docs/3-publish/{16-ffmpeg.md => 17-ffmpeg.md} (83%) rename docs/3-publish/{17-gstreamer.md => 18-gstreamer.md} (88%) rename docs/3-publish/{18-obs-studio.md => 19-obs-studio.md} (98%) rename docs/3-publish/{19-python-opencv.md => 20-python-opencv.md} (96%) rename docs/3-publish/{20-golang.md => 21-golang.md} (100%) rename docs/3-publish/{21-unity.md => 22-unity.md} (98%) create mode 100644 internal/conf/moq_transport.go create mode 100644 internal/staticsources/moq/source.go create mode 100644 internal/staticsources/moq/source_test.go diff --git a/api/openapi.yaml b/api/openapi.yaml index cfd58ed1..c444db15 100644 --- a/api/openapi.yaml +++ b/api/openapi.yaml @@ -141,6 +141,7 @@ components: - rtpSource - webRTCSession - webRTCSource + - moqSource - moqSession ForwardDestProtocol: @@ -947,6 +948,9 @@ components: maxReaders: type: integer format: int64 + moqTransport: + type: string + description: MoQ source mpegtsUDPReadBufferSize: type: integer format: uint64 diff --git a/docs/2-features/03-publish.md b/docs/2-features/03-publish.md index 32ca5aca..f64e9e42 100644 --- a/docs/2-features/03-publish.md +++ b/docs/2-features/03-publish.md @@ -3,29 +3,30 @@ Live streams can be published to the server with the following protocols: - [Media-over-QUIC clients](../3-publish/01-moq-clients.md) -- [SRT clients](../3-publish/02-srt-clients.md) -- [SRT cameras and servers](../3-publish/03-srt-cameras-and-servers.md) -- [WebRTC clients](../3-publish/04-webrtc-clients.md) -- [WebRTC servers](../3-publish/05-webrtc-servers.md) -- [RTSP clients](../3-publish/06-rtsp-clients.md) -- [RTSP cameras and servers](../3-publish/07-rtsp-cameras-and-servers.md) -- [RTMP clients](../3-publish/08-rtmp-clients.md) -- [RTMP cameras and servers](../3-publish/09-rtmp-cameras-and-servers.md) -- [HLS cameras and servers](../3-publish/10-hls-cameras-and-servers.md) -- [MPEG-TS](../3-publish/11-mpeg-ts.md) -- [RTP](../3-publish/12-rtp.md) +- [Media-over-QUIC servers](../3-publish/02-moq-servers.md) +- [SRT clients](../3-publish/03-srt-clients.md) +- [SRT cameras and servers](../3-publish/04-srt-cameras-and-servers.md) +- [WebRTC clients](../3-publish/05-webrtc-clients.md) +- [WebRTC servers](../3-publish/06-webrtc-servers.md) +- [RTSP clients](../3-publish/07-rtsp-clients.md) +- [RTSP cameras and servers](../3-publish/08-rtsp-cameras-and-servers.md) +- [RTMP clients](../3-publish/09-rtmp-clients.md) +- [RTMP cameras and servers](../3-publish/10-rtmp-cameras-and-servers.md) +- [HLS cameras and servers](../3-publish/11-hls-cameras-and-servers.md) +- [MPEG-TS](../3-publish/12-mpeg-ts.md) +- [RTP](../3-publish/13-rtp.md) We provide instructions for publishing with the following devices: -- [Raspberry Pi Cameras](../3-publish/13-raspberry-pi-cameras.md) -- [Generic webcams](../3-publish/14-generic-webcams.md) +- [Raspberry Pi Cameras](../3-publish/14-raspberry-pi-cameras.md) +- [Generic webcams](../3-publish/15-generic-webcams.md) We provide instructions for publishing with the following software: -- [Web browsers](../3-publish/15-web-browsers.md) -- [FFmpeg](../3-publish/16-ffmpeg.md) -- [GStreamer](../3-publish/17-gstreamer.md) -- [OBS Studio](../3-publish/18-obs-studio.md) -- [Python and OpenCV](../3-publish/19-python-opencv.md) -- [Golang](../3-publish/20-golang.md) -- [Unity](../3-publish/21-unity.md) +- [Web browsers](../3-publish/16-web-browsers.md) +- [FFmpeg](../3-publish/17-ffmpeg.md) +- [GStreamer](../3-publish/18-gstreamer.md) +- [OBS Studio](../3-publish/19-obs-studio.md) +- [Python and OpenCV](../3-publish/20-python-opencv.md) +- [Golang](../3-publish/21-golang.md) +- [Unity](../3-publish/22-unity.md) diff --git a/docs/2-features/11-forward.md b/docs/2-features/11-forward.md index c70938c8..92d41a54 100644 --- a/docs/2-features/11-forward.md +++ b/docs/2-features/11-forward.md @@ -40,7 +40,7 @@ paths: whipBearerToken: "" ``` -If the remote server is a _MediaMTX_ instance, remember to add a `/whip` suffix after the stream name, since in _MediaMTX_ [it's part of the WHIP URL](../3-publish/04-webrtc-clients.md). +If the remote server is a _MediaMTX_ instance, remember to add a `/whip` suffix after the stream name, since in _MediaMTX_ [it's part of the WHIP URL](../3-publish/05-webrtc-clients.md). ## RTSP diff --git a/docs/2-features/24-srt-specific-features.md b/docs/2-features/24-srt-specific-features.md index 0bde873d..b0792890 100644 --- a/docs/2-features/24-srt-specific-features.md +++ b/docs/2-features/24-srt-specific-features.md @@ -1,6 +1,6 @@ # SRT-specific features -SRT is a protocol that can be used for publishing and reading streams. Regarding specific tasks, check out [Publish with SRT clients](../3-publish/02-srt-clients.md) and [Read with SRT clients](../4-read/02-srt.md). Features in this page are shared among both tasks. +SRT is a protocol that can be used for publishing and reading streams. Regarding specific tasks, check out [Publish with SRT clients](../3-publish/03-srt-clients.md) and [Read with SRT clients](../4-read/02-srt.md). Features in this page are shared among both tasks. ## Standard stream ID syntax diff --git a/docs/2-features/25-webrtc-specific-features.md b/docs/2-features/25-webrtc-specific-features.md index 093338c2..ab8c86ec 100644 --- a/docs/2-features/25-webrtc-specific-features.md +++ b/docs/2-features/25-webrtc-specific-features.md @@ -1,6 +1,6 @@ # WebRTC-specific features -WebRTC is a protocol that can be used for publishing and reading streams. Regarding specific tasks, check out [Publish with WebRTC clients](../3-publish/04-webrtc-clients.md) and [Read with WebRTC clients](../4-read/03-webrtc.md). Features in this page are shared among both tasks. +WebRTC is a protocol that can be used for publishing and reading streams. Regarding specific tasks, check out [Publish with WebRTC clients](../3-publish/05-webrtc-clients.md) and [Read with WebRTC clients](../4-read/03-webrtc.md). Features in this page are shared among both tasks. ## Codec support in browsers diff --git a/docs/2-features/26-rtsp-specific-features.md b/docs/2-features/26-rtsp-specific-features.md index 48ba2968..ee766f0d 100644 --- a/docs/2-features/26-rtsp-specific-features.md +++ b/docs/2-features/26-rtsp-specific-features.md @@ -1,6 +1,6 @@ # RTSP-specific features -RTSP is a protocol that can be used for publishing and reading streams. Regarding specific tasks, check out [Publish with RTSP clients](../3-publish/06-rtsp-clients.md) and [Read with RTSP clients](../4-read/04-rtsp.md). Features in this page are shared among both tasks. +RTSP is a protocol that can be used for publishing and reading streams. Regarding specific tasks, check out [Publish with RTSP clients](../3-publish/07-rtsp-clients.md) and [Read with RTSP clients](../4-read/04-rtsp.md). Features in this page are shared among both tasks. ## Transport protocols @@ -107,4 +107,4 @@ There are also the `rtsps+http`, `rtsp+ws`, `rtsps+ws` schemes to handle any var ## MPEG-TS inside RTSP -read [MPEG-TS inside RTSP](../3-publish/06-rtsp-clients.md#mpeg-ts-inside-rtsp) in the "Publish with RTSP clients" page. +read [MPEG-TS inside RTSP](../3-publish/07-rtsp-clients.md#mpeg-ts-inside-rtsp) in the "Publish with RTSP clients" page. diff --git a/docs/2-features/27-rtmp-specific-features.md b/docs/2-features/27-rtmp-specific-features.md index b641b373..da913720 100644 --- a/docs/2-features/27-rtmp-specific-features.md +++ b/docs/2-features/27-rtmp-specific-features.md @@ -1,6 +1,6 @@ # RTMP-specific features -RTMP is a protocol that can be used for publishing and reading streams. Regarding specific tasks, check out [Publish with RTMP clients](../3-publish/08-rtmp-clients.md) and [Read with RTMP clients](../4-read/05-rtmp.md). Features in this page are shared among both tasks. +RTMP is a protocol that can be used for publishing and reading streams. Regarding specific tasks, check out [Publish with RTMP clients](../3-publish/09-rtmp-clients.md) and [Read with RTMP clients](../4-read/05-rtmp.md). Features in this page are shared among both tasks. ## Encryption diff --git a/docs/3-publish/02-moq-servers.md b/docs/3-publish/02-moq-servers.md new file mode 100644 index 00000000..2f0a5983 --- /dev/null +++ b/docs/3-publish/02-moq-servers.md @@ -0,0 +1,16 @@ +# Media-over-QUIC servers + +| | supported codecs | +| --------- | ------------------------- | +| **video** | AV1, VP9, VP8, H265, H264 | +| **audio** | Opus, MPEG-4 Audio (AAC) | + +In order to ingest a Media-over-QUIC stream from a remote server, add the corresponding URL into the `source` parameter of a path: + +```yml +paths: + proxied: + source: moqt://user:pass@host:port/path + # Transport protocol used to pull the stream. available values are "quic", "webtransport". + moqTransport: quic +``` diff --git a/docs/3-publish/02-srt-clients.md b/docs/3-publish/03-srt-clients.md similarity index 90% rename from docs/3-publish/02-srt-clients.md rename to docs/3-publish/03-srt-clients.md index 5e415a07..afc540b1 100644 --- a/docs/3-publish/02-srt-clients.md +++ b/docs/3-publish/03-srt-clients.md @@ -18,4 +18,4 @@ If you need to use the standard stream ID syntax instead of the custom one in us If you want to publish a stream by using a client in listening mode (i.e. with `mode=listener` appended to the URL), read the next section. -Some clients that can publish with SRT are [FFmpeg](16-ffmpeg.md), [GStreamer](17-gstreamer.md), [OBS Studio](18-obs-studio.md). +Some clients that can publish with SRT are [FFmpeg](17-ffmpeg.md), [GStreamer](18-gstreamer.md), [OBS Studio](19-obs-studio.md). diff --git a/docs/3-publish/03-srt-cameras-and-servers.md b/docs/3-publish/04-srt-cameras-and-servers.md similarity index 100% rename from docs/3-publish/03-srt-cameras-and-servers.md rename to docs/3-publish/04-srt-cameras-and-servers.md diff --git a/docs/3-publish/04-webrtc-clients.md b/docs/3-publish/05-webrtc-clients.md similarity index 91% rename from docs/3-publish/04-webrtc-clients.md rename to docs/3-publish/05-webrtc-clients.md index bb26e23d..92acfe47 100644 --- a/docs/3-publish/04-webrtc-clients.md +++ b/docs/3-publish/05-webrtc-clients.md @@ -23,4 +23,4 @@ Be aware that not all browsers can publish tracks with any codec, check [Codec s Depending on the network it might be difficult to establish a connection between server and clients, read [Solving WebRTC connectivity issues](../2-features/25-webrtc-specific-features.md#solving-webrtc-connectivity-issues). -Some clients that can publish with WebRTC and WHIP are [FFmpeg](16-ffmpeg.md), [GStreamer](17-gstreamer.md), [OBS Studio](18-obs-studio.md), [Unity](21-unity.md) and [Web browsers](15-web-browsers.md). +Some clients that can publish with WebRTC and WHIP are [FFmpeg](17-ffmpeg.md), [GStreamer](18-gstreamer.md), [OBS Studio](19-obs-studio.md), [Unity](22-unity.md) and [Web browsers](16-web-browsers.md). diff --git a/docs/3-publish/05-webrtc-servers.md b/docs/3-publish/06-webrtc-servers.md similarity index 100% rename from docs/3-publish/05-webrtc-servers.md rename to docs/3-publish/06-webrtc-servers.md diff --git a/docs/3-publish/06-rtsp-clients.md b/docs/3-publish/07-rtsp-clients.md similarity index 90% rename from docs/3-publish/06-rtsp-clients.md rename to docs/3-publish/07-rtsp-clients.md index 70fc426b..32892834 100644 --- a/docs/3-publish/06-rtsp-clients.md +++ b/docs/3-publish/07-rtsp-clients.md @@ -14,7 +14,7 @@ rtsp://localhost:8554/mystream The resulting stream will be available on path `/mystream`. -Some clients that can publish with RTSP are [FFmpeg](16-ffmpeg.md), [GStreamer](17-gstreamer.md), [OBS Studio](18-obs-studio.md), [Python and OpenCV](19-python-opencv.md). +Some clients that can publish with RTSP are [FFmpeg](17-ffmpeg.md), [GStreamer](18-gstreamer.md), [OBS Studio](19-obs-studio.md), [Python and OpenCV](20-python-opencv.md). Advanced RTSP features and settings are described in [RTSP-specific features](../2-features/26-rtsp-specific-features.md). diff --git a/docs/3-publish/07-rtsp-cameras-and-servers.md b/docs/3-publish/08-rtsp-cameras-and-servers.md similarity index 100% rename from docs/3-publish/07-rtsp-cameras-and-servers.md rename to docs/3-publish/08-rtsp-cameras-and-servers.md diff --git a/docs/3-publish/08-rtmp-clients.md b/docs/3-publish/09-rtmp-clients.md similarity index 86% rename from docs/3-publish/08-rtmp-clients.md rename to docs/3-publish/09-rtmp-clients.md index 7b2359da..8c28fd10 100644 --- a/docs/3-publish/08-rtmp-clients.md +++ b/docs/3-publish/09-rtmp-clients.md @@ -13,4 +13,4 @@ rtmp://localhost/mystream The resulting stream will be available on path `/mystream`. -Some clients that can publish with RTMP are [FFmpeg](16-ffmpeg.md), [GStreamer](17-gstreamer.md) and [OBS Studio](18-obs-studio.md). +Some clients that can publish with RTMP are [FFmpeg](17-ffmpeg.md), [GStreamer](18-gstreamer.md) and [OBS Studio](19-obs-studio.md). diff --git a/docs/3-publish/09-rtmp-cameras-and-servers.md b/docs/3-publish/10-rtmp-cameras-and-servers.md similarity index 100% rename from docs/3-publish/09-rtmp-cameras-and-servers.md rename to docs/3-publish/10-rtmp-cameras-and-servers.md diff --git a/docs/3-publish/10-hls-cameras-and-servers.md b/docs/3-publish/11-hls-cameras-and-servers.md similarity index 100% rename from docs/3-publish/10-hls-cameras-and-servers.md rename to docs/3-publish/11-hls-cameras-and-servers.md diff --git a/docs/3-publish/11-mpeg-ts.md b/docs/3-publish/12-mpeg-ts.md similarity index 96% rename from docs/3-publish/11-mpeg-ts.md rename to docs/3-publish/12-mpeg-ts.md index 0bf64b17..2aa7b3a5 100644 --- a/docs/3-publish/11-mpeg-ts.md +++ b/docs/3-publish/12-mpeg-ts.md @@ -40,7 +40,7 @@ paths: source: udp+mpegts://0.0.0.0:1234?source=192.168.3.5 ``` -Some clients that can publish with UDP and MPEG-TS are [FFmpeg](16-ffmpeg.md) and [GStreamer](17-gstreamer.md). +Some clients that can publish with UDP and MPEG-TS are [FFmpeg](17-ffmpeg.md) and [GStreamer](18-gstreamer.md). Unix sockets are more efficient than UDP packets and can be used as transport by specifying the `unix+mpegts` scheme: diff --git a/docs/3-publish/12-rtp.md b/docs/3-publish/13-rtp.md similarity index 97% rename from docs/3-publish/12-rtp.md rename to docs/3-publish/13-rtp.md index cbd3cde7..0556ea36 100644 --- a/docs/3-publish/12-rtp.md +++ b/docs/3-publish/13-rtp.md @@ -58,4 +58,4 @@ paths: a=rtpmap:96 H264/90000 ``` -Some clients that can publish with UDP and MPEG-TS are [FFmpeg](16-ffmpeg.md) and [GStreamer](17-gstreamer.md). +Some clients that can publish with UDP and MPEG-TS are [FFmpeg](17-ffmpeg.md) and [GStreamer](18-gstreamer.md). diff --git a/docs/3-publish/13-raspberry-pi-cameras.md b/docs/3-publish/14-raspberry-pi-cameras.md similarity index 100% rename from docs/3-publish/13-raspberry-pi-cameras.md rename to docs/3-publish/14-raspberry-pi-cameras.md diff --git a/docs/3-publish/14-generic-webcams.md b/docs/3-publish/15-generic-webcams.md similarity index 100% rename from docs/3-publish/14-generic-webcams.md rename to docs/3-publish/15-generic-webcams.md diff --git a/docs/3-publish/15-web-browsers.md b/docs/3-publish/16-web-browsers.md similarity index 98% rename from docs/3-publish/15-web-browsers.md rename to docs/3-publish/16-web-browsers.md index b4a18602..6675f085 100644 --- a/docs/3-publish/15-web-browsers.md +++ b/docs/3-publish/16-web-browsers.md @@ -1,6 +1,6 @@ # Web browsers -Web browsers can publish a stream to the server by acting as a [Media-over-QUIC client](01-moq-clients.md) or a [WebRTC client](04-webrtc-clients.md). +Web browsers can publish a stream to the server by acting as a [Media-over-QUIC client](01-moq-clients.md) or a [WebRTC client](05-webrtc-clients.md). ## Media-over-QUIC diff --git a/docs/3-publish/16-ffmpeg.md b/docs/3-publish/17-ffmpeg.md similarity index 83% rename from docs/3-publish/16-ffmpeg.md rename to docs/3-publish/17-ffmpeg.md index 7e2411b2..7121dcc4 100644 --- a/docs/3-publish/16-ffmpeg.md +++ b/docs/3-publish/17-ffmpeg.md @@ -1,6 +1,6 @@ # FFmpeg -FFmpeg can publish a stream to the server by acting as a [RTSP client](06-rtsp-clients.md), [RTMP client](08-rtmp-clients.md), [SRT client](02-srt-clients.md), [WebRTC client](04-webrtc-clients.md) or by sending [MPEG-TS packets](11-mpeg-ts.md) or [RTP packets](12-rtp.md). The recommended way is acting as a RTSP client. +FFmpeg can publish a stream to the server by acting as a [RTSP client](07-rtsp-clients.md), [RTMP client](09-rtmp-clients.md), [SRT client](03-srt-clients.md), [WebRTC client](05-webrtc-clients.md) or by sending [MPEG-TS packets](12-mpeg-ts.md) or [RTP packets](13-rtp.md). The recommended way is acting as a RTSP client. ## FFmpeg as a RTSP client @@ -52,7 +52,7 @@ ffmpeg -re -f lavfi -i testsrc=size=1280x720:rate=30 \ ## FFmpeg and RTP over UDP -In _MediaMTX_ configuration, add a path with `source: udp+rtp://238.0.0.1:1234` and a valid `rtpSDP` (read [RTP](12-rtp.md)). Then: +In _MediaMTX_ configuration, add a path with `source: udp+rtp://238.0.0.1:1234` and a valid `rtpSDP` (read [RTP](13-rtp.md)). Then: ```sh ffmpeg -re -f lavfi -i testsrc=size=1280x720:rate=30 \ diff --git a/docs/3-publish/17-gstreamer.md b/docs/3-publish/18-gstreamer.md similarity index 88% rename from docs/3-publish/17-gstreamer.md rename to docs/3-publish/18-gstreamer.md index ae4726d4..7d1a9a06 100644 --- a/docs/3-publish/17-gstreamer.md +++ b/docs/3-publish/18-gstreamer.md @@ -1,6 +1,6 @@ # GStreamer -GStreamer can publish a stream to the server by acting as a [RTSP client](06-rtsp-clients.md), [RTMP client](08-rtmp-clients.md), [SRT client](02-srt-clients.md), [WebRTC client](04-webrtc-clients.md) or by sending [MPEG-TS packets](11-mpeg-ts.md) or [RTP packets](12-rtp.md). The recommended way is acting as a RTSP client. +GStreamer can publish a stream to the server by acting as a [RTSP client](07-rtsp-clients.md), [RTMP client](09-rtmp-clients.md), [SRT client](03-srt-clients.md), [WebRTC client](05-webrtc-clients.md) or by sending [MPEG-TS packets](12-mpeg-ts.md) or [RTP packets](13-rtp.md). The recommended way is acting as a RTSP client. ## GStreamer as a RTSP client @@ -58,7 +58,7 @@ audiotestsrc ! audioconvert ! avenc_aac ! mux. ## GStreamer and RTP over UDP -In _MediaMTX_ configuration, add a path with `source: udp+rtp://238.0.0.1:1234` and a valid `rtpSDP` (read [RTP](12-rtp.md)). Then: +In _MediaMTX_ configuration, add a path with `source: udp+rtp://238.0.0.1:1234` and a valid `rtpSDP` (read [RTP](13-rtp.md)). Then: ```sh gst-launch-1.0 -v \ diff --git a/docs/3-publish/18-obs-studio.md b/docs/3-publish/19-obs-studio.md similarity index 98% rename from docs/3-publish/18-obs-studio.md rename to docs/3-publish/19-obs-studio.md index be39c835..4cd0a14d 100644 --- a/docs/3-publish/18-obs-studio.md +++ b/docs/3-publish/19-obs-studio.md @@ -1,6 +1,6 @@ # OBS Studio -OBS Studio can publish streams to the server by acting as a [RTMP client](08-rtmp-clients.md) or a [WebRTC client](04-webrtc-clients.md) protocols. The recommended way is acting as a RTMP client. +OBS Studio can publish streams to the server by acting as a [RTMP client](09-rtmp-clients.md) or a [WebRTC client](05-webrtc-clients.md) protocols. The recommended way is acting as a RTMP client. ## OBS Studio as a RTMP client @@ -177,7 +177,7 @@ OBS Studio requires _MediaMTX_ to use a TLS certificate signed by a public certi ### Standard -Recent versions of OBS Studio can also publish streams to the server with the [WebRTC / WHIP protocol](04-webrtc-clients.md) Use the following parameters: +Recent versions of OBS Studio can also publish streams to the server with the [WebRTC / WHIP protocol](05-webrtc-clients.md) Use the following parameters: - Service: `WHIP` - Server: `http://localhost:8889/mystream/whip` diff --git a/docs/3-publish/19-python-opencv.md b/docs/3-publish/20-python-opencv.md similarity index 96% rename from docs/3-publish/19-python-opencv.md rename to docs/3-publish/20-python-opencv.md index bb5f2b5e..e04adc0d 100644 --- a/docs/3-publish/19-python-opencv.md +++ b/docs/3-publish/20-python-opencv.md @@ -1,6 +1,6 @@ # Python and OpenCV -Python-based software can publish streams to the server with the OpenCV library and its GStreamer plugin, acting as a [RTSP client](06-rtsp-clients.md). OpenCV must be compiled with support for GStreamer, by following this procedure: +Python-based software can publish streams to the server with the OpenCV library and its GStreamer plugin, acting as a [RTSP client](07-rtsp-clients.md). OpenCV must be compiled with support for GStreamer, by following this procedure: ```sh sudo apt install -y libgstreamer1.0-dev libgstreamer-plugins-base1.0-dev gstreamer1.0-plugins-ugly gstreamer1.0-rtsp python3-dev python3-numpy diff --git a/docs/3-publish/20-golang.md b/docs/3-publish/21-golang.md similarity index 100% rename from docs/3-publish/20-golang.md rename to docs/3-publish/21-golang.md diff --git a/docs/3-publish/21-unity.md b/docs/3-publish/22-unity.md similarity index 98% rename from docs/3-publish/21-unity.md rename to docs/3-publish/22-unity.md index 43bb77ce..45f32f60 100644 --- a/docs/3-publish/21-unity.md +++ b/docs/3-publish/22-unity.md @@ -1,6 +1,6 @@ # Unity -Software written with the Unity Engine can publish a stream to the server by acting as a [WebRTC client](04-webrtc-clients.md). +Software written with the Unity Engine can publish a stream to the server by acting as a [WebRTC client](05-webrtc-clients.md). Create a new Unity project or open an existing one. diff --git a/internal/conf/conf_test.go b/internal/conf/conf_test.go index 672f84c5..3232cc35 100644 --- a/internal/conf/conf_test.go +++ b/internal/conf/conf_test.go @@ -56,6 +56,7 @@ func TestConfFromFile(t *testing.T) { RecordSegmentDuration: 3600000000000, RecordDeleteAfter: 86400000000000, RTSPUDPSourcePortRange: []uint{32768, 60999}, + MoQTransport: MoQTransportQUIC, WHEPSTUNGatherTimeout: 5 * Duration(time.Second), WHEPHandshakeTimeout: 10 * Duration(time.Second), WHEPTrackGatherTimeout: 2 * Duration(time.Second), @@ -132,6 +133,10 @@ func TestConfFromFile(t *testing.T) { name: "wheps with placeholders", source: "wheps://$G1:$G2/$G3", }, + { + name: "moqt with placeholders", + source: "moqt://$G1:$G2/$G3", + }, { name: "udp with placeholders", source: "udp://$G1:$G2", diff --git a/internal/conf/moq_transport.go b/internal/conf/moq_transport.go new file mode 100644 index 00000000..5bc8b039 --- /dev/null +++ b/internal/conf/moq_transport.go @@ -0,0 +1,38 @@ +package conf + +import ( + "fmt" + + "github.com/bluenviron/mediamtx/internal/conf/jsonwrapper" +) + +// MoQTransport is the moqTransport parameter. +type MoQTransport string + +// supported values. +const ( + MoQTransportQUIC MoQTransport = "quic" + MoQTransportWebTransport MoQTransport = "webtransport" +) + +// UnmarshalJSON implements json.Unmarshaler. +func (d *MoQTransport) UnmarshalJSON(b []byte) error { + type alias MoQTransport + if err := jsonwrapper.Unmarshal(b, (*alias)(d)); err != nil { + return err + } + + switch *d { + case MoQTransportQUIC, MoQTransportWebTransport: + + default: + return fmt.Errorf("invalid MoQ transport '%s'", *d) + } + + return nil +} + +// UnmarshalEnv implements env.Unmarshaler. +func (d *MoQTransport) UnmarshalEnv(_ string, v string) error { + return d.UnmarshalJSON([]byte(`"` + v + `"`)) +} diff --git a/internal/conf/path.go b/internal/conf/path.go index dbbe76bf..28ed8792 100644 --- a/internal/conf/path.go +++ b/internal/conf/path.go @@ -272,6 +272,9 @@ type Path struct { RTPSDP string `json:"rtpSDP"` RTPUDPReadBufferSize *uint `json:"rtpUDPReadBufferSize,omitempty" deprecated:"true"` + // MoQ source + MoQTransport MoQTransport `json:"moqTransport"` + // WHEP source WHEPBearerToken string `json:"whepBearerToken"` WHEPSTUNGatherTimeout Duration `json:"whepSTUNGatherTimeout"` @@ -381,6 +384,9 @@ func (pconf *Path) setDefaults() { // RTSP source pconf.RTSPUDPSourcePortRange = []uint{32768, 60999} + // MoQ source + pconf.MoQTransport = MoQTransportQUIC + // WHEP source pconf.WHEPSTUNGatherTimeout = Duration(5 * time.Second) pconf.WHEPHandshakeTimeout = Duration(10 * time.Second) @@ -567,6 +573,12 @@ func (pconf *Path) validate( return err } + case strings.HasPrefix(pconf.Source, "moqt://"): + _, err := validateURL(pconf.Source) + if err != nil { + return err + } + case strings.HasPrefix(pconf.Source, "whep://") || strings.HasPrefix(pconf.Source, "wheps://"): _, err := validateURL(pconf.Source) diff --git a/internal/defs/api_path.go b/internal/defs/api_path.go index 33d8c377..7ef405ea 100644 --- a/internal/defs/api_path.go +++ b/internal/defs/api_path.go @@ -34,6 +34,7 @@ const ( APIPathSourceTypeRTPSource APIPathSourceType = "rtpSource" APIPathSourceTypeWebRTCSession APIPathSourceType = "webRTCSession" APIPathSourceTypeWebRTCSource APIPathSourceType = "webRTCSource" + APIPathSourceTypeMoQSource APIPathSourceType = "moqSource" APIPathSourceTypeMoQSession APIPathSourceType = "moqSession" ) diff --git a/internal/staticsources/handler.go b/internal/staticsources/handler.go index dc97f800..fd7b6daf 100644 --- a/internal/staticsources/handler.go +++ b/internal/staticsources/handler.go @@ -12,6 +12,7 @@ import ( "github.com/bluenviron/mediamtx/internal/defs" "github.com/bluenviron/mediamtx/internal/logger" sshls "github.com/bluenviron/mediamtx/internal/staticsources/hls" + ssmoq "github.com/bluenviron/mediamtx/internal/staticsources/moq" ssmpegts "github.com/bluenviron/mediamtx/internal/staticsources/mpegts" ssrpicamera "github.com/bluenviron/mediamtx/internal/staticsources/rpicamera" ssrtmp "github.com/bluenviron/mediamtx/internal/staticsources/rtmp" @@ -142,6 +143,12 @@ func (s *Handler) Initialize() { Parent: s, } + case strings.HasPrefix(s.Conf.Source, "moqt://"): + s.instance = &ssmoq.Source{ + ReadTimeout: s.ReadTimeout, + Parent: s, + } + case strings.HasPrefix(s.Conf.Source, "whep://") || strings.HasPrefix(s.Conf.Source, "wheps://"): s.instance = &sswebrtc.Source{ diff --git a/internal/staticsources/handler_test.go b/internal/staticsources/handler_test.go index 3f606789..34021963 100644 --- a/internal/staticsources/handler_test.go +++ b/internal/staticsources/handler_test.go @@ -91,6 +91,13 @@ func TestResolveSource(t *testing.T) { query: "", expected: "whep://example.com:443/mystream", }, + { + name: "moq protocol", + source: "moqt://$G1:$G2/$G3", + matches: []string{"srv_example.com_443_mystream", "example.com", "443", "mystream"}, + query: "", + expected: "moqt://example.com:443/mystream", + }, { name: "mpeg ts udp", source: "udp://$G1:$G2", diff --git a/internal/staticsources/moq/source.go b/internal/staticsources/moq/source.go new file mode 100644 index 00000000..1f468967 --- /dev/null +++ b/internal/staticsources/moq/source.go @@ -0,0 +1,539 @@ +// Package moq contains the MoQ static source. +package moq + +import ( + "bufio" + "context" + "crypto/tls" + "encoding/base64" + "encoding/json" + "fmt" + "io" + "net/url" + "strings" + + "github.com/bluenviron/gortsplib/v5/pkg/description" + "github.com/quic-go/quic-go" + "github.com/quic-go/webtransport-go" + + "github.com/bluenviron/mediamtx/internal/conf" + "github.com/bluenviron/mediamtx/internal/defs" + "github.com/bluenviron/mediamtx/internal/logger" + protomoq "github.com/bluenviron/mediamtx/internal/protocols/moq" + "github.com/bluenviron/mediamtx/internal/protocols/moq/catalog" + "github.com/bluenviron/mediamtx/internal/protocols/moq/controlmessage" + "github.com/bluenviron/mediamtx/internal/protocols/moq/parameter" + "github.com/bluenviron/mediamtx/internal/protocols/moq/reorderer" + "github.com/bluenviron/mediamtx/internal/protocols/moq/subgroup" + ptls "github.com/bluenviron/mediamtx/internal/protocols/tls" + "github.com/bluenviron/mediamtx/internal/stream" +) + +const maxReorderedSubGroups = 50 + +type parent interface { + logger.Writer + SetReady(req defs.PathSourceStaticSetReadyReq) defs.PathSourceStaticSetReadyRes + SetNotReady(req defs.PathSourceStaticSetNotReadyReq) +} + +type conn interface { + OpenUniStreamSync(ctx context.Context) (io.WriteCloser, error) + AcceptUniStream(ctx context.Context) (io.Reader, error) + OpenStreamSync(ctx context.Context) (io.ReadWriteCloser, error) + CloseWithError(code uint64, msg string) error + Close() error +} + +type connQUIC struct { + conn *quic.Conn +} + +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) CloseWithError(code uint64, msg string) error { + return c.conn.CloseWithError(quic.ApplicationErrorCode(code), msg) +} + +func (c *connQUIC) Close() error { + return c.conn.CloseWithError(0, "") +} + +type connWebTransport struct { + session *webtransport.Session + transport *webtransport.Transport + responseBody io.Closer +} + +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) CloseWithError(code uint64, msg string) error { + return c.session.CloseWithError(webtransport.SessionErrorCode(code), msg) +} + +func (c *connWebTransport) Close() error { + c.session.CloseWithError(0, "") //nolint:errcheck + if c.responseBody != nil { + c.responseBody.Close() //nolint:errcheck + } + if c.transport != nil { + return c.transport.Close() + } + return nil +} + +type inboundTrack struct { + onSubGroup func(sg *subgroup.SubGroup) error + parent logger.Writer + + reorderer *reorderer.Reorderer +} + +func (t *inboundTrack) initialize() { + t.reorderer = &reorderer.Reorderer{ + MaxReordered: maxReorderedSubGroups, + Parent: t.parent, + } + t.reorderer.Initialize() +} + +func (t *inboundTrack) push(sg *subgroup.SubGroup) error { + sgs, err := t.reorderer.Push(sg) + if err != nil { + return err + } + + for _, sg := range sgs { + err = t.onSubGroup(sg) + if err != nil { + return err + } + } + + return nil +} + +func encodeAuthorization(user *url.Userinfo) parameter.Parameters { + if user == nil { + return nil + } + + username := user.Username() + password, ok := user.Password() + if username == "" || !ok { + return nil + } + + credentials := base64.StdEncoding.EncodeToString([]byte(username + ":" + password)) + + return parameter.Parameters{ + ¶meter.AuthorizationToken{ + AliasType: parameter.AuthorizationTokenAliasTypeUseValue, + TokenType: 1, + TokenValue: []byte("Basic " + credentials), + }, + } +} + +func parseVersionFromWTHeader(v string) defs.APIMoQVersion { + v = strings.Trim(v, "\"") + + switch defs.APIMoQVersion(v) { + case defs.APIMoQVersionDraft19, + defs.APIMoQVersionDraft18, + defs.APIMoQVersionDraft17, + defs.APIMoQVersionDraft16: + return defs.APIMoQVersion(v) + default: + return "" + } +} + +func performSetup(ctx context.Context, c conn, version defs.APIMoQVersion, path string) error { + if version == defs.APIMoQVersionDraft16 { + setupBidi, err := c.OpenStreamSync(ctx) + if err != nil { + return err + } + defer setupBidi.Close() //nolint:errcheck + + _, err = setupBidi.Write(controlmessage.ClientSetup(controlmessage.Setup{Path: path}).Marshal()) + if err != nil { + return err + } + + msg, err := controlmessage.Read(setupBidi) + if err != nil { + return err + } + + if _, ok := msg.(*controlmessage.ServerSetup); !ok { + return fmt.Errorf("unexpected setup response: %T", msg) + } + + return nil + } + + setupStream, err := c.AcceptUniStream(ctx) + if err != nil { + return err + } + + msg, err := controlmessage.Read(setupStream) + if err != nil { + return err + } + + if _, ok := msg.(*controlmessage.Setup); !ok { + return fmt.Errorf("unexpected setup response: %T", msg) + } + + clientSetup, err := c.OpenUniStreamSync(ctx) + if err != nil { + return err + } + defer clientSetup.Close() //nolint:errcheck + + payload := controlmessage.Setup{} + if path != "" { + payload.Path = path + } + + _, err = clientSetup.Write(payload.Marshal()) + return err +} + +func subscribe( + ctx context.Context, + c conn, + requestID uint64, + trackName string, + params parameter.Parameters, +) (uint64, io.ReadWriteCloser, error) { + bidi, err := c.OpenStreamSync(ctx) + if err != nil { + return 0, nil, err + } + + _, err = bidi.Write(controlmessage.Subscribe{ + RequestID: requestID, + TrackName: trackName, + Parameters: params, + }.Marshal()) + if err != nil { + bidi.Close() //nolint:errcheck + return 0, nil, err + } + + msg, err := controlmessage.Read(bidi) + if err != nil { + bidi.Close() //nolint:errcheck + return 0, nil, err + } + + switch msg := msg.(type) { + case *controlmessage.SubscribeOk: + return msg.TrackAlias, bidi, nil + + case *controlmessage.RequestError: + bidi.Close() //nolint:errcheck + return 0, nil, fmt.Errorf("subscribe failed: %s", msg.Reason) + + default: + bidi.Close() //nolint:errcheck + return 0, nil, fmt.Errorf("unexpected subscribe response: %T", msg) + } +} + +func readSubGroup(r io.Reader) (*subgroup.SubGroup, error) { + br := bufio.NewReader(r) + firstByte, err := br.Peek(1) + if err != nil { + return nil, err + } + if (firstByte[0] & 0x90) != 0x10 { + return nil, fmt.Errorf("unexpected unidirectional stream") + } + + var sg subgroup.SubGroup + err = sg.Read(br) + if err != nil { + return nil, err + } + + return &sg, nil +} + +func dialQUIC(ctx context.Context, u *url.URL, tlsConfig *tls.Config) (conn, defs.APIMoQVersion, error) { + host := u.Host + if !strings.Contains(host, ":") { + host += ":443" + } + + cfg := &tls.Config{} + if tlsConfig != nil { + cfg = tlsConfig.Clone() + } + cfg.NextProtos = []string{ + string(defs.APIMoQVersionDraft19), + string(defs.APIMoQVersionDraft18), + string(defs.APIMoQVersionDraft17), + string(defs.APIMoQVersionDraft16), + } + if cfg.ServerName == "" { + hostname := u.Hostname() + if hostname != "" && tlsConfig == nil { + cfg.ServerName = hostname + } + } + + qconn, err := quic.DialAddr(ctx, host, cfg, &quic.Config{EnableDatagrams: true}) + if err != nil { + return nil, "", err + } + + version := defs.APIMoQVersion(qconn.ConnectionState().TLS.NegotiatedProtocol) + if version == "" { + qconn.CloseWithError(0, "") //nolint:errcheck + return nil, "", fmt.Errorf("missing negotiated MoQ version") + } + + return &connQUIC{conn: qconn}, version, nil +} + +func dialWebTransport(ctx context.Context, u *url.URL, tlsConfig *tls.Config) (conn, defs.APIMoQVersion, error) { + httpsURL := &url.URL{ + Scheme: "https", + Host: u.Host, + Path: u.Path, + RawQuery: u.RawQuery, + } + + cfg := &tls.Config{} + if tlsConfig != nil { + cfg = tlsConfig.Clone() + } + if cfg.ServerName == "" { + hostname := u.Hostname() + if hostname != "" && tlsConfig == nil { + cfg.ServerName = hostname + } + } + + transport := &webtransport.Transport{ + TLSClientConfig: cfg, + QUICConfig: &quic.Config{ + EnableDatagrams: true, + EnableStreamResetPartialDelivery: true, + }, + ApplicationProtocols: []string{ + string(defs.APIMoQVersionDraft19), + string(defs.APIMoQVersionDraft18), + string(defs.APIMoQVersionDraft17), + string(defs.APIMoQVersionDraft16), + }, + } + + res, session, err := transport.Dial(ctx, httpsURL.String(), nil) + if err != nil { + transport.Close() //nolint:errcheck + return nil, "", err + } + + version := parseVersionFromWTHeader(res.Header.Get("WT-Protocol")) + if version == "" { + res.Body.Close() //nolint:errcheck + session.CloseWithError(0, "") //nolint:errcheck + transport.Close() //nolint:errcheck + return nil, "", fmt.Errorf("missing negotiated MoQ version") + } + + return &connWebTransport{session: session, transport: transport, responseBody: res.Body}, version, nil +} + +// Source is a MoQ static source. +type Source struct { + ReadTimeout conf.Duration + Parent parent +} + +// Log implements logger.Writer. +func (s *Source) Log(level logger.Level, format string, args ...any) { + s.Parent.Log(level, "[MoQ source] "+format, args...) +} + +// Run implements StaticSource. +func (s *Source) Run(params defs.StaticSourceRunParams) error { + u, err := url.Parse(params.ResolvedSource) + if err != nil { + return err + } + + tlsConfig := ptls.MakeConfig(params.Conf.SourceFingerprint) + + var c conn + var version defs.APIMoQVersion + + switch params.Conf.MoQTransport { + case conf.MoQTransportWebTransport: + c, version, err = dialWebTransport(params.Context, u, tlsConfig) + default: + c, version, err = dialQUIC(params.Context, u, tlsConfig) + } + if err != nil { + return err + } + defer c.Close() //nolint:errcheck + + setupPath := "" + if params.Conf.MoQTransport == conf.MoQTransportQUIC { + setupPath = u.RequestURI() + if setupPath == "" { + setupPath = "/" + } + } + + err = performSetup(params.Context, c, version, setupPath) + if err != nil { + return err + } + + authParams := encodeAuthorization(u.User) + + catalogAlias, catalogBidi, err := subscribe(params.Context, c, 1, ".catalog", authParams) + if err != nil { + return err + } + go io.Copy(io.Discard, catalogBidi) //nolint:errcheck + + catalogStream, err := c.AcceptUniStream(params.Context) + if err != nil { + return err + } + + sg, err := readSubGroup(catalogStream) + if err != nil { + return err + } + if sg.Header.TrackAlias != catalogAlias { + return fmt.Errorf("unexpected catalog track alias: expected %d, got %d", catalogAlias, sg.Header.TrackAlias) + } + if len(sg.Objects) == 0 { + return fmt.Errorf("received empty catalog") + } + + var cat catalog.Catalog + err = json.Unmarshal(sg.Objects[0].Payload, &cat) + if err != nil { + return fmt.Errorf("failed to parse catalog JSON: %w", err) + } + + var subStream *stream.SubStream + medias, writeFuncs, err := protomoq.ToStream(&cat, &subStream) + if err != nil { + return err + } + + res := s.Parent.SetReady(defs.PathSourceStaticSetReadyReq{ + Desc: &description.Session{Medias: medias}, + UseRTPPackets: false, + ReplaceNTP: !params.Conf.UseAbsoluteTimestamp, + }) + if res.Err != nil { + return res.Err + } + defer s.Parent.SetNotReady(defs.PathSourceStaticSetNotReadyReq{}) + + subStream = res.SubStream + + tracks := make(map[uint64]*inboundTrack) + for i, track := range cat.Tracks { + writer := writeFuncs[uint64(i+1)] + if writer == nil { + return fmt.Errorf("missing writer for track %s", track.Name) + } + + alias, bidi, err2 := subscribe(params.Context, c, uint64(i+2), track.Name, authParams) + if err2 != nil { + return err2 + } + go io.Copy(io.Discard, bidi) //nolint:errcheck + + tr := &inboundTrack{onSubGroup: writer, parent: s} + tr.initialize() + tracks[alias] = tr + } + + readErr := make(chan error, 1) + go func() { + for { + uni, err2 := c.AcceptUniStream(params.Context) + if err2 != nil { + if params.Context.Err() != nil { + readErr <- nil + return + } + readErr <- err2 + return + } + + sg2, err2 := readSubGroup(uni) + if err2 != nil { + readErr <- err2 + return + } + + track := tracks[sg2.Header.TrackAlias] + if track == nil { + continue + } + + err2 = track.push(sg2) + if err2 != nil { + readErr <- err2 + return + } + } + }() + + for { + select { + case err = <-readErr: + return err + + case <-params.ReloadConf: + + case <-params.Context.Done(): + c.CloseWithError(0, "") //nolint:errcheck + return nil + } + } +} + +// APISourceDescribe implements StaticSource. +func (*Source) APISourceDescribe() *defs.APIPathSource { + return &defs.APIPathSource{ + Type: defs.APIPathSourceTypeMoQSource, + ID: "", + } +} diff --git a/internal/staticsources/moq/source_test.go b/internal/staticsources/moq/source_test.go new file mode 100644 index 00000000..3f6a82a3 --- /dev/null +++ b/internal/staticsources/moq/source_test.go @@ -0,0 +1,607 @@ +package moq_test + +import ( + "context" + "crypto/sha256" + "crypto/tls" + "encoding/base64" + "encoding/hex" + "encoding/json" + "fmt" + "io" + "net" + "net/http" + "strings" + "testing" + "time" + + mediacommonh264 "github.com/bluenviron/mediacommon/v2/pkg/codecs/h264" + "github.com/quic-go/quic-go" + "github.com/quic-go/webtransport-go" + "github.com/stretchr/testify/require" + + "github.com/bluenviron/mediamtx/internal/conf" + "github.com/bluenviron/mediamtx/internal/defs" + "github.com/bluenviron/mediamtx/internal/logger" + "github.com/bluenviron/mediamtx/internal/protocols/httpp3" + "github.com/bluenviron/mediamtx/internal/protocols/moq/catalog" + "github.com/bluenviron/mediamtx/internal/protocols/moq/controlmessage" + "github.com/bluenviron/mediamtx/internal/protocols/moq/parameter" + "github.com/bluenviron/mediamtx/internal/protocols/moq/property" + "github.com/bluenviron/mediamtx/internal/protocols/moq/subgroup" + ssmoq "github.com/bluenviron/mediamtx/internal/staticsources/moq" + "github.com/bluenviron/mediamtx/internal/test" + "github.com/bluenviron/mediamtx/internal/unit" +) + +const testVersion = defs.APIMoQVersionDraft19 + +type testServerConn interface { + OpenUniStreamSync(context.Context) (io.WriteCloser, error) + AcceptUniStream(context.Context) (io.Reader, error) + AcceptStream(context.Context) (io.ReadWriteCloser, error) + CloseWithError(uint64, string) error +} + +type testQUICConn struct { + conn *quic.Conn +} + +func (c *testQUICConn) OpenUniStreamSync(ctx context.Context) (io.WriteCloser, error) { + return c.conn.OpenUniStreamSync(ctx) +} + +func (c *testQUICConn) AcceptUniStream(ctx context.Context) (io.Reader, error) { + return c.conn.AcceptUniStream(ctx) +} + +func (c *testQUICConn) AcceptStream(ctx context.Context) (io.ReadWriteCloser, error) { + return c.conn.AcceptStream(ctx) +} + +func (c *testQUICConn) CloseWithError(code uint64, msg string) error { + return c.conn.CloseWithError(quic.ApplicationErrorCode(code), msg) +} + +type testWTConn struct { + session *webtransport.Session +} + +func (c *testWTConn) OpenUniStreamSync(ctx context.Context) (io.WriteCloser, error) { + return c.session.OpenUniStreamSync(ctx) +} + +func (c *testWTConn) AcceptUniStream(ctx context.Context) (io.Reader, error) { + return c.session.AcceptUniStream(ctx) +} + +func (c *testWTConn) AcceptStream(ctx context.Context) (io.ReadWriteCloser, error) { + return c.session.AcceptStream(ctx) +} + +func (c *testWTConn) CloseWithError(code uint64, msg string) error { + return c.session.CloseWithError(webtransport.SessionErrorCode(code), msg) +} + +type testMoqServer struct { + address string + expectedRequestURI string + expectedAuth string + fingerprint string + transport conf.MoQTransport + + ctx context.Context + ctxCancel func() + closeFunc func() + err chan error +} + +func newTestMoqServer( + t *testing.T, + transport conf.MoQTransport, + expectedRequestURI string, + expectedAuth string, +) *testMoqServer { + t.Helper() + + ctx, ctxCancel := context.WithCancel(context.Background()) + ts := &testMoqServer{ + address: "", + expectedRequestURI: expectedRequestURI, + expectedAuth: expectedAuth, + fingerprint: "", + transport: transport, + ctx: ctx, + ctxCancel: ctxCancel, + err: make(chan error, 1), + } + + switch transport { + case conf.MoQTransportWebTransport: + ts.initializeWebTransport(t) + + default: + ts.initializeQUIC(t) + } + + return ts +} + +func (s *testMoqServer) initializeQUIC(t *testing.T) { + t.Helper() + + cert, err := tls.X509KeyPair(test.TLSCertPub, test.TLSCertKey) + require.NoError(t, err) + + ln, err := quic.ListenAddr("127.0.0.1:0", &tls.Config{ + Certificates: []tls.Certificate{cert}, + NextProtos: []string{string(testVersion)}, + }, &quic.Config{EnableDatagrams: true}) + require.NoError(t, err) + + s.address = ln.Addr().String() + s.fingerprint = fingerprintFromRaw(t, cert.Certificate[0]) + s.closeFunc = func() { + ln.Close() //nolint:errcheck + } + + go func() { + for { + acceptedConn, acceptErr := ln.Accept(s.ctx) + if s.ctx.Err() != nil { + return + } + if acceptErr != nil { + s.fail(acceptErr) + return + } + + go s.runSession(&testQUICConn{conn: acceptedConn}, false) + } + }() +} + +func (s *testMoqServer) initializeWebTransport(t *testing.T) { + t.Helper() + + addr := freeUDPAddress(t) + + h3s := &httpp3.Server{ + Address: addr, + EnableWebTransport: true, + Parent: test.NilLogger, + } + h3s.Handler = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.RequestURI() != s.expectedRequestURI { + s.fail(fmt.Errorf("unexpected request URI: expected %s, got %s", s.expectedRequestURI, r.URL.RequestURI())) + http.Error(w, "bad request", http.StatusBadRequest) + return + } + + offered := r.Header.Get("WT-Available-Protocols") + if !strings.Contains(offered, string(testVersion)) { + s.fail(fmt.Errorf("unexpected WT-Available-Protocols header: %s", offered)) + http.Error(w, "bad request", http.StatusBadRequest) + return + } + + w.Header().Set("WT-Protocol", `"`+string(testVersion)+`"`) + + session, err := h3s.Upgrade(w, r) + if err != nil { + if s.ctx.Err() == nil { + s.fail(err) + } + return + } + + go s.runSession(&testWTConn{session: session}, true) + }) + + err := h3s.Initialize() + require.NoError(t, err) + + s.address = addr + s.fingerprint = fingerprintFromRaw(t, h3s.Certificate().Certificate[0]) + s.closeFunc = h3s.Close +} + +func (s *testMoqServer) runSession(c testServerConn, webTransport bool) { + defer c.CloseWithError(0, "") //nolint:errcheck + + err := s.performSetup(c, webTransport) + if err != nil { + s.fail(err) + return + } + + for { + bidi, acceptErr := c.AcceptStream(s.ctx) + if s.ctx.Err() != nil { + return + } + if acceptErr != nil { + if strings.Contains(acceptErr.Error(), "Application error 0x0") || + strings.Contains(acceptErr.Error(), "context canceled") { + return + } + s.fail(acceptErr) + return + } + + go s.handleSubscribe(c, bidi) + } +} + +func (s *testMoqServer) performSetup(c testServerConn, webTransport bool) error { + setupWriter, err := c.OpenUniStreamSync(s.ctx) + if err != nil { + return err + } + + _, err = setupWriter.Write(controlmessage.Setup{}.Marshal()) + setupWriter.Close() //nolint:errcheck + if err != nil { + return err + } + + setupReader, err := c.AcceptUniStream(s.ctx) + if err != nil { + if strings.Contains(err.Error(), "Application error 0x0") || + strings.Contains(err.Error(), "context canceled") { + return nil + } + return err + } + + msg, err := controlmessage.Read(setupReader) + if err != nil { + return err + } + + setup, ok := msg.(*controlmessage.Setup) + if !ok { + return fmt.Errorf("unexpected setup message: %T", msg) + } + + if webTransport { + if setup.Path != "" { + return fmt.Errorf("unexpected WebTransport setup path: %s", setup.Path) + } + } else if setup.Path != s.expectedRequestURI { + return fmt.Errorf("unexpected QUIC setup path: expected %s, got %s", s.expectedRequestURI, setup.Path) + } + + return nil +} + +func (s *testMoqServer) handleSubscribe(c testServerConn, bidi io.ReadWriteCloser) { + msg, err := controlmessage.Read(bidi) + if err != nil { + if s.ctx.Err() == nil { + s.fail(err) + } + bidi.Close() //nolint:errcheck + return + } + + sub, ok := msg.(*controlmessage.Subscribe) + if !ok { + s.fail(fmt.Errorf("unexpected control message: %T", msg)) + bidi.Close() //nolint:errcheck + return + } + + authErr := s.checkAuthorization(sub.Parameters) + if authErr != nil { + s.fail(authErr) + _, _ = bidi.Write(controlmessage.RequestError{ + Code: controlmessage.RequestErrorCodeUnauthorized, + Reason: authErr.Error(), + }.Marshal()) + bidi.Close() //nolint:errcheck + return + } + + var payload []byte + switch sub.TrackName { + case ".catalog": + payload, err = catalogPayload() + + case "0": + payload, err = mediaPayload() + + default: + _, _ = bidi.Write(controlmessage.RequestError{ + Code: controlmessage.RequestErrorCodeDoesNotExist, + Reason: "unknown track", + }.Marshal()) + bidi.Close() //nolint:errcheck + return + } + if err != nil { + s.fail(err) + bidi.Close() //nolint:errcheck + return + } + + _, err = bidi.Write(controlmessage.SubscribeOk{TrackAlias: sub.RequestID}.Marshal()) + if err != nil { + s.fail(err) + bidi.Close() //nolint:errcheck + return + } + + go io.Copy(io.Discard, bidi) //nolint:errcheck + + err = s.writeSubGroup(c, sub.TrackName, sub.RequestID, payload) + if err != nil { + s.fail(err) + } +} + +func (s *testMoqServer) checkAuthorization(params parameter.Parameters) error { + if s.expectedAuth == "" { + if len(params) != 0 { + return fmt.Errorf("unexpected authorization parameter") + } + return nil + } + + if len(params) != 1 { + return fmt.Errorf("expected exactly one authorization parameter") + } + + tok, ok := params[0].(*parameter.AuthorizationToken) + if !ok { + return fmt.Errorf("unexpected authorization parameter type: %T", params[0]) + } + if tok.AliasType != parameter.AuthorizationTokenAliasTypeUseValue { + return fmt.Errorf("unexpected authorization alias type: %d", tok.AliasType) + } + if tok.TokenType != 1 { + return fmt.Errorf("unexpected authorization token type: %d", tok.TokenType) + } + if string(tok.TokenValue) != s.expectedAuth { + return fmt.Errorf("unexpected authorization token: expected %q, got %q", s.expectedAuth, string(tok.TokenValue)) + } + + return nil +} + +func (s *testMoqServer) writeSubGroup(c testServerConn, trackName string, trackAlias uint64, payload []byte) error { + uni, err := c.OpenUniStreamSync(s.ctx) + if err != nil { + return err + } + defer uni.Close() //nolint:errcheck + + sg := &subgroup.SubGroup{ + Header: subgroup.Header{ + FirstObject: true, + TrackAlias: trackAlias, + GroupID: 0, + }, + Objects: []subgroup.Object{{ + Payload: payload, + }}, + } + + if trackName != ".catalog" { + ts := property.Timestamp(0) + sg.Header.Properties = true + sg.Objects[0].Properties = property.Properties{&ts} + } + + _, err = uni.Write(sg.Marshal()) + return err +} + +func (s *testMoqServer) fail(err error) { + select { + case s.err <- err: + default: + } +} + +func (s *testMoqServer) Close() { + s.ctxCancel() + if s.closeFunc != nil { + s.closeFunc() + } +} + +func (s *testMoqServer) Check(t *testing.T) { + t.Helper() + + select { + case err := <-s.err: + require.NoError(t, err) + default: + } +} + +func basicAuthToken(user string, pass string) string { + return "Basic " + base64.StdEncoding.EncodeToString([]byte(user+":"+pass)) +} + +func catalogPayload() ([]byte, error) { + return json.Marshal(catalog.Catalog{ + Version: 1, + Tracks: []catalog.Track{{ + Name: "0", + Packaging: "loc", + IsLive: true, + Codec: "avc3.640028", + }}, + }) +} + +func mediaPayload() ([]byte, error) { + return mediacommonh264.AVCC{test.FormatH264.SPS, test.FormatH264.PPS, {5, 1}}.Marshal() +} + +func fingerprintFromRaw(t *testing.T, raw []byte) string { + t.Helper() + + sum := sha256.Sum256(raw) + return hex.EncodeToString(sum[:]) +} + +func freeUDPAddress(t *testing.T) string { + t.Helper() + + pc, err := net.ListenPacket("udp", "127.0.0.1:0") + require.NoError(t, err) + addr := pc.LocalAddr().String() + pc.Close() //nolint:errcheck + return addr +} + +type fingerprintErrorParent struct{} + +func (*fingerprintErrorParent) Log(_ logger.Level, _ string, _ ...any) {} + +func (*fingerprintErrorParent) SetReady(_ defs.PathSourceStaticSetReadyReq) defs.PathSourceStaticSetReadyRes { + panic("should not happen") +} + +func (*fingerprintErrorParent) SetNotReady(_ defs.PathSourceStaticSetNotReadyReq) {} + +func TestSource(t *testing.T) { + for _, ca := range []struct { + name string + transport conf.MoQTransport + withAuth bool + withQuery bool + }{ + { + name: "quic", + transport: conf.MoQTransportQUIC, + }, + { + name: "quic_auth_query", + transport: conf.MoQTransportQUIC, + withAuth: true, + withQuery: true, + }, + { + name: "webtransport", + transport: conf.MoQTransportWebTransport, + }, + } { + t.Run(ca.name, func(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + expectedRequestURI := "/teststream" + if ca.withQuery { + expectedRequestURI += "?key=value" + } + + expectedAuth := "" + if ca.withAuth { + expectedAuth = basicAuthToken("myuser", "mypass") + } + + server := newTestMoqServer(t, ca.transport, expectedRequestURI, expectedAuth) + defer func() { + server.Close() + server.Check(t) + }() + + p := &test.StaticSourceParent{} + p.Initialize() + + so := &ssmoq.Source{ + ReadTimeout: conf.Duration(10 * time.Second), + Parent: p, + } + + resolved := "moqt://" + server.address + "/teststream" + if ca.withAuth { + resolved = "moqt://myuser:mypass@" + server.address + "/teststream" + } + if ca.withQuery { + resolved += "?key=value" + } + + sourceErr := make(chan error, 1) + go func() { + sourceErr <- so.Run(defs.StaticSourceRunParams{ + Context: ctx, + ResolvedSource: resolved, + Conf: &conf.Path{ + MoQTransport: ca.transport, + SourceFingerprint: server.fingerprint, + }, + ReloadConf: make(chan *conf.Path), + }) + }() + + select { + case u := <-p.Unit: + require.Equal(t, unit.PayloadH264{test.FormatH264.SPS, test.FormatH264.PPS, {5, 1}}, u.Payload) + + case err := <-sourceErr: + require.NoError(t, err) + return + + case <-ctx.Done(): + t.Fatal("timeout waiting for unit") + } + + cancel() + require.NoError(t, <-sourceErr) + p.Close() + }) + } +} + +func TestSourceWebTransportFingerprintError(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + server := newTestMoqServer(t, conf.MoQTransportWebTransport, "/teststream", "") + defer func() { + server.Close() + server.Check(t) + }() + + so := &ssmoq.Source{Parent: &fingerprintErrorParent{}} + + err := so.Run(defs.StaticSourceRunParams{ + Context: ctx, + ResolvedSource: "moqt://" + server.address + "/teststream", + Conf: &conf.Path{ + MoQTransport: conf.MoQTransportWebTransport, + SourceFingerprint: strings.Repeat("0", 64), + }, + ReloadConf: make(chan *conf.Path), + }) + require.ErrorContains(t, err, "source fingerprint does not match") +} + +func TestSourceFingerprintError(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + server := newTestMoqServer(t, conf.MoQTransportQUIC, "/teststream", "") + defer func() { + server.Close() + server.Check(t) + }() + + so := &ssmoq.Source{Parent: &fingerprintErrorParent{}} + + err := so.Run(defs.StaticSourceRunParams{ + Context: ctx, + ResolvedSource: "moqt://" + server.address + "/teststream", + Conf: &conf.Path{ + MoQTransport: conf.MoQTransportQUIC, + SourceFingerprint: strings.Repeat("0", 64), + }, + ReloadConf: make(chan *conf.Path), + }) + require.ErrorContains(t, err, "source fingerprint does not match") +} diff --git a/mediamtx.yml b/mediamtx.yml index 05603992..c92020b0 100644 --- a/mediamtx.yml +++ b/mediamtx.yml @@ -486,6 +486,7 @@ pathDefaults: # * unix+mpegts://socketPath -> the stream is pulled from MPEG-TS over a Unix socket # * udp+rtp://ip:port -> the stream is pulled from RTP over UDP, by listening on the specified address # * srt://host:port?streamid=streamid -> the stream is pulled from another SRT server / camera + # * moqt://user:pass@host:port/path -> the stream is pulled from another MoQ server # * whep://host:port/path -> the stream is pulled from another WebRTC server / camera with HTTP+WHEP # * wheps://host:port/path -> the stream is pulled from another WebRTC server / camera with HTTPS+WHEP # * redirect -> the stream is provided by another path or server @@ -630,6 +631,12 @@ pathDefaults: # session description protocol (SDP) of the RTP stream. rtpSDP: + ############################################### + # Default path settings -> MoQ source (when source is a MoQ URL) + + # Transport protocol used to pull the stream. available values are "quic", "webtransport". + moqTransport: quic + ############################################### # Default path settings -> WebRTC / WHEP source (when source is WHEP)