rtp: Add Opus RTP payloader/depayloader

Part-of: <https://gitlab.freedesktop.org/gstreamer/gst-plugins-rs/-/merge_requests/1571>

docs/plugins/gst_plugins_cache.json

@@ -7453,6 +7453,72 @@
                 },
                 "rank": "marginal"
             },
+            "rtpopusdepay2": {
+                "author": "Tim-Philipp Müller <tim centricular com>",
+                "description": "Depayload an Opus audio stream from RTP packets (RFC 7587)",
+                "hierarchy": [
+                    "GstRtpOpusDepay2",
+                    "GstRtpBaseDepay2",
+                    "GstElement",
+                    "GstObject",
+                    "GInitiallyUnowned",
+                    "GObject"
+                ],
+                "klass": "Codec/Depayloader/Network/RTP",
+                "pad-templates": {
+                    "sink": {
+                        "caps": "application/x-rtp:\n          media: audio\n  encoding-name: { (string)OPUS, (string)MULTIOPUS }\n     clock-rate: 48000\n",
+                        "direction": "sink",
+                        "presence": "always"
+                    },
+                    "src": {
+                        "caps": "audio/x-opus:\nchannel-mapping-family: [ 0, 1 ]\n",
+                        "direction": "src",
+                        "presence": "always"
+                    }
+                },
+                "rank": "marginal"
+            },
+            "rtpopuspay2": {
+                "author": "Tim-Philipp Müller <tim centricular com>",
+                "description": "Payload an Opus audio stream into RTP packets (RFC 7587)",
+                "hierarchy": [
+                    "GstRtpOpusPay2",
+                    "GstRtpBasePay2",
+                    "GstElement",
+                    "GstObject",
+                    "GInitiallyUnowned",
+                    "GObject"
+                ],
+                "klass": "Codec/Payloader/Network/RTP",
+                "pad-templates": {
+                    "sink": {
+                        "caps": "audio/x-opus:\nchannel-mapping-family: 0\naudio/x-opus:\nchannel-mapping-family: 0\n       channels: [ 1, 2 ]\naudio/x-opus:\nchannel-mapping-family: 1\n       channels: [ 3, 255 ]\n",
+                        "direction": "sink",
+                        "presence": "always"
+                    },
+                    "src": {
+                        "caps": "application/x-rtp:\n          media: audio\n  encoding-name: { (string)OPUS, (string)MULTIOPUS }\n     clock-rate: 48000\n",
+                        "direction": "src",
+                        "presence": "always"
+                    }
+                },
+                "properties": {
+                    "dtx": {
+                        "blurb": "Do not send out empty packets for transmission (requires opusenc dtx=true)",
+                        "conditionally-available": false,
+                        "construct": false,
+                        "construct-only": false,
+                        "controllable": false,
+                        "default": "false",
+                        "mutable": "playing",
+                        "readable": true,
+                        "type": "gboolean",
+                        "writable": true
+                    }
+                },
+                "rank": "marginal"
+            },
             "rtppcmadepay2": {
                 "author": "Sebastian Dröge <sebastian@centricular.com>",
                 "description": "Depayload A-law from RTP packets (RFC 3551)",

net/rtp/src/lib.rs

@@ -31,6 +31,7 @@ mod jpeg;
 mod mp2t;
 mod mp4a;
 mod mp4g;
+mod opus;
 mod pcmau;
 mod vp8;
 mod vp9;
@@ -68,6 +69,9 @@ fn plugin_init(plugin: &gst::Plugin) -> Result<(), glib::BoolError> {
     mp4g::depay::register(plugin)?;
     mp4g::pay::register(plugin)?;
 
+    opus::depay::register(plugin)?;
+    opus::pay::register(plugin)?;
+
     pcmau::depay::register(plugin)?;
     pcmau::pay::register(plugin)?;
 

net/rtp/src/opus/depay/imp.rs

@@ -0,0 +1,302 @@
+// GStreamer RTP Opus Depayloader
+//
+// Copyright (C) 2023 Tim-Philipp Müller <tim centricular com>
+//
+// This Source Code Form is subject to the terms of the Mozilla Public License, v2.0.
+// If a copy of the MPL was not distributed with this file, You can obtain one at
+// <https://mozilla.org/MPL/2.0/>.
+//
+// SPDX-License-Identifier: MPL-2.0
+
+/**
+ * SECTION:element-rtpopusdepay2
+ * @see_also: rtpopuspay2, rtpopuspay, rtpopusdepay, opusdec, opusenc
+ *
+ * Extracts an Opus audio stream from RTP packets as per [RFC 7587][rfc-7587] or libwebrtc's
+ * multiopus extension.
+ *
+ * [rfc-7587]: https://www.rfc-editor.org/rfc/rfc7587.html
+ *
+ * ## Example pipeline
+ *
+ * |[
+ * gst-launch-1.0 udpsrc caps='application/x-rtp, media=audio, clock-rate=48000, encoding-name=OPUS, encoding-params=(string)2, sprop-stereo=(string)1, payload=96' ! rtpjitterbuffer latency=50 ! rtpopusdepay2 ! opusdec ! audioconvert ! audioresample ! autoaudiosink
+ * ]| This will depayload an incoming RTP Opus audio stream. You can use the #opusenc and
+ * #rtpopuspay2 elements to create such an RTP stream.
+ *
+ * Since: plugins-rs-0.13.0
+ */
+use gst::{glib, subclass::prelude::*};
+
+use once_cell::sync::Lazy;
+
+use crate::basedepay::{RtpBaseDepay2Ext, RtpBaseDepay2Impl};
+
+#[derive(Default)]
+pub struct RtpOpusDepay {}
+
+static CAT: Lazy<gst::DebugCategory> = Lazy::new(|| {
+    gst::DebugCategory::new(
+        "rtpopusdepay2",
+        gst::DebugColorFlags::empty(),
+        Some("RTP Opus Depayloader"),
+    )
+});
+
+#[glib::object_subclass]
+impl ObjectSubclass for RtpOpusDepay {
+    const NAME: &'static str = "GstRtpOpusDepay2";
+    type Type = super::RtpOpusDepay;
+    type ParentType = crate::basedepay::RtpBaseDepay2;
+}
+
+impl ObjectImpl for RtpOpusDepay {}
+
+impl GstObjectImpl for RtpOpusDepay {}
+
+impl ElementImpl for RtpOpusDepay {
+    fn metadata() -> Option<&'static gst::subclass::ElementMetadata> {
+        static ELEMENT_METADATA: Lazy<gst::subclass::ElementMetadata> = Lazy::new(|| {
+            gst::subclass::ElementMetadata::new(
+                "RTP Opus Depayloader",
+                "Codec/Depayloader/Network/RTP",
+                "Depayload an Opus audio stream from RTP packets (RFC 7587)",
+                "Tim-Philipp Müller <tim centricular com>",
+            )
+        });
+
+        Some(&*ELEMENT_METADATA)
+    }
+
+    fn pad_templates() -> &'static [gst::PadTemplate] {
+        static PAD_TEMPLATES: Lazy<Vec<gst::PadTemplate>> = Lazy::new(|| {
+            let sink_pad_template = gst::PadTemplate::new(
+                "sink",
+                gst::PadDirection::Sink,
+                gst::PadPresence::Always,
+                &gst::Caps::builder_full()
+                    .structure(
+                        // Note: not advertising X-GST-OPUS-DRAFT-SPITTKA-00 any longer
+                        gst::Structure::builder("application/x-rtp")
+                            .field("media", "audio")
+                            .field("encoding-name", gst::List::new(["OPUS", "MULTIOPUS"]))
+                            .field("clock-rate", 48000i32)
+                            .build(),
+                    )
+                    .build(),
+            )
+            .unwrap();
+
+            let src_pad_template = gst::PadTemplate::new(
+                "src",
+                gst::PadDirection::Src,
+                gst::PadPresence::Always,
+                &gst::Caps::builder("audio/x-opus")
+                    .field("channel-mapping-family", gst::IntRange::new(0i32, 1i32))
+                    .build(),
+            )
+            .unwrap();
+
+            vec![src_pad_template, sink_pad_template]
+        });
+
+        PAD_TEMPLATES.as_ref()
+    }
+}
+impl RtpBaseDepay2Impl for RtpOpusDepay {
+    const ALLOWED_META_TAGS: &'static [&'static str] = &["audio"];
+
+    fn set_sink_caps(&self, caps: &gst::Caps) -> bool {
+        let s = caps.structure(0).unwrap();
+
+        let encoding_name = s.get::<&str>("encoding-name").unwrap();
+
+        let res = match encoding_name {
+            "OPUS" => self.handle_sink_caps_opus(s),
+            "MULTIOPUS" => self.handle_sink_caps_multiopus(s),
+            _ => unreachable!(),
+        };
+
+        let Ok(src_caps) = res else {
+            gst::warning!(CAT, imp: self,
+                "Failed to parse {encoding_name} RTP input caps {s}: {}",
+                res.unwrap_err());
+            return false;
+        };
+
+        self.obj().set_src_caps(&src_caps);
+
+        true
+    }
+
+    // https://www.rfc-editor.org/rfc/rfc7587.html
+    //
+    fn handle_packet(
+        &self,
+        packet: &crate::basedepay::Packet,
+    ) -> Result<gst::FlowSuccess, gst::FlowError> {
+        // Parse frames in Opus packet to figure out duration
+        let duration = self.parse_opus_packet(packet.payload());
+
+        let mut outbuf = packet.payload_buffer();
+        let outbuf_ref = outbuf.get_mut().unwrap();
+
+        if let Some(duration) = duration {
+            outbuf_ref.set_duration(duration);
+        }
+
+        // Mark start of talkspurt with RESYNC flag
+        if packet.marker_bit() {
+            outbuf_ref.set_flags(gst::BufferFlags::RESYNC);
+        }
+
+        gst::trace!(CAT, imp: self, "Finishing buffer {outbuf:?}");
+
+        self.obj().queue_buffer(packet.into(), outbuf)
+    }
+}
+
+// Default is mono according to the RFC, but we still default to stereo here since there's
+// no guarantee that it will always be mono then, and the stream may switch between stereo
+// and mono at any time, so stereo is a slightly better default (even if possible more expensive
+// in terms of processing overhead down the line), and it's always safe to downmix to mono.
+const DEFAULT_CHANNELS: i32 = 2;
+
+impl RtpOpusDepay {
+    // Opus Media Type Registration:
+    // https://www.rfc-editor.org/rfc/rfc7587.html#section-6.1
+    //
+    fn handle_sink_caps_opus(&self, s: &gst::StructureRef) -> Result<gst::Caps, &'static str> {
+        let channels = s
+            .get::<&str>("sprop-stereo")
+            .ok()
+            .and_then(|params| params.trim().parse::<i32>().ok())
+            .map(|v| match v {
+                0 => 1, // mono
+                1 => 2, // stereo
+                _ => {
+                    gst::warning!(CAT, imp: self, "Unexpected sprop-stereo value {v} in input caps {s}");
+                    DEFAULT_CHANNELS
+                }
+            })
+            .unwrap_or(DEFAULT_CHANNELS);
+
+        let rate = s
+            .get::<&str>("sprop-maxcapturerate")
+            .ok()
+            .and_then(|params| params.trim().parse::<i32>().ok())
+            .filter(|&v| v > 0 && v <= 48000)
+            .unwrap_or(48000);
+
+        let src_caps = gst::Caps::builder("audio/x-opus")
+            .field("channel-mapping-family", 0i32)
+            .field("channels", channels)
+            .field("rate", rate)
+            .build();
+
+        Ok(src_caps)
+    }
+
+    // MULTIOPUS mapping is a Google libwebrtc concoction, see
+    // https://webrtc-review.googlesource.com/c/src/+/129768
+    //
+    fn handle_sink_caps_multiopus(&self, s: &gst::StructureRef) -> Result<gst::Caps, &'static str> {
+        let channels = s
+            .get::<&str>("encoding-params")
+            .map_err(|_| "Missing 'encoding-params' field")?
+            .trim()
+            .parse::<i32>()
+            .ok()
+            .filter(|&v| v > 0 && v <= 255)
+            .ok_or("Invalid 'encoding-params' field")?;
+
+        let num_streams = s
+            .get::<&str>("num_streams")
+            .map_err(|_| "Missing 'num_streams' field")?
+            .trim()
+            .parse::<i32>()
+            .ok()
+            .filter(|&v| v > 0 && v <= channels)
+            .ok_or("Invalid 'num_streams' field")?;
+
+        let coupled_streams = s
+            .get::<&str>("coupled_streams")
+            .map_err(|_| "Missing 'coupled_streams' field")?
+            .trim()
+            .parse::<i32>()
+            .ok()
+            .filter(|&v| v > 0 && v <= num_streams)
+            .ok_or("Invalid 'coupled_streams' field")?;
+
+        let channel_mapping: Vec<_> = s
+            .get::<&str>("channel_mapping")
+            .map_err(|_| "Missing 'channel_mapping' field")?
+            .split(',')
+            .map(|p| p.trim().parse::<i32>())
+            .collect::<Result<Vec<_>, _>>()
+            .map_err(|_| "Invalid 'channel_mapping' field")?;
+
+        let src_caps = gst::Caps::builder("audio/x-opus")
+            .field("channel-mapping-family", 1i32)
+            .field("stream-count", num_streams)
+            .field("coupled-count", coupled_streams)
+            .field("channel-mapping", gst::Array::new(channel_mapping))
+            .field("channels", channels)
+            .field("rate", 48000i32)
+            .build();
+
+        Ok(src_caps)
+    }
+
+    // https://www.rfc-editor.org/rfc/rfc6716#section-3
+    // (Note: bit numbering in diagram has bit 0 as the highest bit and bit 7 as the lowest.)
+    //
+    fn parse_opus_packet(&self, data: &[u8]) -> Option<gst::ClockTime> {
+        if data.is_empty() {
+            return gst::ClockTime::NONE;
+        }
+
+        let toc = data[0];
+        let config = (toc >> 3) & 0x1f;
+
+        let frame_duration_usecs = match config {
+            // Silk NB / MB / WB
+            0 | 4 | 8 => 10_000,
+            1 | 5 | 9 => 20_000,
+            2 | 6 | 10 => 40_000,
+            3 | 7 | 11 => 60_000,
+            // Hybrid SWB / FB
+            12 | 14 => 10_000,
+            13 | 15 => 20_000,
+            // CELT NB / WB / SWB / FB
+            16 | 20 | 24 | 28 => 2_500,
+            17 | 21 | 25 | 29 => 5_000,
+            18 | 22 | 26 | 30 => 10_000,
+            19 | 23 | 27 | 31 => 20_000,
+            _ => unreachable!(),
+        };
+
+        let frame_duration = gst::ClockTime::from_useconds(frame_duration_usecs);
+
+        let n_frames = match toc & 0b11 {
+            0 => 1,
+            1 => 2,
+            3 => {
+                if data.len() < 2 {
+                    return gst::ClockTime::NONE;
+                }
+                data[1] & 0b0011_1111
+            }
+            _ => unreachable!(),
+        } as u64;
+
+        let duration = frame_duration * n_frames;
+
+        if duration > gst::ClockTime::from_mseconds(120) {
+            gst::warning!(CAT, imp: self, "Opus packet with frame duration {duration:?} > 120ms");
+            return gst::ClockTime::NONE;
+        }
+
+        Some(duration)
+    }
+}

net/rtp/src/opus/depay/mod.rs

@@ -0,0 +1,28 @@
+// GStreamer RTP Opus Depayloader
+//
+// Copyright (C) 2023 Tim-Philipp Müller <tim centricular com>
+//
+// This Source Code Form is subject to the terms of the Mozilla Public License, v2.0.
+// If a copy of the MPL was not distributed with this file, You can obtain one at
+// <https://mozilla.org/MPL/2.0/>.
+//
+// SPDX-License-Identifier: MPL-2.0
+
+use gst::glib;
+use gst::prelude::*;
+
+pub mod imp;
+
+glib::wrapper! {
+    pub struct RtpOpusDepay(ObjectSubclass<imp::RtpOpusDepay>)
+        @extends crate::basedepay::RtpBaseDepay2, gst::Element, gst::Object;
+}
+
+pub fn register(plugin: &gst::Plugin) -> Result<(), glib::BoolError> {
+    gst::Element::register(
+        Some(plugin),
+        "rtpopusdepay2",
+        gst::Rank::MARGINAL,
+        RtpOpusDepay::static_type(),
+    )
+}

net/rtp/src/opus/mod.rs

@@ -0,0 +1,4 @@
+// SPDX-License-Identifier: MPL-2.0
+
+pub mod depay;
+pub mod pay;

net/rtp/src/opus/pay/imp.rs

@@ -0,0 +1,420 @@
+// GStreamer RTP Opus Payloader
+//
+// Copyright (C) 2023 Tim-Philipp Müller <tim centricular com>
+//
+// This Source Code Form is subject to the terms of the Mozilla Public License, v2.0.
+// If a copy of the MPL was not distributed with this file, You can obtain one at
+// <https://mozilla.org/MPL/2.0/>.
+//
+// SPDX-License-Identifier: MPL-2.0
+
+/**
+ * SECTION:element-rtpopuspay2
+ * @see_also: rtpopusdepay2, rtpopuspay, rtpopusdepay, opusdec, opusenc
+ *
+ * Payloads an Opus audio stream into RTP packets as per [RFC 7587][rfc-7587] or
+ * [libwebrtc's multiopus extension][libwebrtc-multiopus].
+ *
+ * The multi-channel extension adds extra fields to the output caps and the SDP in line with
+ * what libwebrtc expects, e.g.
+ * |[
+ *  a=rtpmap:96 multiopus/48000/6
+ *  a=fmtp:96 num_streams=4;coupled_streams=2;channel_mapping=0,4,1,2,3,5
+ * ]|
+ * for 5.1 surround sound audio.
+ *
+ * [rfc-7587]: https://www.rfc-editor.org/rfc/rfc7587.html
+ * [libwebrtc-multiopus]: https://webrtc-review.googlesource.com/c/src/+/129768
+ *
+ * ## Example pipeline
+ *
+ * |[
+ * gst-launch-1.0 audiotestsrc wave=ticks ! audio/x-raw,channels=2 ! opusenc ! rtpopuspay2 ! udpsink host=127.0.0.1 port=5004
+ * ]| This will encode and audio test signal as Opus audio and payload it as RTP and send it out
+ * over UDP to localhost port 5004.
+ *
+ * Since: plugins-rs-0.13.0
+ */
+use atomic_refcell::AtomicRefCell;
+
+use gst::{glib, prelude::*, subclass::prelude::*};
+
+use once_cell::sync::Lazy;
+
+use crate::basepay::{RtpBasePay2Ext, RtpBasePay2Impl, RtpBasePay2ImplExt};
+
+use std::sync::atomic::AtomicBool;
+
+struct State {
+    marker_pending: bool,
+}
+
+impl Default for State {
+    fn default() -> Self {
+        State {
+            marker_pending: true,
+        }
+    }
+}
+
+impl State {
+    fn marker_pending(&mut self) -> bool {
+        std::mem::replace(&mut self.marker_pending, false)
+    }
+}
+
+const DEFAULT_DTX: bool = false;
+
+#[derive(Default)]
+struct Settings {
+    dtx: AtomicBool,
+}
+
+#[derive(Default)]
+pub struct RtpOpusPay {
+    // Streaming state
+    state: AtomicRefCell<State>,
+
+    // Settings
+    settings: Settings,
+}
+
+static CAT: Lazy<gst::DebugCategory> = Lazy::new(|| {
+    gst::DebugCategory::new(
+        "rtpopuspay2",
+        gst::DebugColorFlags::empty(),
+        Some("RTP Opus Payloader"),
+    )
+});
+
+#[glib::object_subclass]
+impl ObjectSubclass for RtpOpusPay {
+    const NAME: &'static str = "GstRtpOpusPay2";
+    type Type = super::RtpOpusPay;
+    type ParentType = crate::basepay::RtpBasePay2;
+}
+
+impl ObjectImpl for RtpOpusPay {
+    fn properties() -> &'static [glib::ParamSpec] {
+        static PROPERTIES: Lazy<Vec<glib::ParamSpec>> = Lazy::new(|| {
+            vec![glib::ParamSpecBoolean::builder("dtx")
+                .nick("Discontinuous Transmission")
+                .blurb("Do not send out empty packets for transmission (requires opusenc dtx=true)")
+                .default_value(DEFAULT_DTX)
+                .mutable_playing()
+                .build()]
+        });
+
+        PROPERTIES.as_ref()
+    }
+    fn set_property(&self, _id: usize, value: &glib::Value, pspec: &glib::ParamSpec) {
+        match pspec.name() {
+            "dtx" => self.settings.dtx.store(
+                value.get().expect("type checked upstream"),
+                std::sync::atomic::Ordering::Relaxed,
+            ),
+            name => unimplemented!("Property '{name}'"),
+        };
+    }
+
+    fn property(&self, _id: usize, pspec: &glib::ParamSpec) -> glib::Value {
+        match pspec.name() {
+            "dtx" => self
+                .settings
+                .dtx
+                .load(std::sync::atomic::Ordering::Relaxed)
+                .to_value(),
+            name => unimplemented!("Property '{name}'"),
+        }
+    }
+}
+
+impl GstObjectImpl for RtpOpusPay {}
+
+impl ElementImpl for RtpOpusPay {
+    fn metadata() -> Option<&'static gst::subclass::ElementMetadata> {
+        static ELEMENT_METADATA: Lazy<gst::subclass::ElementMetadata> = Lazy::new(|| {
+            gst::subclass::ElementMetadata::new(
+                "RTP Opus Payloader",
+                "Codec/Payloader/Network/RTP",
+                "Payload an Opus audio stream into RTP packets (RFC 7587)",
+                "Tim-Philipp Müller <tim centricular com>",
+            )
+        });
+
+        Some(&*ELEMENT_METADATA)
+    }
+
+    fn pad_templates() -> &'static [gst::PadTemplate] {
+        static PAD_TEMPLATES: Lazy<Vec<gst::PadTemplate>> = Lazy::new(|| {
+            let sink_pad_template = gst::PadTemplate::new(
+                "sink",
+                gst::PadDirection::Sink,
+                gst::PadPresence::Always,
+                &gst::Caps::builder_full()
+                    .structure(
+                        gst::Structure::builder("audio/x-opus")
+                            .field("channel-mapping-family", 0i32)
+                            .build(),
+                    )
+                    .structure(
+                        gst::Structure::builder("audio/x-opus")
+                            .field("channel-mapping-family", 0i32)
+                            .field("channels", gst::IntRange::new(1i32, 2i32))
+                            .build(),
+                    )
+                    .structure(
+                        gst::Structure::builder("audio/x-opus")
+                            .field("channel-mapping-family", 1i32)
+                            .field("channels", gst::IntRange::new(3i32, 255i32))
+                            .build(),
+                    )
+                    .build(),
+            )
+            .unwrap();
+
+            let src_pad_template = gst::PadTemplate::new(
+                "src",
+                gst::PadDirection::Src,
+                gst::PadPresence::Always,
+                &gst::Caps::builder_full()
+                    .structure(
+                        gst::Structure::builder("application/x-rtp")
+                            .field("media", "audio")
+                            .field("encoding-name", gst::List::new(["OPUS", "MULTIOPUS"]))
+                            .field("clock-rate", 48000i32)
+                            .build(),
+                    )
+                    .build(),
+            )
+            .unwrap();
+
+            vec![src_pad_template, sink_pad_template]
+        });
+
+        PAD_TEMPLATES.as_ref()
+    }
+}
+impl RtpBasePay2Impl for RtpOpusPay {
+    const ALLOWED_META_TAGS: &'static [&'static str] = &["audio"];
+
+    fn set_sink_caps(&self, caps: &gst::Caps) -> bool {
+        let mut src_caps = gst::Caps::builder("application/x-rtp")
+            .field("media", "audio")
+            .field("clock-rate", 48000i32);
+
+        let s = caps.structure(0).unwrap();
+
+        let channels_field = s.get::<i32>("channels").ok();
+        let rate_field = s.get::<i32>("rate").ok();
+
+        let channel_mapping_family = s.get::<i32>("channel-mapping-family").unwrap();
+
+        let encoding_name = match channel_mapping_family {
+            // Normal Opus, mono or stereo
+            0 => {
+                if channels_field == Some(1) {
+                    src_caps = src_caps.field("sprop-stereo", "0");
+                } else {
+                    src_caps = src_caps.field("sprop-stereo", "1");
+                };
+
+                "OPUS"
+            }
+
+            // MULTIOPUS mapping is a Google libwebrtc concoction, see
+            // https://webrtc-review.googlesource.com/c/src/+/129768
+            //
+            // Stereo and Mono must always be payloaded using the normal OPUS mapping,
+            // so this is only for multi-channel Opus.
+            1 => {
+                if let Ok(stream_count) = s.get::<i32>("stream-count") {
+                    src_caps = src_caps.field("num_streams", stream_count.to_string());
+                }
+
+                if let Ok(coupled_count) = s.get::<i32>("coupled-count") {
+                    src_caps = src_caps.field("coupled_streams", coupled_count.to_string());
+                }
+
+                if let Ok(channel_mapping) = s.get::<gst::ArrayRef>("channel-mapping") {
+                    let comma_separated_channel_nums = {
+                        let res = channel_mapping
+                            .iter()
+                            .map(|v| v.get::<i32>().map(|i| i.to_string()))
+                            .collect::<Result<Vec<_>, _>>();
+
+                        // Can't use .collect().map_err()? because it doesn't work for funcs with bool returns
+                        match res {
+                            Err(_) => {
+                                gst::error!(CAT, imp: self, "Invalid 'channel-mapping' field types");
+                                return false;
+                            }
+                            Ok(num_strings) => num_strings.join(","),
+                        }
+                    };
+                    src_caps = src_caps.field("channel_mapping", comma_separated_channel_nums);
+                }
+
+                "MULTIOPUS"
+            }
+
+            _ => unreachable!(),
+        };
+
+        let channels = channels_field.unwrap_or(2);
+
+        src_caps = src_caps
+            .field("encoding-name", encoding_name)
+            .field("encoding-params", channels.to_string());
+
+        if let Some(rate) = rate_field {
+            src_caps = src_caps.field("sprop-maxcapturerate", rate.to_string());
+        }
+
+        self.obj().set_src_caps(&src_caps.build());
+
+        true
+    }
+
+    // https://www.rfc-editor.org/rfc/rfc7587.html#section-4.2
+    //
+    // We just payload whatever the Opus encoder gives us, ptime constraints and
+    // such will have to be configured on the encoder side.
+    //
+    fn handle_buffer(
+        &self,
+        buffer: &gst::Buffer,
+        id: u64,
+    ) -> Result<gst::FlowSuccess, gst::FlowError> {
+        let mut state = self.state.borrow_mut();
+
+        let map = buffer.map_readable().map_err(|_| {
+            gst::error!(CAT, imp: self, "Can't map buffer readable");
+            gst::FlowError::Error
+        })?;
+
+        let data = map.as_slice();
+
+        let dtx = self.settings.dtx.load(std::sync::atomic::Ordering::Relaxed);
+
+        // Don't output DTX packets if discontinuous transmission was enabled (in encoder and here)
+        // (Although seeing that it's opt-in in the encoder already one wonders whether we
+        // shouldn't just do it automatically here)
+        //
+        // Even in DTX mode there will still be a non-DTX packet going through every 400ms.
+        if dtx && data.len() <= 2 {
+            gst::log!(CAT, imp: self, "Not sending out empty DTX packet {:?}", buffer);
+            // The first non-DTX packet will be the start of a talkspurt
+            state.marker_pending = true;
+            self.obj().drop_buffers(..=id);
+            return Ok(gst::FlowSuccess::Ok);
+        }
+
+        let marker_pending = state.marker_pending();
+
+        self.obj().queue_packet(
+            id.into(),
+            rtp_types::RtpPacketBuilder::new()
+                .payload(data)
+                .marker_bit(marker_pending),
+        )
+    }
+
+    #[allow(clippy::single_match)]
+    fn sink_query(&self, query: &mut gst::QueryRef) -> bool {
+        match query.view_mut() {
+            gst::QueryViewMut::Caps(query) => {
+                let src_tmpl_caps = self.obj().src_pad().pad_template_caps();
+
+                let peer_caps = self.obj().src_pad().peer_query_caps(Some(&src_tmpl_caps));
+
+                if peer_caps.is_empty() {
+                    query.set_result(&peer_caps);
+                    return true;
+                }
+
+                let rtp_opus_caps = gst::Caps::builder("application/x-rtp")
+                    .field("encoding-name", "OPUS")
+                    .build();
+
+                let rtp_multiopus_caps = gst::Caps::builder("application/x-rtp")
+                    .field("encoding-name", "MULTIOPUS")
+                    .build();
+
+                // Baseline: sink pad template caps (normal opus and multi-channel opus)
+                let mut ret_caps = self.obj().sink_pad().pad_template_caps();
+
+                // Downstream doesn't support plain opus?
+                // Only multi-channel opus options left then
+                if !peer_caps.can_intersect(&rtp_opus_caps) {
+                    ret_caps = gst::Caps::builder("audio/x-opus")
+                        .field("channel-mapping-family", 1i32)
+                        .field("channels", gst::IntRange::new(3i32, 255i32))
+                        .build();
+                }
+
+                // Downstream doesn't support multi-channel opus?
+                // Only mono/stereo Opus left then
+                if !peer_caps.can_intersect(&rtp_multiopus_caps) {
+                    ret_caps = gst::Caps::builder("audio/x-opus")
+                        .field("channel-mapping-family", 0i32)
+                        .field("channels", gst::IntRange::new(1i32, 2i32))
+                        .build();
+                }
+
+                // If downstream has a preference re. mono/stereo, try to express that
+                // in the returned caps by appending a first structure with the preference
+
+                let s = ret_caps.structure(0).unwrap();
+
+                if s.get::<i32>("channel-mapping-family") == Ok(0) {
+                    let peer_s = peer_caps.structure(0).unwrap();
+                    let pref_chans = peer_s.get::<&str>("sprop-stereo")
+                        .ok()
+                        .and_then(|params| params.trim().parse::<i32>().ok())
+                        .map(|v| match v {
+                            0 => 1, // mono
+                            1 => 2, // stereo
+                            _ => {
+                                gst::warning!(CAT, imp: self, "Unexpected sprop-stereo value {v} in input caps {s}");
+                                2 // default is stereo
+                            }
+                        });
+
+                    if let Some(pref_chans) = pref_chans {
+                        let mut pref_s = peer_s.to_owned();
+                        pref_s.set("channels", pref_chans);
+                        let mut pref_caps = gst::Caps::builder_full().structure(pref_s).build();
+                        pref_caps.merge(ret_caps);
+                        ret_caps = pref_caps;
+                    }
+                }
+
+                if let Some(filter) = query.filter() {
+                    ret_caps = ret_caps.intersect_with_mode(filter, gst::CapsIntersectMode::First);
+                }
+
+                query.set_result(&ret_caps);
+
+                return true;
+            }
+            _ => (),
+        }
+
+        self.parent_sink_query(query)
+    }
+
+    fn start(&self) -> Result<(), gst::ErrorMessage> {
+        *self.state.borrow_mut() = State::default();
+
+        Ok(())
+    }
+
+    fn stop(&self) -> Result<(), gst::ErrorMessage> {
+        *self.state.borrow_mut() = State::default();
+
+        Ok(())
+    }
+}
+
+impl RtpOpusPay {}

net/rtp/src/opus/pay/mod.rs

@@ -0,0 +1,28 @@
+// GStreamer RTP Opus Payloader
+//
+// Copyright (C) 2023 Tim-Philipp Müller <tim centricular com>
+//
+// This Source Code Form is subject to the terms of the Mozilla Public License, v2.0.
+// If a copy of the MPL was not distributed with this file, You can obtain one at
+// <https://mozilla.org/MPL/2.0/>.
+//
+// SPDX-License-Identifier: MPL-2.0
+
+use gst::glib;
+use gst::prelude::*;
+
+pub mod imp;
+
+glib::wrapper! {
+    pub struct RtpOpusPay(ObjectSubclass<imp::RtpOpusPay>)
+        @extends crate::basepay::RtpBasePay2, gst::Element, gst::Object;
+}
+
+pub fn register(plugin: &gst::Plugin) -> Result<(), glib::BoolError> {
+    gst::Element::register(
+        Some(plugin),
+        "rtpopuspay2",
+        gst::Rank::MARGINAL,
+        RtpOpusPay::static_type(),
+    )
+}