fix(cast): localize UPnP URL compatibility and harden crash diagnostics

Keep http 1.5 and tolerate empty or relative service URLs in a local rupnp patch. Cover standard DLNA SOAP requests and malformed descriptions.

Avoid logger locks in fatal handlers, flush emitted log batches, and preserve CI symbols after smoke failures.

cx

.github/workflows/build-qt-linux.yml

@@ -243,6 +243,9 @@ jobs:
       # Hide the SDK and use a fresh profile/private bus; early exit (even 0)
       # fails. The helper reaps the launcher and its private process group.
       - name: Offscreen smoke (the AppDir that ships)
+        env:
+          # Explicitly opt into best-effort, non-signal-safe unwinding in CI.
+          QBZ_FATAL_BACKTRACE: "1"
         run: |
           python3 scripts/packaging/smoke_release.py ./AppDir/AppRun \
             --hide-qt "$QT_ROOT_DIR" --seconds 60 --log smoke.log
@@ -265,6 +268,7 @@ jobs:
           retention-days: ${{ startsWith(github.ref, 'refs/tags/v') && 90 || 7 }}
 
       - name: Upload unstripped binary (symbolication)
+        if: ${{ !cancelled() && hashFiles('qbz-unstripped') != '' }}
         uses: actions/upload-artifact@v7
         with:
           name: qbz-linux-${{ inputs.arch }}-unstripped

.github/workflows/test-crates.yml

@@ -15,7 +15,8 @@
 #               2048/2064-sector TOC/frame/seek equivalence and scanner import/
 #               failed-read preservation, plus the DLNA device-description
 #               tolerance suite (empty and relative SCPDURL/controlURL must
-#               still parse — the guard on qbz-cast's http =1.4.0 pin, #745).
+#               still parse with current http — local rupnp patch, #745).
+#               Normal-renderer v1/v3 transport/volume SOAP is also gated.
 #               ~5 min cold.
 #   qt-gate   — the Qt frontend: the five static QML audits, `cargo test -p
 #               qbz-qt`, the Slint-free dep-graph gate, and an OFFSCREEN BOOT of

crates/Cargo.lock

@@ -2115,9 +2115,9 @@ dependencies = [
 
 [[package]]
 name = "http"
-version = "1.4.0"
+version = "1.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e3ba2a386d7f85a81f119ad7498ebe444d2e22c2af0b86b069416ace48b3311a"
+checksum = "918d3568bebf352712bc2ef3d46a8bcf1a75b373be6539de198e9105cbbf9ce0"
 dependencies = [
  "bytes",
  "itoa",
@@ -5241,8 +5241,6 @@ checksum = "fae8ee26b0371a29a77d2b2d6b3ae13aa81def6f9bf1b1b92a32d279a5e709b7"
 [[package]]
 name = "rupnp"
 version = "3.0.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "adade717fef50ae742ee4341495dfb1f78b2bc39e50ad8a6d485ffb81c1edbf6"
 dependencies = [
  "bytes",
  "futures-core",

crates/Cargo.toml

@@ -124,6 +124,8 @@ strsim = "0.11"
 # contract. Drop this patch once a rust_cast release carries the fix upstream.
 [patch.crates-io]
 rust_cast = { path = "vendor/rust_cast" }
+# Narrow UPnP endpoint compatibility; retain current http for every consumer.
+rupnp = { path = "../third-party/rupnp" }
 
 [profile.release]
 # Strip symbol tables from release binaries. Measured on 2026-07-02:

crates/qbz-cast/Cargo.toml

@@ -18,18 +18,9 @@ tiny_http = "0.12"
 rupnp = { version = "3", features = ["full_device_spec"] }
 futures-util = "0.3"
 
-# PINNED, do not bump without reading this (#745, 2.1.0 regression):
-# rupnp parses a device description's SCPDURL / controlURL / eventSubURL as
-# `http::uri::PathAndQuery`. http 1.5.0 rewrote that parser and now REJECTS
-# two shapes 1.4.0 accepted -- an empty element (ErrorKind::Empty, printed as
-# "empty string") and any relative URL without a leading slash
-# (PathDoesNotStartWithSlash). Both are common in old UPnP firmware, so the
-# bump made rupnp drop those renderers entirely: a KEF LSX gen 1 that worked
-# in 2.0.2 (http 1.4.0) reported "Visible Renderers: 0" in 2.1.0. The pin is
-# workspace-wide by unification and is guarded by
-# tests/dlna_description_tolerance.rs -- if that suite fails, this pin was
-# lifted. Drop it only once rupnp stops using PathAndQuery for relative URLs.
-http = "=1.4.0"
+# The workspace patches rupnp's service endpoint parser for legacy UPnP URLs
+# (#745). Keep http current; never downgrade the whole HTTP graph for a device.
+http = "1.5"
 
 # Serialization
 serde = { version = "1", features = ["derive"] }

crates/qbz-cast/tests/dlna_description_tolerance.rs

@@ -1,14 +1,14 @@
 //! Regression gate for #745 — DLNA renderers vanished in 2.1.0.
 //!
-//! A KEF LSX (gen 1) answered the SSDP M-SEARCH but was dropped with
-//! `Invalid response: empty string` while fetching its device description.
-//! The cause is not the network: `rupnp` parses `SCPDURL` / `controlURL` /
-//! `eventSubURL` as `http::uri::PathAndQuery`, and `http` 1.5.0 started
+//! A report of a missing KEF LSX (gen 1) included discovery failures with
+//! `Invalid response: empty string`; the failing device XML was not attached.
+//! A reproduced cause consistent with that error: `rupnp` parses service
+//! URLs as `http::uri::PathAndQuery`, and `http` 1.5.0 started
 //! rejecting inputs that 1.4.0 accepted — an empty element, and any relative
-//! URL without a leading slash. Both shapes are common in old UPnP firmware.
+//! URL without a leading slash. These fixtures are synthetic, not captured KEF XML.
 //!
-//! These tests serve a description locally and assert the device still parses.
-//! They fail with `http` 1.5.0 and pass with 1.4.0.
+//! The tolerance cases fail with upstream rupnp + http 1.5.0. The local
+//! rupnp patch restores them while keeping http 1.5.0 for the whole workspace.
 
 use std::net::SocketAddr;
 use std::thread::JoinHandle;
@@ -83,7 +83,7 @@ fn serve_once(body: &'static str) -> (SocketAddr, JoinHandle<()>) {
         .expect("loopback listener has an ip");
 
     let handle = std::thread::spawn(move || {
-        if let Ok(request) = server.recv() {
+        if let Ok(Some(request)) = server.recv_timeout(std::time::Duration::from_secs(5)) {
             let response = tiny_http::Response::from_string(body).with_header(
                 tiny_http::Header::from_bytes(&b"Content-Type"[..], &b"text/xml"[..])
                     .expect("static header"),
@@ -132,3 +132,39 @@ async fn description_with_relative_service_urls_still_parses() {
     assert!(has_service(&device, ":service:AVTransport:"));
     assert!(has_service(&device, ":service:RenderingControl:"));
 }
+
+#[tokio::test]
+async fn empty_http_body_is_an_xml_error_not_empty_uri() {
+    let (addr, handle) = serve_once("");
+    let result =
+        rupnp::Device::from_url(format!("http://{addr}/description.xml").parse().unwrap()).await;
+    handle.join().unwrap();
+    assert!(
+        matches!(result, Err(rupnp::Error::XmlError(_))),
+        "{result:?}"
+    );
+}
+
+#[tokio::test]
+async fn malformed_and_oversized_service_urls_remain_rejected() {
+    for invalid in [
+        "bad path".to_string(),
+        "/bad path".to_string(),
+        "x".repeat(70_000),
+    ] {
+        let body: &'static str = Box::leak(
+            EMPTY_SERVICE_URLS
+                .replace("/AVTransport/control", &invalid)
+                .into_boxed_str(),
+        );
+        let (addr, handle) = serve_once(body);
+        let result =
+            rupnp::Device::from_url(format!("http://{addr}/description.xml").parse().unwrap())
+                .await;
+        handle.join().unwrap();
+        assert!(
+            matches!(result, Err(rupnp::Error::InvalidResponse(_))),
+            "{result:?}"
+        );
+    }
+}

crates/qbz-cast/tests/dlna_wire_compatibility.rs

@@ -0,0 +1,217 @@
+//! Normal-renderer wire contract. Exercise the public QBZ connection, not just
+//! URI parsing: service versions, SOAP paths/headers/payloads and state replies.
+use qbz_cast::{DiscoveredDlnaDevice, DlnaConnection, DlnaMetadata};
+use std::sync::{
+    atomic::{AtomicBool, Ordering},
+    Arc,
+};
+use std::time::Duration;
+
+#[derive(Debug)]
+struct Call {
+    path: String,
+    action: String,
+    body: String,
+}
+struct Renderer {
+    addr: std::net::SocketAddr,
+    done: Arc<AtomicBool>,
+    worker: Option<std::thread::JoinHandle<Vec<Call>>>,
+}
+impl Renderer {
+    fn new(version: u8, empty_optional_urls: bool, relative_urls: bool) -> Self {
+        let server = tiny_http::Server::http("127.0.0.1:0").unwrap();
+        let addr = server.server_addr().to_ip().unwrap();
+        let done = Arc::new(AtomicBool::new(false));
+        let stopped = done.clone();
+        let worker = std::thread::spawn(move || {
+            let mut services = String::new();
+            for name in ["AVTransport", "RenderingControl"] {
+                let scpd = if empty_optional_urls {
+                    String::new()
+                } else {
+                    format!("/{name}/scpd.xml")
+                };
+                let event = if empty_optional_urls {
+                    String::new()
+                } else {
+                    format!("/{name}/event")
+                };
+                services.push_str(&format!("<service><serviceType>urn:schemas-upnp-org:service:{name}:{version}</serviceType><serviceId>urn:upnp-org:serviceId:{name}</serviceId><SCPDURL>{scpd}</SCPDURL><controlURL>/{name}/control?zone=main</controlURL><eventSubURL>{event}</eventSubURL></service>"));
+            }
+            if relative_urls {
+                services = services.replace(">/", ">");
+            }
+            let description = format!("<root xmlns=\"urn:schemas-upnp-org:device-1-0\"><specVersion><major>1</major><minor>0</minor></specVersion><device><deviceType>urn:schemas-upnp-org:device:MediaRenderer:1</deviceType><friendlyName>Reference renderer</friendlyName><manufacturer>Test</manufacturer><modelName>Reference</modelName><UDN>uuid:reference</UDN><serviceList>{services}</serviceList></device></root>");
+            let mut calls = Vec::new();
+            while !stopped.load(Ordering::Relaxed) {
+                let Some(mut request) = server.recv_timeout(Duration::from_millis(100)).unwrap()
+                else {
+                    continue;
+                };
+                let response = if request.method() == &tiny_http::Method::Get {
+                    assert_eq!(request.url(), "/device/description.xml");
+                    description.clone()
+                } else {
+                    assert_eq!(request.method(), &tiny_http::Method::Post);
+                    let header = request
+                        .headers()
+                        .iter()
+                        .find(|h| h.field.equiv("SOAPAction"))
+                        .unwrap()
+                        .value
+                        .as_str()
+                        .trim_matches('"')
+                        .to_string();
+                    let (service, action) = header.rsplit_once('#').unwrap();
+                    let body = match action {
+                        "GetPositionInfo" => {
+                            "<RelTime>00:01:23</RelTime><TrackDuration>00:05:00</TrackDuration>"
+                        }
+                        "GetTransportInfo" => {
+                            "<CurrentTransportState>PLAYING</CurrentTransportState>"
+                        }
+                        "GetMute" => "<CurrentMute>1</CurrentMute>",
+                        _ => "",
+                    };
+                    let response = format!("<s:Envelope xmlns:s=\"http://schemas.xmlsoap.org/soap/envelope/\"><s:Body><u:{action}Response xmlns:u=\"{service}\">{body}</u:{action}Response></s:Body></s:Envelope>");
+                    let mut body = String::new();
+                    request.as_reader().read_to_string(&mut body).unwrap();
+                    let name = if matches!(action, "SetVolume" | "SetMute" | "GetMute") {
+                        "RenderingControl"
+                    } else {
+                        "AVTransport"
+                    };
+                    assert_eq!(
+                        service,
+                        format!("urn:schemas-upnp-org:service:{name}:{version}")
+                    );
+                    assert_eq!(request.url(), format!("/{name}/control?zone=main"));
+                    assert!(
+                        body.contains(&format!("<u:{action} xmlns:u=\"{service}\">")),
+                        "{body}"
+                    );
+                    calls.push(Call {
+                        path: request.url().into(),
+                        action: action.into(),
+                        body,
+                    });
+                    response
+                };
+                request
+                    .respond(tiny_http::Response::from_string(response).with_header(
+                        tiny_http::Header::from_bytes("Content-Type", "text/xml").unwrap(),
+                    ))
+                    .unwrap();
+            }
+            calls
+        });
+        Self {
+            addr,
+            done,
+            worker: Some(worker),
+        }
+    }
+    fn finish(mut self) -> Vec<Call> {
+        self.done.store(true, Ordering::Relaxed);
+        self.worker.take().unwrap().join().unwrap()
+    }
+}
+impl Drop for Renderer {
+    fn drop(&mut self) {
+        self.done.store(true, Ordering::Relaxed);
+        if let Some(worker) = self.worker.take() {
+            let _ = worker.join();
+        }
+    }
+}
+
+async fn exercise(version: u8, empty_optional_urls: bool, relative_urls: bool) {
+    let renderer = Renderer::new(version, empty_optional_urls, relative_urls);
+    let device = DiscoveredDlnaDevice {
+        id: "uuid:reference".into(),
+        name: "Reference renderer".into(),
+        manufacturer: "Test".into(),
+        model: "Reference".into(),
+        ip: "127.0.0.1".into(),
+        url: format!("http://{}/device/description.xml", renderer.addr),
+        has_av_transport: true,
+        has_rendering_control: true,
+    };
+    let mut connection = DlnaConnection::connect(device).await.unwrap();
+    let uri = "http://127.0.0.1:54321/audio/1?token=test&quality=hires";
+    connection
+        .load_media(
+            uri,
+            &DlnaMetadata {
+                title: "Title & more".into(),
+                artist: "Artist".into(),
+                album: "Album".into(),
+                artwork_url: None,
+                duration_secs: Some(300),
+            },
+            "audio/flac",
+        )
+        .await
+        .unwrap();
+    connection.play().await.unwrap();
+    assert!(connection.get_status().is_playing);
+    let position = connection.get_position_info().await.unwrap();
+    assert_eq!(
+        (
+            position.position_secs,
+            position.duration_secs,
+            position.transport_state.as_str()
+        ),
+        (83, 300, "PLAYING")
+    );
+    connection.pause().await.unwrap();
+    assert!(!connection.get_status().is_playing);
+    connection.seek(83).await.unwrap();
+    connection.set_volume(0.42).await.unwrap();
+    connection.set_volume(0.42).await.unwrap();
+    connection.set_mute(true).await.unwrap();
+    assert!(connection.get_mute().await.unwrap());
+    connection.stop().await.unwrap();
+    assert!(connection.get_status().current_uri.is_none());
+    let calls = renderer.finish();
+    let actions: Vec<_> = calls.iter().map(|c| c.action.as_str()).collect();
+    assert_eq!(
+        actions,
+        [
+            "Stop",
+            "SetAVTransportURI",
+            "Play",
+            "GetPositionInfo",
+            "GetTransportInfo",
+            "Pause",
+            "Seek",
+            "SetVolume",
+            "SetMute",
+            "GetMute",
+            "Stop"
+        ]
+    );
+    assert!(calls[1].body.contains("&amp;quality=hires"));
+    assert!(calls[6].body.contains("<Target>00:01:23</Target>"));
+    assert!(calls[7].body.contains("<DesiredVolume>42</DesiredVolume>"));
+    assert_eq!(calls[7].path, "/RenderingControl/control?zone=main");
+}
+
+#[tokio::test]
+async fn normal_renderer_v1_preserves_transport_and_volume_wire_contract() {
+    exercise(1, false, false).await;
+}
+#[tokio::test]
+async fn normal_renderer_v3_preserves_transport_and_volume_wire_contract() {
+    exercise(3, false, false).await;
+}
+#[tokio::test]
+async fn empty_optional_urls_do_not_change_valid_control_actions() {
+    exercise(1, true, false).await;
+}
+
+#[tokio::test]
+async fn relative_urls_are_normalized_to_root_relative_control_paths() {
+    exercise(1, false, true).await;
+}

crates/qbz-log/src/fatal.rs

@@ -1,51 +1,27 @@
-//! Fatal-signal handler: turn a bare `-11` into a named stack.
+//! Best-effort fatal-signal evidence on stderr, preserving the fatal status.
 //!
-//! WHY THIS EXISTS. The packaged AppDir has an intermittent SIGSEGV during
-//! startup — inside the window between `QGuiApplication` construction and the
-//! QML engine coming up, with ~30 threads (tokio workers, the audio thread,
-//! the loudness analyzer, the visualizer producer, connectivity) already
-//! running beside it. Observed twice on 2026-09-08 on different commits: once
-//! on the `v2.1.0` tag's own `release-flatpak` attempt 1, which then passed on
-//! a plain re-run of the identical commit, and once on the #745 bugfix branch.
-//! Every one of those left exactly the same evidence: `status -11` and a log
-//! that stops mid-startup. Nothing to act on.
+//! The ordinary handler uses a stack buffer and write(2), never the logger,
+//! Rust stderr locks, allocation, TLS thread names or a stack unwinder.
+//! On Linux x86_64/aarch64 it also records the interrupted program counter.
+//! Symbolication needs the matching binary and process mappings/core.
 //!
-//! Re-running is not a fix and a user who hits this cannot re-run anything, so
-//! the process now reports its own death: signal, faulting address, thread,
-//! and a backtrace, on stderr and in the log file. The next occurrence names
-//! the frame instead of costing another round trip.
-//!
-//! SAFETY POSTURE. A crash handler must never make things worse:
-//!
-//!   * A dedicated signal stack (`sigaltstack`) so a stack-overflow SIGSEGV
-//!     still has room to run this.
-//!   * A re-entrancy latch: a fault raised *by the handler* takes the default
-//!     disposition instead of recursing.
-//!   * `alarm()` armed FIRST. Capturing a backtrace allocates and takes the
-//!     dynamic-loader lock, neither of which is async-signal-safe, so a crash
-//!     inside `malloc` can deadlock here. The alarm guarantees the process
-//!     still dies, and dies promptly, instead of turning a crash into a hang.
-//!   * The signal is re-raised with the default handler restored, so the wait
-//!     status the parent sees is unchanged (`-11` stays `-11`) and a core is
-//!     still produced where cores are enabled.
-//!
-//! Best effort by construction: if the backtrace cannot be taken we still get
-//! the signal, the address and the thread, which is already more than a bare
-//! exit status.
+//! QBZ_FATAL_BACKTRACE=1 explicitly opts into an UNSAFE diagnostic backtrace
+//! for isolated CI runs. Unwinding in a signal handler can deadlock in malloc
+//! or the loader. A ten-second alarm bounds that attempt under the ordinary
+//! SIGALRM disposition; if it fires, the exit signal is SIGALRM, not the
+//! original fault. This mode is never enabled by default for users.
+//! Evidence goes to stderr, not the ring or qbz.log. A full/broken stderr
+//! destination can still prevent output; this is not a crash-dump service.
 
 #[cfg(unix)]
 mod imp {
-    use std::backtrace::Backtrace;
-    use std::io::Write;
+    use std::fmt::Write;
     use std::sync::atomic::{AtomicBool, Ordering};
+    use std::sync::Once;
 
-    /// Set once the handler is running, so a fault inside the handler falls
-    /// through to the default disposition instead of recursing forever.
     static IN_HANDLER: AtomicBool = AtomicBool::new(false);
-
-    /// Upper bound on the whole handler. Ten seconds is far longer than a
-    /// backtrace needs and far shorter than any smoke deadline.
-    const HANDLER_DEADLINE_SECS: libc::c_uint = 10;
+    static BACKTRACE: AtomicBool = AtomicBool::new(false);
+    static INSTALL: Once = Once::new();
 
     fn signal_name(signum: libc::c_int) -> &'static str {
         match signum {
@@ -58,102 +34,146 @@ mod imp {
         }
     }
 
-    /// Write to fd 2 directly. `println!` locks stdout and can deadlock if the
-    /// crashing thread already held that lock.
-    fn emit(text: &str) {
-        let mut stderr = std::io::stderr();
-        let _ = stderr.write_all(text.as_bytes());
-        let _ = stderr.flush();
+    struct Report {
+        bytes: [u8; 256],
+        len: usize,
+    }
+
+    impl std::fmt::Write for Report {
+        fn write_str(&mut self, text: &str) -> std::fmt::Result {
+            let count = text.len().min(self.bytes.len() - self.len);
+            self.bytes[self.len..self.len + count].copy_from_slice(&text.as_bytes()[..count]);
+            self.len += count;
+            Ok(())
+        }
+    }
+
+    fn emit(bytes: &[u8]) {
+        // No stdio or logger mutex. Best effort: do not retry a failed write.
+        unsafe { libc::write(libc::STDERR_FILENO, bytes.as_ptr().cast(), bytes.len()) };
+    }
+
+    unsafe fn program_counter(context: *mut libc::c_void) -> Option<usize> {
+        if context.is_null() {
+            return None;
+        }
+        #[cfg(all(target_os = "linux", target_arch = "x86_64"))]
+        {
+            let context = &*context.cast::<libc::ucontext_t>();
+            return Some(context.uc_mcontext.gregs[libc::REG_RIP as usize] as usize);
+        }
+        #[cfg(all(target_os = "linux", target_arch = "aarch64"))]
+        {
+            let context = &*context.cast::<libc::ucontext_t>();
+            return Some(context.uc_mcontext.pc as usize);
+        }
+        #[allow(unreachable_code)]
+        None
     }
 
     extern "C" fn handle(
         signum: libc::c_int,
         info: *mut libc::siginfo_t,
-        _context: *mut libc::c_void,
+        context: *mut libc::c_void,
     ) {
         if IN_HANDLER.swap(true, Ordering::SeqCst) {
-            // Faulted while reporting a fault. Stop trying.
             restore_and_reraise(signum);
             return;
         }
-
-        // FIRST, before anything that can block: guarantee an exit.
-        // SAFETY: `alarm` is async-signal-safe and only arms a timer.
-        unsafe { libc::alarm(HANDLER_DEADLINE_SECS) };
-
-        // SAFETY: the kernel hands us a valid `siginfo_t` for these signals.
+        let backtrace = BACKTRACE.load(Ordering::Relaxed);
+        if backtrace {
+            unsafe { libc::alarm(10) };
+        }
+        // SA_SIGINFO supplies valid pointers for this invocation.
         let fault_addr = unsafe { info.as_ref() }
             .map(|info| unsafe { info.si_addr() } as usize)
             .unwrap_or(0);
-
-        emit(&format!(
-            "\n=== qbz fatal {} (code {}) at {:#x}, thread {:?} ===\n",
+        let mut report = Report {
+            bytes: [0; 256],
+            len: 0,
+        };
+        let _ = write!(
+            report,
+            "\n=== qbz fatal {} (signal {}) at {:#x}",
             signal_name(signum),
             signum,
-            fault_addr,
-            std::thread::current().name().unwrap_or("<unnamed>"),
-        ));
-        // Allocates and takes loader locks. Guarded by the alarm above.
-        emit(&format!("{}\n", Backtrace::force_capture()));
-
-        // Push whatever the file sink still holds, so the log ends at the
-        // crash instead of up to FILE_FLUSH_INTERVAL before it.
-        log::logger().flush();
+            fault_addr
+        );
+        if let Some(pc) = unsafe { program_counter(context) } {
+            let _ = write!(report, " pc={pc:#x}");
+        }
+        let _ = writeln!(report, " ===");
+        emit(&report.bytes[..report.len]);
 
+        if backtrace {
+            // Deliberately opt-in: neither capture nor formatting is signal-safe.
+            emit(format!("{}\n", std::backtrace::Backtrace::force_capture()).as_bytes());
+        }
+        // Never flush the logger here: the fault may have interrupted it while
+        // holding its non-reentrant output mutex.
         restore_and_reraise(signum);
     }
 
     fn restore_and_reraise(signum: libc::c_int) {
-        // SAFETY: restoring SIG_DFL and re-raising on the current thread is
-        // the documented way to keep the original wait status and core file.
         unsafe {
             let mut action: libc::sigaction = std::mem::zeroed();
             action.sa_sigaction = libc::SIG_DFL;
             libc::sigemptyset(&mut action.sa_mask);
             libc::sigaction(signum, &action, std::ptr::null_mut());
+            // The current signal is blocked inside its handler. It is delivered
+            // with SIG_DFL when the handler returns and restores the mask.
             libc::raise(signum);
         }
     }
 
-    /// Give the handler its own stack, so a stack-overflow SIGSEGV can run it.
     fn install_alt_stack() {
-        // Leaked on purpose: the handler may run at any point in the process
-        // lifetime, including after main's locals are gone.
-        let size = libc::SIGSTKSZ.max(64 * 1024);
-        let stack = vec![0u8; size].into_boxed_slice();
-        let stack = Box::leak(stack);
-        // SAFETY: `stack` is a live, uniquely owned, leaked allocation of
-        // exactly `size` bytes.
+        // Alternate stacks are PER THREAD. This covers the installing thread,
+        // not every worker. Reuse a sufficiently large existing alternate stack.
         unsafe {
+            let mut current: libc::stack_t = std::mem::zeroed();
+            if libc::sigaltstack(std::ptr::null(), &mut current) != 0
+                || (current.ss_flags & libc::SS_DISABLE == 0
+                    && current.ss_size >= libc::SIGSTKSZ.max(64 * 1024))
+            {
+                return;
+            }
+            let size = libc::SIGSTKSZ.max(64 * 1024);
+            let mut stack = vec![0u8; size].into_boxed_slice();
             let alt = libc::stack_t {
                 ss_sp: stack.as_mut_ptr().cast(),
                 ss_flags: 0,
                 ss_size: size,
             };
-            libc::sigaltstack(&alt, std::ptr::null_mut());
+            if libc::sigaltstack(&alt, std::ptr::null_mut()) == 0 {
+                // The kernel retains the pointer until this thread exits.
+                let _ = Box::leak(stack);
+            }
         }
     }
 
     pub fn install() {
         install_alt_stack();
-        for signum in [
-            libc::SIGSEGV,
-            libc::SIGBUS,
-            libc::SIGILL,
-            libc::SIGFPE,
-            libc::SIGABRT,
-        ] {
-            // SAFETY: a zeroed `sigaction` with SA_SIGINFO and our `extern "C"`
-            // handler is the standard installation; SA_ONSTACK pairs with the
-            // alternate stack above.
-            unsafe {
-                let mut action: libc::sigaction = std::mem::zeroed();
-                action.sa_sigaction = handle as *const () as usize;
-                action.sa_flags = libc::SA_SIGINFO | libc::SA_ONSTACK;
-                libc::sigemptyset(&mut action.sa_mask);
-                libc::sigaction(signum, &action, std::ptr::null_mut());
+        INSTALL.call_once(|| {
+            BACKTRACE.store(
+                std::env::var("QBZ_FATAL_BACKTRACE").as_deref() == Ok("1"),
+                Ordering::Relaxed,
+            );
+            for signum in [
+                libc::SIGSEGV,
+                libc::SIGBUS,
+                libc::SIGILL,
+                libc::SIGFPE,
+                libc::SIGABRT,
+            ] {
+                unsafe {
+                    let mut action: libc::sigaction = std::mem::zeroed();
+                    action.sa_sigaction = handle as *const () as usize;
+                    action.sa_flags = libc::SA_SIGINFO | libc::SA_ONSTACK;
+                    libc::sigemptyset(&mut action.sa_mask);
+                    libc::sigaction(signum, &action, std::ptr::null_mut());
+                }
             }
-        }
+        });
     }
 }
 
@@ -162,24 +182,8 @@ mod imp {
     pub fn install() {}
 }
 
-/// Install the fatal-signal reporter. Idempotent in practice (re-installing
-/// the same handler is a no-op) and a no-op off unix. Call as early in `main`
-/// as possible, right after the logger: everything that crashes later is then
-/// covered, including the whole Qt/QML startup window.
+/// Install process-wide handlers once and an alternate stack on this thread
+/// if it has no sufficiently large one. No-op off Unix. Call after logger installation.
 pub fn install() {
     imp::install();
 }
-
-#[cfg(all(test, unix))]
-mod tests {
-    /// The handler must survive being installed twice and must not disturb an
-    /// ordinary process: the interesting behaviour is only reachable by
-    /// actually faulting, which a test process must not do.
-    #[test]
-    fn installing_is_idempotent_and_harmless() {
-        super::install();
-        super::install();
-        // Still alive, still able to allocate and log.
-        log::logger().flush();
-    }
-}

crates/qbz-log/src/tee.rs

@@ -3,7 +3,7 @@
 use std::fs::File;
 use std::io::{BufWriter, Write};
 use std::sync::Mutex;
-use std::time::{Duration, Instant};
+use std::time::Instant;
 
 use log::{Log, Metadata, Record};
 
@@ -26,40 +26,6 @@ pub struct TeeLogger {
 struct Output {
     file: Option<BufWriter<File>>,
     consecutive: ConsecutiveRecords,
-    file_flush: FileFlush,
-}
-
-/// How stale the on-disk log may be. The file sink is a `BufWriter`, so
-/// without a bound a process KILLED while wedged loses up to 8 KiB of the
-/// most interesting lines — and "I had to force-kill it" is the state every
-/// hang report is filed from (#749). Flushing per line instead would turn a
-/// burst (CMAF segment progress, discovery chatter) into one write syscall
-/// per record, so the file is allowed to lag by at most this much.
-const FILE_FLUSH_INTERVAL: Duration = Duration::from_millis(200);
-
-/// When the file sink must reach the disk. Split out so the policy is
-/// testable without touching a file.
-struct FileFlush {
-    last: Instant,
-    interval: Duration,
-}
-
-impl FileFlush {
-    fn new(now: Instant, interval: Duration) -> Self {
-        Self {
-            last: now,
-            interval,
-        }
-    }
-
-    /// `urgent` — a warning/error or a repeat summary — never waits.
-    fn due(&mut self, now: Instant, urgent: bool) -> bool {
-        if urgent || now.duration_since(self.last) >= self.interval {
-            self.last = now;
-            return true;
-        }
-        false
-    }
 }
 
 impl TeeLogger {
@@ -69,7 +35,6 @@ impl TeeLogger {
             output: Mutex::new(Output {
                 file,
                 consecutive: ConsecutiveRecords::default(),
-                file_flush: FileFlush::new(Instant::now(), FILE_FLUSH_INTERVAL),
             }),
         }
     }
@@ -124,20 +89,14 @@ impl Log for TeeLogger {
         };
         let now = Instant::now();
         let [summary, first] = output.consecutive.push(line, now);
-        let summarized = summary.is_some();
         for line in [summary, first].into_iter().flatten() {
             output.write(line);
         }
-        // Compaction may no longer fill BufWriter for minutes, and a wedged
-        // process is killed rather than exiting: make periodic summaries,
-        // anything at WARN or worse, and otherwise the last
-        // FILE_FLUSH_INTERVAL of ordinary records visible to a live file tail
-        // and survivable across a SIGKILL.
-        let urgent = summarized || record.level() <= log::Level::Warn;
-        if output.file_flush.due(now, urgent) {
-            if let Some(writer) = &mut output.file {
-                let _ = writer.flush();
-            }
+        // Flush each emitted batch before returning, including the last INFO
+        // before a hang. A time check on the next record cannot bound an idle
+        // buffer. Duplicate compaction still avoids writes for repeated lines.
+        if let Some(writer) = &mut output.file {
+            let _ = writer.flush();
         }
     }
 
@@ -159,29 +118,41 @@ mod tests {
     use super::*;
     use log::Level;
 
-    /// A wedged process is force-killed, so the file sink may never lag by
-    /// more than the interval, and a warning must never sit in the buffer.
     #[test]
-    fn file_flush_is_bounded_and_urgent_records_never_wait() {
-        let start = Instant::now();
-        let mut policy = FileFlush::new(start, Duration::from_millis(200));
-
-        assert!(
-            !policy.due(start + Duration::from_millis(10), false),
-            "an ordinary record inside the window stays buffered"
-        );
-        assert!(
-            policy.due(start + Duration::from_millis(20), true),
-            "a warning or a summary flushes immediately"
-        );
-        assert!(
-            !policy.due(start + Duration::from_millis(100), false),
-            "the urgent flush restarts the window"
-        );
-        assert!(
-            policy.due(start + Duration::from_millis(220), false),
-            "an ordinary record past the window flushes"
+    fn last_info_is_visible_without_another_record_or_explicit_flush() {
+        // The real logger writes to the process-global ring. Isolate this
+        // exercise from the ring unit test, which checks exact FIFO contents.
+        const MARKER: &str = "QBZ_LOG_LAST_INFO_TEST_CHILD";
+        if std::env::var_os(MARKER).is_none() {
+            let status = std::process::Command::new(std::env::current_exe().unwrap())
+                .args([
+                    "tee::tests::last_info_is_visible_without_another_record_or_explicit_flush",
+                    "--exact",
+                ])
+                .env(MARKER, "1")
+                .status()
+                .unwrap();
+            assert!(status.success());
+            return;
+        }
+        let path = std::env::temp_dir().join(format!("qbz-log-last-info-{}", std::process::id()));
+        let file = File::create(&path).unwrap();
+        let inner = env_logger::Builder::new()
+            .filter_level(log::LevelFilter::Info)
+            .build();
+        let logger = TeeLogger::new(inner, Some(BufWriter::new(file)));
+        logger.log(
+            &Record::builder()
+                .level(Level::Info)
+                .target("test")
+                .args(format_args!("last startup message"))
+                .build(),
         );
+        assert!(std::fs::read_to_string(&path)
+            .unwrap()
+            .contains("last startup message"));
+        drop(logger);
+        std::fs::remove_file(path).unwrap();
     }
 
     #[test]

crates/qbz-log/tests/fatal_signal.rs

@@ -1,65 +1,127 @@
-//! End-to-end proof that a crash reports itself instead of vanishing as `-11`.
-//!
-//! The intermittent packaged-AppDir SIGSEGV (see `qbz-log/src/fatal.rs`) left
-//! nothing but an exit status, so every occurrence cost a re-run and taught us
-//! nothing. This test crashes a real child process on purpose and requires
-//! both halves of the contract:
-//!
-//!   1. the handler names the signal, the faulting address and the thread, and
-//!      prints a backtrace, and
-//!   2. the wait status the parent sees is STILL the signal — the reporter
-//!      must not turn a segfault into a clean exit and hide it from CI, the
-//!      smoke harness or a core dump.
-
+//! Child-process regressions for signal status, real faults and locked loggers.
 #![cfg(unix)]
 
 use std::os::unix::process::ExitStatusExt;
-use std::process::Command;
+use std::process::{Command, Stdio};
+use std::time::{Duration, Instant};
 
-/// Set in the child so it takes the crashing branch instead of recursing.
 const CHILD_MARKER: &str = "QBZ_LOG_FATAL_TEST_CHILD";
 
+struct MustNotFlush;
+impl log::Log for MustNotFlush {
+    fn enabled(&self, _: &log::Metadata<'_>) -> bool {
+        true
+    }
+    fn log(&self, _: &log::Record<'_>) {}
+    fn flush(&self) {
+        panic!("a signal handler must not call the logger");
+    }
+}
+static LOGGER: MustNotFlush = MustNotFlush;
+
+#[test]
+fn fatal_child() {
+    let Ok(mode) = std::env::var(CHILD_MARKER) else {
+        return;
+    };
+    // No core files from deliberately crashing tests.
+    unsafe {
+        let limit = libc::rlimit {
+            rlim_cur: 0,
+            rlim_max: 0,
+        };
+        libc::setrlimit(libc::RLIMIT_CORE, &limit);
+    }
+    log::set_logger(&LOGGER).unwrap();
+    qbz_log::install_fatal_signal_reporter();
+    qbz_log::install_fatal_signal_reporter();
+    if mode == "stderr-locked" {
+        let (tx, rx) = std::sync::mpsc::channel();
+        std::thread::spawn(move || {
+            let _stderr = std::io::stderr().lock();
+            tx.send(()).unwrap();
+            std::thread::sleep(Duration::from_secs(30));
+        });
+        rx.recv().unwrap();
+    }
+    if mode == "memory-fault" {
+        unsafe {
+            let page = libc::mmap(
+                std::ptr::null_mut(),
+                4096,
+                libc::PROT_NONE,
+                libc::MAP_PRIVATE | libc::MAP_ANON,
+                -1,
+                0,
+            );
+            assert_ne!(page, libc::MAP_FAILED);
+            // A real protection fault in C; no Rust null-pointer UB.
+            libc::memset(page, 1, 4096);
+        }
+    } else {
+        unsafe { libc::raise(libc::SIGSEGV) };
+    }
+    panic!("child survived fatal signal");
+}
+
+fn crash(mode: &str, backtrace: bool) -> String {
+    let mut command = Command::new(std::env::current_exe().unwrap());
+    command
+        .args(["fatal_child", "--exact", "--nocapture"])
+        .env(CHILD_MARKER, mode)
+        .env_remove("QBZ_FATAL_BACKTRACE")
+        .stdout(Stdio::null())
+        .stderr(Stdio::piped());
+    if backtrace {
+        command.env("QBZ_FATAL_BACKTRACE", "1");
+    }
+    let mut child = command.spawn().unwrap();
+    let deadline = Instant::now() + Duration::from_secs(5);
+    loop {
+        if child.try_wait().unwrap().is_some() {
+            break;
+        }
+        if Instant::now() >= deadline {
+            child.kill().unwrap();
+            let output = child.wait_with_output().unwrap();
+            panic!(
+                "handler hung ({mode}): {}",
+                String::from_utf8_lossy(&output.stderr)
+            );
+        }
+        std::thread::sleep(Duration::from_millis(10));
+    }
+    let output = child.wait_with_output().unwrap();
+    let stderr = String::from_utf8(output.stderr).unwrap();
+    assert_eq!(output.status.signal(), Some(libc::SIGSEGV), "{stderr}");
+    assert!(stderr.contains("qbz fatal SIGSEGV"), "{stderr}");
+    #[cfg(all(
+        target_os = "linux",
+        any(target_arch = "x86_64", target_arch = "aarch64")
+    ))]
+    assert!(stderr.contains("pc=0x"), "{stderr}");
+    stderr
+}
+
 #[test]
 fn a_segfault_reports_itself_and_still_dies_of_the_signal() {
-    if std::env::var_os(CHILD_MARKER).is_some() {
-        qbz_log::install_fatal_signal_reporter();
-        // `write_volatile` so the null store cannot be optimized away.
-        unsafe { std::ptr::null_mut::<u8>().write_volatile(1) };
-        unreachable!("the child must not survive the store");
-    }
-
-    let exe = std::env::current_exe().expect("test binary path");
-    let output = Command::new(exe)
-        .args([
-            "a_segfault_reports_itself_and_still_dies_of_the_signal",
-            "--exact",
-            "--nocapture",
-        ])
-        .env(CHILD_MARKER, "1")
-        .env("RUST_BACKTRACE", "1")
-        .output()
-        .expect("run the crashing child");
-
-    let stderr = String::from_utf8_lossy(&output.stderr);
-
+    let report = crash("memory-fault", false);
     assert!(
-        stderr.contains("qbz fatal SIGSEGV"),
-        "the handler did not announce the signal; stderr was:\n{stderr}"
+        !report.contains("qbz_log::fatal::imp::handle"),
+        "default must not unwind"
     );
+}
+
+#[test]
+fn handler_bypasses_locked_stderr_and_never_flushes_logger() {
+    crash("stderr-locked", false);
+}
+
+#[test]
+fn diagnostic_opt_in_captures_an_actual_frame() {
+    let report = crash("raised", true);
     assert!(
-        stderr.contains("thread"),
-        "the report must name the crashing thread; stderr was:\n{stderr}"
-    );
-    // A backtrace was attempted and produced frames, not the "disabled" note.
-    assert!(
-        stderr.contains("0x") || stderr.contains("qbz_log"),
-        "the report carries no frames; stderr was:\n{stderr}"
-    );
-    assert_eq!(
-        output.status.signal(),
-        Some(11),
-        "the child must still die of SIGSEGV so CI, the smoke harness and any \
-         core dump see the real cause; status was {:?}",
-        output.status
+        report.contains("qbz_log::fatal::imp::handle"),
+        "no actual handler frame: {report}"
     );
 }

licenses/MIT-rupnp.txt

@@ -0,0 +1,25 @@
+Copyright (c) 2016 ssdp-rs Developers
+
+Permission is hereby granted, free of charge, to any
+person obtaining a copy of this software and associated
+documentation files (the "Software"), to deal in the
+Software without restriction, including without
+limitation the rights to use, copy, modify, merge,
+publish, distribute, sublicense, and/or sell copies of
+the Software, and to permit persons to whom the Software
+is furnished to do so, subject to the following
+conditions:
+
+The above copyright notice and this permission notice
+shall be included in all copies or substantial portions
+of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF
+ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED
+TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A
+PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT
+SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
+CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
+OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR
+IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+DEALINGS IN THE SOFTWARE.

scripts/cargo-test.sh

@@ -17,7 +17,7 @@
 #     paused writes nothing, clear + VACUUM, is_play/is_skip bounds).
 #   + the DLNA device-description tolerance suite (qbz-cast): a renderer whose
 #     SCPDURL/controlURL/eventSubURL is empty or relative must still parse.
-#     It is the guard on qbz-cast's `http = "=1.4.0"` pin (#745).
+#     The local rupnp compatibility patch must preserve current http (#745).
 #   + the fatal-signal reporter check (qbz-log): a real child process is
 #     segfaulted on purpose; the handler must name the signal and the process
 #     must still die OF that signal.
@@ -137,15 +137,14 @@ cargo test --manifest-path crates/Cargo.toml -p qconnect-protocol --lib -- decod
 cargo test --manifest-path crates/Cargo.toml -p qbz-log --lib -- repeat::tests::
 
 say "gate: DLNA device-description tolerance present and green (#745)"
-# rupnp parses SCPDURL/controlURL/eventSubURL as http::uri::PathAndQuery, and
-# http 1.5.0 started rejecting empty and non-slash-prefixed values that 1.4.0
-# accepted -- which silently dropped every renderer with that firmware shape
-# (KEF LSX gen 1, "Visible Renderers: 0" in 2.1.0). qbz-cast pins http to
-# =1.4.0; this suite is the guard that notices if the pin is ever lifted.
+# The local rupnp patch accepts legacy service URLs without downgrading http.
+# Invalid/oversized URLs must remain rejected.
 n=$(cargo test --manifest-path crates/Cargo.toml -p qbz-cast --test dlna_description_tolerance -- --list 2>/dev/null \
     | grep -c ': test$' || true)
-(( n >= 2 )) || { echo "DLNA description tolerance suite has $n tests (expected >= 2)"; exit 1; }
+(( n >= 4 )) || { echo "DLNA description tolerance suite has $n tests (expected >= 4)"; exit 1; }
 cargo test --manifest-path crates/Cargo.toml -p qbz-cast --test dlna_description_tolerance
+# Normal v1/v3 renderers must retain the same transport/volume SOAP contract.
+cargo test --manifest-path crates/Cargo.toml -p qbz-cast --test dlna_wire_compatibility
 
 say "gate: a fatal signal reports itself (#749 / packaged-AppDir SIGSEGV)"
 # The intermittent startup segfault in the shipped AppDir left nothing but

third-party/rupnp/.cargo_vcs_info.json

@@ -0,0 +1,6 @@
+{
+  "git": {
+    "sha1": "00ec066e75f95a0874aaef238baebb643b354f83"
+  },
+  "path_in_vcs": ""
+}

third-party/rupnp/.gitignore

@@ -0,0 +1,5 @@
+/target
+/.idea
+/descriptions
+**/*.rs.bk
+Cargo.lock

third-party/rupnp/Cargo.toml

@@ -0,0 +1,143 @@
+# THIS FILE IS AUTOMATICALLY GENERATED BY CARGO
+#
+# When uploading crates to the registry Cargo will automatically
+# "normalize" Cargo.toml files for maximal compatibility
+# with all versions of Cargo and also rewrite `path` dependencies
+# to registry (e.g., crates.io) dependencies.
+#
+# If you are reading this file be aware that the original Cargo.toml
+# will likely look very different (and much more reasonable).
+# See Cargo.toml.orig for the original contents.
+
+[package]
+edition = "2018"
+name = "rupnp"
+version = "3.0.0"
+authors = ["Jakob Hellermann <jakob.hellermann@protonmail.com>"]
+build = false
+autolib = false
+autobins = false
+autoexamples = false
+autotests = false
+autobenches = false
+description = "An asynchronous library for finding UPnP control points, performing actions on them and reading their service descriptions."
+readme = "README.md"
+keywords = [
+    "async",
+    "upnp",
+    "networking",
+    "protocols",
+]
+categories = [
+    "network-programming",
+    "asynchronous",
+]
+license = "MIT/Apache-2.0"
+repository = "https://github.com/jakobhellermann/rupnp"
+
+[features]
+default = [
+    "subscribe",
+    "genawaiter",
+    "if-addrs",
+]
+full_device_spec = []
+subscribe = ["tokio"]
+
+[lib]
+name = "rupnp"
+path = "src/lib.rs"
+
+[[example]]
+name = "discover"
+path = "examples/discover.rs"
+
+[[example]]
+name = "discover_extra_fields"
+path = "examples/discover_extra_fields.rs"
+
+[[example]]
+name = "dump_scpd"
+path = "examples/dump_scpd.rs"
+
+[[example]]
+name = "rendering_control"
+path = "examples/rendering_control.rs"
+
+[[example]]
+name = "search_and_action"
+path = "examples/search_and_action.rs"
+
+[[example]]
+name = "subscribe"
+path = "examples/subscribe.rs"
+required-features = ["subscribe"]
+
+[dependencies.bytes]
+version = "1.0"
+
+[dependencies.futures-core]
+version = "0.3"
+
+[dependencies.futures-util]
+version = "0.3"
+default-features = false
+
+[dependencies.genawaiter]
+version = "0.99"
+features = ["futures03"]
+optional = true
+default-features = false
+
+[dependencies.http]
+version = "1.0"
+
+[dependencies.http-body-util]
+version = "0.1.3"
+
+[dependencies.hyper]
+version = "1.0"
+features = [
+    "client",
+    "http1",
+]
+
+[dependencies.hyper-util]
+version = "0.1.11"
+features = [
+    "client",
+    "client-legacy",
+    "http1",
+]
+
+[dependencies.if-addrs]
+version = "0.13"
+optional = true
+
+[dependencies.roxmltree]
+version = "0.20"
+
+[dependencies.ssdp-client]
+version = "2.0"
+
+[dependencies.tokio]
+version = "1.0"
+features = [
+    "net",
+    "io-util",
+]
+optional = true
+
+[dev-dependencies.anyhow]
+version = "1.0"
+
+[dev-dependencies.futures]
+version = "0.3"
+
+[dev-dependencies.tokio]
+version = "1.0"
+features = [
+    "macros",
+    "fs",
+    "rt-multi-thread",
+]

third-party/rupnp/Cargo.toml.orig

@@ -0,0 +1,41 @@
+[package]
+name = "rupnp"
+version = "3.0.0"
+authors = ["Jakob Hellermann <jakob.hellermann@protonmail.com>"]
+readme = "README.md"
+repository = "https://github.com/jakobhellermann/rupnp"
+description = "An asynchronous library for finding UPnP control points, performing actions on them and reading their service descriptions."
+categories = ["network-programming", "asynchronous"]
+keywords = ["async", "upnp", "networking", "protocols"]
+license = "MIT/Apache-2.0"
+edition = "2018"
+
+[features]
+default = ["subscribe", "genawaiter", "if-addrs"]
+
+full_device_spec = []
+subscribe = ["tokio"]
+
+[dependencies]
+tokio = { version = "1.0", features = ["net", "io-util"], optional = true }
+futures-core = "0.3"
+futures-util = { version = "0.3", default-features = false }
+genawaiter = { version = "0.99", default-features = false, features = ["futures03"], optional = true }
+if-addrs = { version = "0.13", optional = true }
+http = "1.0"
+ssdp-client = "2.0"
+roxmltree = "0.20"
+hyper = { version = "1.0", features = ["client", "http1"] }
+hyper-util = { version = "0.1.11", features = ["client", "client-legacy", "http1"] }
+bytes = "1.0"
+http-body-util = "0.1.3"
+
+[dev-dependencies]
+tokio = { version = "1.0", features = ["macros", "fs", "rt-multi-thread"] }
+futures = "0.3"
+anyhow = "1.0"
+
+
+[[example]]
+name = "subscribe"
+required-features = ["subscribe"]

third-party/rupnp/LICENSE-APACHE

@@ -0,0 +1,201 @@
+                              Apache License
+                        Version 2.0, January 2004
+                     http://www.apache.org/licenses/
+
+TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
+
+1. Definitions.
+
+   "License" shall mean the terms and conditions for use, reproduction,
+   and distribution as defined by Sections 1 through 9 of this document.
+
+   "Licensor" shall mean the copyright owner or entity authorized by
+   the copyright owner that is granting the License.
+
+   "Legal Entity" shall mean the union of the acting entity and all
+   other entities that control, are controlled by, or are under common
+   control with that entity. For the purposes of this definition,
+   "control" means (i) the power, direct or indirect, to cause the
+   direction or management of such entity, whether by contract or
+   otherwise, or (ii) ownership of fifty percent (50%) or more of the
+   outstanding shares, or (iii) beneficial ownership of such entity.
+
+   "You" (or "Your") shall mean an individual or Legal Entity
+   exercising permissions granted by this License.
+
+   "Source" form shall mean the preferred form for making modifications,
+   including but not limited to software source code, documentation
+   source, and configuration files.
+
+   "Object" form shall mean any form resulting from mechanical
+   transformation or translation of a Source form, including but
+   not limited to compiled object code, generated documentation,
+   and conversions to other media types.
+
+   "Work" shall mean the work of authorship, whether in Source or
+   Object form, made available under the License, as indicated by a
+   copyright notice that is included in or attached to the work
+   (an example is provided in the Appendix below).
+
+   "Derivative Works" shall mean any work, whether in Source or Object
+   form, that is based on (or derived from) the Work and for which the
+   editorial revisions, annotations, elaborations, or other modifications
+   represent, as a whole, an original work of authorship. For the purposes
+   of this License, Derivative Works shall not include works that remain
+   separable from, or merely link (or bind by name) to the interfaces of,
+   the Work and Derivative Works thereof.
+
+   "Contribution" shall mean any work of authorship, including
+   the original version of the Work and any modifications or additions
+   to that Work or Derivative Works thereof, that is intentionally
+   submitted to Licensor for inclusion in the Work by the copyright owner
+   or by an individual or Legal Entity authorized to submit on behalf of
+   the copyright owner. For the purposes of this definition, "submitted"
+   means any form of electronic, verbal, or written communication sent
+   to the Licensor or its representatives, including but not limited to
+   communication on electronic mailing lists, source code control systems,
+   and issue tracking systems that are managed by, or on behalf of, the
+   Licensor for the purpose of discussing and improving the Work, but
+   excluding communication that is conspicuously marked or otherwise
+   designated in writing by the copyright owner as "Not a Contribution."
+
+   "Contributor" shall mean Licensor and any individual or Legal Entity
+   on behalf of whom a Contribution has been received by Licensor and
+   subsequently incorporated within the Work.
+
+2. Grant of Copyright License. Subject to the terms and conditions of
+   this License, each Contributor hereby grants to You a perpetual,
+   worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+   copyright license to reproduce, prepare Derivative Works of,
+   publicly display, publicly perform, sublicense, and distribute the
+   Work and such Derivative Works in Source or Object form.
+
+3. Grant of Patent License. Subject to the terms and conditions of
+   this License, each Contributor hereby grants to You a perpetual,
+   worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+   (except as stated in this section) patent license to make, have made,
+   use, offer to sell, sell, import, and otherwise transfer the Work,
+   where such license applies only to those patent claims licensable
+   by such Contributor that are necessarily infringed by their
+   Contribution(s) alone or by combination of their Contribution(s)
+   with the Work to which such Contribution(s) was submitted. If You
+   institute patent litigation against any entity (including a
+   cross-claim or counterclaim in a lawsuit) alleging that the Work
+   or a Contribution incorporated within the Work constitutes direct
+   or contributory patent infringement, then any patent licenses
+   granted to You under this License for that Work shall terminate
+   as of the date such litigation is filed.
+
+4. Redistribution. You may reproduce and distribute copies of the
+   Work or Derivative Works thereof in any medium, with or without
+   modifications, and in Source or Object form, provided that You
+   meet the following conditions:
+
+   (a) You must give any other recipients of the Work or
+       Derivative Works a copy of this License; and
+
+   (b) You must cause any modified files to carry prominent notices
+       stating that You changed the files; and
+
+   (c) You must retain, in the Source form of any Derivative Works
+       that You distribute, all copyright, patent, trademark, and
+       attribution notices from the Source form of the Work,
+       excluding those notices that do not pertain to any part of
+       the Derivative Works; and
+
+   (d) If the Work includes a "NOTICE" text file as part of its
+       distribution, then any Derivative Works that You distribute must
+       include a readable copy of the attribution notices contained
+       within such NOTICE file, excluding those notices that do not
+       pertain to any part of the Derivative Works, in at least one
+       of the following places: within a NOTICE text file distributed
+       as part of the Derivative Works; within the Source form or
+       documentation, if provided along with the Derivative Works; or,
+       within a display generated by the Derivative Works, if and
+       wherever such third-party notices normally appear. The contents
+       of the NOTICE file are for informational purposes only and
+       do not modify the License. You may add Your own attribution
+       notices within Derivative Works that You distribute, alongside
+       or as an addendum to the NOTICE text from the Work, provided
+       that such additional attribution notices cannot be construed
+       as modifying the License.
+
+   You may add Your own copyright statement to Your modifications and
+   may provide additional or different license terms and conditions
+   for use, reproduction, or distribution of Your modifications, or
+   for any such Derivative Works as a whole, provided Your use,
+   reproduction, and distribution of the Work otherwise complies with
+   the conditions stated in this License.
+
+5. Submission of Contributions. Unless You explicitly state otherwise,
+   any Contribution intentionally submitted for inclusion in the Work
+   by You to the Licensor shall be under the terms and conditions of
+   this License, without any additional terms or conditions.
+   Notwithstanding the above, nothing herein shall supersede or modify
+   the terms of any separate license agreement you may have executed
+   with Licensor regarding such Contributions.
+
+6. Trademarks. This License does not grant permission to use the trade
+   names, trademarks, service marks, or product names of the Licensor,
+   except as required for reasonable and customary use in describing the
+   origin of the Work and reproducing the content of the NOTICE file.
+
+7. Disclaimer of Warranty. Unless required by applicable law or
+   agreed to in writing, Licensor provides the Work (and each
+   Contributor provides its Contributions) on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+   implied, including, without limitation, any warranties or conditions
+   of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
+   PARTICULAR PURPOSE. You are solely responsible for determining the
+   appropriateness of using or redistributing the Work and assume any
+   risks associated with Your exercise of permissions under this License.
+
+8. Limitation of Liability. In no event and under no legal theory,
+   whether in tort (including negligence), contract, or otherwise,
+   unless required by applicable law (such as deliberate and grossly
+   negligent acts) or agreed to in writing, shall any Contributor be
+   liable to You for damages, including any direct, indirect, special,
+   incidental, or consequential damages of any character arising as a
+   result of this License or out of the use or inability to use the
+   Work (including but not limited to damages for loss of goodwill,
+   work stoppage, computer failure or malfunction, or any and all
+   other commercial damages or losses), even if such Contributor
+   has been advised of the possibility of such damages.
+
+9. Accepting Warranty or Additional Liability. While redistributing
+   the Work or Derivative Works thereof, You may choose to offer,
+   and charge a fee for, acceptance of support, warranty, indemnity,
+   or other liability obligations and/or rights consistent with this
+   License. However, in accepting such obligations, You may act only
+   on Your own behalf and on Your sole responsibility, not on behalf
+   of any other Contributor, and only if You agree to indemnify,
+   defend, and hold each Contributor harmless for any liability
+   incurred by, or claims asserted against, such Contributor by reason
+   of your accepting any such warranty or additional liability.
+
+END OF TERMS AND CONDITIONS
+
+APPENDIX: How to apply the Apache License to your work.
+
+   To apply the Apache License to your work, attach the following
+   boilerplate notice, with the fields enclosed by brackets "[]"
+   replaced with your own identifying information. (Don't include
+   the brackets!)  The text should be enclosed in the appropriate
+   comment syntax for the file format. We also recommend that a
+   file or class name and description of purpose be included on the
+   same "printed page" as the copyright notice for easier
+   identification within third-party archives.
+
+Copyright [2016] [ssdp-rs Developers]
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+	http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.

third-party/rupnp/LICENSE-MIT

@@ -0,0 +1,25 @@
+Copyright (c) 2016 ssdp-rs Developers
+
+Permission is hereby granted, free of charge, to any
+person obtaining a copy of this software and associated
+documentation files (the "Software"), to deal in the
+Software without restriction, including without
+limitation the rights to use, copy, modify, merge,
+publish, distribute, sublicense, and/or sell copies of
+the Software, and to permit persons to whom the Software
+is furnished to do so, subject to the following
+conditions:
+
+The above copyright notice and this permission notice
+shall be included in all copies or substantial portions
+of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF
+ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED
+TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A
+PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT
+SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
+CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
+OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR
+IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+DEALINGS IN THE SOFTWARE.

third-party/rupnp/QBZ-PATCH.md

@@ -0,0 +1,28 @@
+# QBZ UPnP endpoint compatibility
+
+Source: crates.io `rupnp` 3.0.0 (MIT / Apache-2.0); upstream repository and
+commit are recorded in Cargo.toml and .cargo_vcs_info.json. All original files
+are retained. The only implementation change is in src/service.rs:
+`parse_service_endpoint` handles empty or missing-leading-slash service URLs.
+An annotation in device.rs also silences the upstream unused presentation_url
+field, whose warning is normally capped for registry dependencies.
+
+Valid `PathAndQuery` values are returned by the original parser. The fallback
+restores empty-path tolerance and normalizes rejected relative URLs into
+root-relative request targets, reusing the current http parser so character
+and length checks still apply. In 1.4.0 a relative endpoint was actually sent
+without a leading slash; this patch deliberately uses valid origin-form instead.
+It does not change endpoints that were valid in 2.1.0. Device fetching,
+SSDP, SOAP, service selection and URLBase semantics are unchanged.
+
+This replaces the workspace-wide http 1.4.0 pin from #745, retaining the
+subsequent HTTP validation and HeaderMap fixes. Fixtures are synthetic; KEF
+hardware confirmation remains required. Tests live in qbz-cast/tests:
+`dlna_description_tolerance.rs` and `dlna_wire_compatibility.rs`.
+
+The workspace uses a local Cargo patch, never a network git dependency. This
+folder ships in the QBZ source archive; registry dependencies still come from
+the locked release vendor archive for offline AUR/Gentoo builds.
+
+Remove this patch once an upstream release provides the same compatibility
+and passes those tests. This is not a general UPnP URL-resolution rewrite.

third-party/rupnp/README.md

@@ -0,0 +1,63 @@
+![GitHub last commit](https://img.shields.io/github/last-commit/jakobhellermann/rupnp.svg)
+[![Crates.io](https://img.shields.io/crates/v/rupnp.svg)](https://crates.io/crates/rupnp)
+
+# rupnp
+An asynchronous library for finding UPnP control points, performing actions on them
+and reading their service descriptions.
+UPnP stand for `Universal Plug and Play` and is widely used for routers, WiFi-enabled speakers
+and media servers.
+
+Spec:
+[http://rupnp.org/specs/arch/UPnP-arch-DeviceArchitecture-v2.0.pdf](http://upnp.org/specs/arch/UPnP-arch-DeviceArchitecture-v2.0.pdf)
+
+# Example usage:
+The following code searches for devices that have a `RenderingControl` service 
+and print their names along with their current volume.
+```rust
+use futures::prelude::*;
+use std::time::Duration;
+use rupnp::ssdp::{SearchTarget, URN};
+
+const RENDERING_CONTROL: URN = URN::service("schemas-upnp-org", "RenderingControl", 1);
+
+#[tokio::main]
+async fn main() -> Result<(), rupnp::Error> {
+    let search_target = SearchTarget::URN(RENDERING_CONTROL);
+    let devices = rupnp::discover(&search_target, Duration::from_secs(3), None).await?;
+    let mut devices = std::pin::pin!(devices);
+
+    while let Some(device) = devices.try_next().await? {
+        let service = device
+            .find_service(&RENDERING_CONTROL)
+            .expect("searched for RenderingControl, got something else");
+
+        let args = "<InstanceID>0</InstanceID><Channel>Master</Channel>";
+        let response = service.action(device.url(), "GetVolume", args).await?;
+
+        let volume = response.get("CurrentVolume").unwrap();
+
+        println!("'{}' is at volume {}", device.friendly_name(), volume);
+    }
+
+    Ok(())
+}
+```
+
+License
+-------
+
+Licensed under either of
+
+ * Apache License, Version 2.0 ([LICENSE-APACHE](LICENSE-APACHE) or http://www.apache.org/licenses/LICENSE-2.0)
+ * MIT license ([LICENSE-MIT](LICENSE-MIT) or http://opensource.org/licenses/MIT)
+
+at your option.
+
+Contribution
+------------
+
+Please use [rustfmt](https://github.com/rust-lang/rustfmt) before any pull requests.
+
+Unless you explicitly state otherwise, any contribution intentionally submitted
+for inclusion in the work by you, as defined in the Apache-2.0 license, shall be dual licensed as above, without any
+additional terms or conditions.

third-party/rupnp/examples/discover.rs

@@ -0,0 +1,29 @@
+use anyhow::Result;
+use futures::prelude::*;
+use rupnp::ssdp::SearchTarget;
+use std::time::Duration;
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    let devices = rupnp::discover(&SearchTarget::RootDevice, Duration::from_secs(3), None).await?;
+    let mut devices = std::pin::pin!(devices);
+
+    while let Some(device) = devices.next().await {
+        let device = match device {
+            Ok(device) => device,
+            Err(e) => {
+                eprintln!("{e}");
+                continue;
+            }
+        };
+
+        println!(
+            "{} - {} @ {}",
+            device.device_type(),
+            device.friendly_name(),
+            device.url()
+        );
+    }
+
+    Ok(())
+}

third-party/rupnp/examples/discover_extra_fields.rs

@@ -0,0 +1,28 @@
+use futures::prelude::*;
+use rupnp::ssdp::SearchTarget;
+use std::time::Duration;
+
+const EXTRA: &[&str; 2] = &["manufacturer", "manufacturerURL"];
+
+#[tokio::main]
+async fn main() -> Result<(), rupnp::Error> {
+    let devices = rupnp::discover_with_properties(
+        &SearchTarget::RootDevice,
+        Duration::from_secs(3),
+        None,
+        EXTRA,
+    )
+    .await?;
+    let mut devices = std::pin::pin!(devices);
+
+    while let Some(maybe_device) = devices.next().await {
+        if let Ok(device) = maybe_device { println!(
+            "{} from {} @ {}",
+            device.friendly_name(),
+            device.get_extra_property(EXTRA[0]).unwrap_or_default(),
+            device.get_extra_property(EXTRA[1]).unwrap_or_default()
+        ) };
+    }
+
+    Ok(())
+}

third-party/rupnp/examples/dump_scpd.rs

@@ -0,0 +1,116 @@
+use futures::prelude::*;
+use rupnp::{
+    http::Uri,
+    ssdp::{SearchTarget, URN},
+    DeviceSpec, Error, Service,
+};
+use std::{
+    fs,
+    path::{Path, PathBuf},
+    time::Duration,
+};
+use tokio::task::{spawn, JoinHandle};
+
+#[tokio::main]
+async fn main() -> Result<(), Error> {
+    let mut devices: Vec<_> =
+        rupnp::discover(&SearchTarget::RootDevice, Duration::from_secs(1), None)
+            .await?
+            .filter_map(|result| async { result.map_err(|e| println!("{e}")).ok() })
+            .collect()
+            .await;
+
+    devices.sort_by_key(|d| d.device_type().clone());
+    devices.dedup_by_key(|d| d.device_type().clone());
+
+    let path = PathBuf::from("descriptions");
+    if path.exists() {
+        fs::remove_dir_all(&path)?;
+    }
+
+    let mut handles = Vec::new();
+
+    for device in devices {
+        print(&device, device.url(), 0, &path, &mut handles)?;
+        println!();
+    }
+
+    for handle in handles {
+        handle.await.unwrap()?;
+    }
+
+    Ok(())
+}
+
+fn print(
+    device: &DeviceSpec,
+    url: &Uri,
+    indentation: usize,
+    path: &Path,
+    handles: &mut Vec<JoinHandle<Result<(), rupnp::Error>>>,
+) -> Result<(), Error> {
+    let path = path.join(urn_to_str(device.device_type()));
+    fs::create_dir_all(&path)?;
+
+    let i = "  ".repeat(indentation);
+
+    println!("{}{}", i, urn_to_str(device.device_type()));
+
+    for service in device.services() {
+        let svc = urn_to_str(service.service_type());
+
+        println!("{i}  - {svc}");
+
+        let url = url.clone();
+        let path = path.clone();
+        let service = service.clone();
+
+        handles.push(spawn(async move {
+            use std::io::Write;
+
+            // let mut svc_file = tokio::fs::File::create(path.join(&svc)).await?;
+            let mut svc_file = std::fs::File::create(path.join(&svc))?;
+
+            let mut buf = Vec::with_capacity(128);
+            write_service(&mut buf, service, url).await?;
+
+            svc_file.write_all(&buf)?;
+            // svc_file.write_all(&buf).await?;
+            Ok(())
+        }));
+    }
+
+    for device in device.devices() {
+        print(device, url, indentation + 1, &path, handles)?;
+    }
+
+    Ok(())
+}
+
+async fn write_service(
+    mut w: impl std::io::Write,
+    service: Service,
+    url: Uri,
+) -> Result<(), Error> {
+    let scpd = service.scpd(&url).await?;
+
+    writeln!(w, "StateVars {{")?;
+    for state_var in scpd.state_variables() {
+        if state_var.sends_events() {
+            writeln!(w, "  {state_var} (sends events)")?;
+        } else {
+            writeln!(w, "  {state_var}")?;
+        }
+    }
+    writeln!(w, "}}\n\nActions {{")?;
+    for action in scpd.actions() {
+        writeln!(w, "  {action}")?;
+    }
+    writeln!(w, "}}")?;
+
+    Ok(())
+}
+
+fn urn_to_str(urn: &URN) -> String {
+    urn.typ().to_string().to_lowercase()
+}

third-party/rupnp/examples/rendering_control.rs

@@ -0,0 +1,23 @@
+use rupnp::{http::Uri, ssdp::URN, Device};
+
+#[tokio::main]
+async fn main() -> Result<(), rupnp::Error> {
+    let url = Uri::from_static("http://192.168.178.40:1400/xml/device_description.xml");
+    let service_urn = URN::service("schemas-upnp-org", "RenderingControl", 1);
+
+    let device = Device::from_url(url).await?;
+    let service = device.find_service(&service_urn).unwrap();
+
+    let args = "<InstanceID>0</InstanceID><Channel>Master</Channel>";
+    let response: u8 = service
+        .action(device.url(), "GetVolume", args)
+        .await?
+        .get("CurrentVolume")
+        .unwrap()
+        .parse()
+        .map_err(rupnp::Error::invalid_response)?;
+
+    println!("{response}");
+
+    Ok(())
+}

third-party/rupnp/examples/search_and_action.rs

@@ -0,0 +1,27 @@
+use futures::prelude::*;
+use rupnp::ssdp::{SearchTarget, URN};
+use std::time::Duration;
+
+const RENDERING_CONTROL: URN = URN::service("schemas-upnp-org", "RenderingControl", 1);
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    let search_target = SearchTarget::URN(RENDERING_CONTROL);
+    let devices = rupnp::discover(&search_target, Duration::from_secs(3), None).await?;
+    let mut devices = std::pin::pin!(devices);
+
+    while let Some(device) = devices.try_next().await? {
+        let service = device
+            .find_service(&RENDERING_CONTROL)
+            .expect("searched for RenderingControl, got something else");
+
+        let args = "<InstanceID>0</InstanceID><Channel>Master</Channel>";
+        let response = service.action(device.url(), "GetVolume", args).await?;
+
+        let volume = response.get("CurrentVolume").unwrap();
+
+        println!("'{}' is at volume {}", device.friendly_name(), volume);
+    }
+
+    Ok(())
+}

third-party/rupnp/examples/subscribe.rs

@@ -0,0 +1,33 @@
+use futures::prelude::*;
+use rupnp::{http::Uri, ssdp::URN, Device};
+use std::collections::HashMap;
+
+#[tokio::main]
+async fn main() -> Result<(), rupnp::Error> {
+    let url = Uri::from_static("http://192.168.178.40:1400/xml/device_description.xml");
+    let service_urn = URN::service("schemas-upnp-org", "ZoneGroupTopology", 1);
+
+    let device = Device::from_url(url).await?;
+    let service = device.find_service(&service_urn).unwrap();
+
+    let (sid, mut stream) = service.subscribe(device.url(), 10).await?;
+
+    while let Some(state_vars) = stream.try_next().await? {
+        handle(state_vars);
+        service.renew_subscription(device.url(), &sid, 10).await?;
+    }
+
+    Ok(())
+}
+
+fn handle(state_vars: HashMap<String, String>) {
+    println!("Change {{");
+    for (key, value) in state_vars {
+        if value.len() > 256 {
+            println!("  {key}: ...");
+        } else {
+            println!("  {key}: {value}");
+        }
+    }
+    println!("}}");
+}

third-party/rupnp/src/device.rs

@@ -0,0 +1,286 @@
+use crate::{
+    find_in_xml,
+    utils::{self, HttpResponseExt, HyperBodyExt},
+    Result, Service,
+};
+use bytes::Bytes;
+use http::Uri;
+use http_body_util::Empty;
+use hyper_util::rt::TokioExecutor;
+use roxmltree::{Document, Node};
+use ssdp_client::URN;
+use std::collections::HashMap;
+use std::hash::Hash;
+use std::hash::Hasher;
+
+#[derive(Debug, Clone)]
+/// A UPnP Device.
+/// It stores its [`Uri`] and a [`DeviceSpec`], which contains information like the device type and
+/// its list of inner devices and services.
+pub struct Device {
+    url: Uri,
+    device_spec: DeviceSpec,
+}
+impl Device {
+    pub fn url(&self) -> &Uri {
+        &self.url
+    }
+
+    /// Creates a UPnP device from the given url.
+    /// The url should point to the `/device_description.xml` or similar of the device.
+    /// If you dont know the concrete location, use [`discover`](fn.discover.html) instead.
+    pub async fn from_url(url: Uri) -> Result<Self> {
+        Self::from_url_and_properties(url, &[]).await
+    }
+
+    /// Creates a UPnP device from the given url, defining extra device properties
+    /// to be accessed with `get_extra_property`.
+    pub async fn from_url_and_properties(url: Uri, extra_keys: &[&str]) -> Result<Self> {
+        let body = hyper_util::client::legacy::Client::builder(TokioExecutor::new())
+            .build_http::<Empty<Bytes>>()
+            .get(url.clone())
+            .await?
+            .err_if_not_200()?
+            .into_body()
+            .bytes()
+            .await?;
+
+        // .into_body()
+        // .text()
+        // .await?;
+        let body = std::str::from_utf8(&body)?;
+
+        let document = Document::parse(body)?;
+        let device = utils::find_root(&document, "device", "Device Description")?;
+        let device_spec = DeviceSpec::from_xml(device, extra_keys)?;
+
+        Ok(Self { url, device_spec })
+    }
+}
+impl std::ops::Deref for Device {
+    type Target = DeviceSpec;
+
+    fn deref(&self) -> &Self::Target {
+        &self.device_spec
+    }
+}
+impl Hash for Device {
+    fn hash<H: Hasher>(&self, state: &mut H) {
+        self.url.hash(state);
+    }
+}
+impl PartialEq for Device {
+    fn eq(&self, other: &Self) -> bool {
+        self.url == other.url
+    }
+}
+impl Eq for Device {}
+
+/// Information about a device.
+///
+/// By default it only includes its *friendly name*, device type, a list of subdevices and
+/// services, and a `HashMap` of extra properties in order to keep the structs size small.
+///
+/// If you also want the `ManufacturerURL`, `Model{Description,Number,Url}`, `serial number`, `UDN` and
+/// `UPC` as struct fields, enable the `full_device_spec` feature.
+#[derive(Debug, Clone)]
+pub struct DeviceSpec {
+    device_type: URN,
+    friendly_name: String,
+
+    devices: Vec<DeviceSpec>,
+    services: Vec<Service>,
+
+    extra_properties: HashMap<String, Option<String>>,
+
+    #[cfg(feature = "full_device_spec")]
+    manufacturer: String,
+    #[cfg(feature = "full_device_spec")]
+    manufacturer_url: Option<String>,
+    #[cfg(feature = "full_device_spec")]
+    model_name: String,
+    #[cfg(feature = "full_device_spec")]
+    model_description: Option<String>,
+    #[cfg(feature = "full_device_spec")]
+    model_number: Option<String>,
+    #[cfg(feature = "full_device_spec")]
+    model_url: Option<String>,
+    #[cfg(feature = "full_device_spec")]
+    serial_number: Option<String>,
+    #[cfg(feature = "full_device_spec")]
+    udn: String,
+    #[cfg(feature = "full_device_spec")]
+    upc: Option<String>,
+    #[cfg(feature = "full_device_spec")]
+    #[allow(dead_code)] // Upstream retains this parsed field without a getter.
+    presentation_url: Option<String>,
+}
+
+impl DeviceSpec {
+    fn from_xml<'a, 'input: 'a>(node: Node<'a, 'input>, extra_keys: &[&str]) -> Result<Self> {
+        #[rustfmt::skip]
+        #[allow(non_snake_case)]
+        let (device_type, friendly_name, services, devices, extra_properties) =
+            find_in_xml! { node => deviceType, friendlyName, ?serviceList, ?deviceList, #extra_keys };
+
+        #[cfg(feature = "full_device_spec")]
+        #[allow(non_snake_case)]
+        let (
+            manufacturer,
+            manufacturer_url,
+            model_name,
+            model_description,
+            model_number,
+            model_url,
+            serial_number,
+            udn,
+            upc,
+            presentation_url,
+        ) = find_in_xml! { node => manufacturer, ?manufacturerURL, modelName, ?modelDescription, ?modelNumber, ?modelURL, ?serialNumber, UDN, ?UPC, ?PresentationURL};
+
+        #[cfg(feature = "full_device_spec")]
+        let manufacturer_url = manufacturer_url.map(utils::parse_node_text).transpose()?;
+        #[cfg(feature = "full_device_spec")]
+        let model_description = model_description.map(utils::parse_node_text).transpose()?;
+        #[cfg(feature = "full_device_spec")]
+        let model_number = model_number.map(utils::parse_node_text).transpose()?;
+        #[cfg(feature = "full_device_spec")]
+        let model_url = model_url.map(utils::parse_node_text).transpose()?;
+        #[cfg(feature = "full_device_spec")]
+        let serial_number = serial_number.map(utils::parse_node_text).transpose()?;
+        #[cfg(feature = "full_device_spec")]
+        let upc = upc.map(utils::parse_node_text).transpose()?;
+        #[cfg(feature = "full_device_spec")]
+        let presentation_url = presentation_url.map(utils::parse_node_text).transpose()?;
+
+        let devices = match devices {
+            Some(d) => d
+                .children()
+                .filter(Node::is_element)
+                .map(|node| DeviceSpec::from_xml(node, extra_keys))
+                .collect::<Result<_>>()?,
+            None => Vec::new(),
+        };
+        let services = match services {
+            Some(s) => s
+                .children()
+                .filter(Node::is_element)
+                .map(Service::from_xml)
+                .collect::<Result<_>>()?,
+            None => Vec::new(),
+        };
+
+        Ok(Self {
+            device_type: utils::parse_node_text(device_type)?,
+            friendly_name: utils::parse_node_text(friendly_name)?,
+            #[cfg(feature = "full_device_spec")]
+            manufacturer: utils::parse_node_text(manufacturer)?,
+            #[cfg(feature = "full_device_spec")]
+            udn: utils::parse_node_text(udn)?,
+            #[cfg(feature = "full_device_spec")]
+            manufacturer_url,
+            #[cfg(feature = "full_device_spec")]
+            model_name: utils::parse_node_text(model_name)?,
+            #[cfg(feature = "full_device_spec")]
+            model_description,
+            #[cfg(feature = "full_device_spec")]
+            model_number,
+            #[cfg(feature = "full_device_spec")]
+            model_url,
+            #[cfg(feature = "full_device_spec")]
+            serial_number,
+            #[cfg(feature = "full_device_spec")]
+            upc,
+            #[cfg(feature = "full_device_spec")]
+            presentation_url,
+            devices,
+            services,
+            extra_properties,
+        })
+    }
+
+    pub fn device_type(&self) -> &URN {
+        &self.device_type
+    }
+    pub fn friendly_name(&self) -> &str {
+        &self.friendly_name
+    }
+    pub fn get_extra_property(&self, elem: &str) -> Option<&str> {
+        self.extra_properties
+            .get(elem)
+            .and_then(|o| o.as_ref())
+            .map(String::as_str)
+    }
+
+    #[cfg(feature = "full_device_spec")]
+    pub fn manufacturer(&self) -> &str {
+        &self.manufacturer
+    }
+    #[cfg(feature = "full_device_spec")]
+    pub fn manufacturer_url(&self) -> Option<&str> {
+        self.manufacturer_url.as_ref().map(String::as_str)
+    }
+    #[cfg(feature = "full_device_spec")]
+    pub fn model_name(&self) -> &str {
+        &self.model_name
+    }
+    #[cfg(feature = "full_device_spec")]
+    pub fn model_description(&self) -> Option<&str> {
+        self.model_description.as_ref().map(String::as_str)
+    }
+    #[cfg(feature = "full_device_spec")]
+    pub fn model_number(&self) -> Option<&str> {
+        self.model_number.as_ref().map(String::as_str)
+    }
+    #[cfg(feature = "full_device_spec")]
+    pub fn model_url(&self) -> Option<&str> {
+        self.model_url.as_ref().map(String::as_str)
+    }
+    #[cfg(feature = "full_device_spec")]
+    pub fn serial_number(&self) -> Option<&str> {
+        self.serial_number.as_ref().map(String::as_str)
+    }
+    #[cfg(feature = "full_device_spec")]
+    pub fn udn(&self) -> &str {
+        &self.udn
+    }
+    #[cfg(feature = "full_device_spec")]
+    pub fn upc(&self) -> Option<&str> {
+        self.upc.as_ref().map(String::as_str)
+    }
+
+    /// Returns a list of this devices subdevices.
+    /// Note that this does not recurse, if you want that behaviour use
+    /// [devices_iter](struct.DeviceSpec.html#method.devices_iter) instead.
+    pub fn devices(&self) -> &Vec<DeviceSpec> {
+        &self.devices
+    }
+
+    /// Returns a list of this devices services.
+    /// Note that this does not recurse, if you want that behaviour use
+    /// [services_iter](struct.DeviceSpec.html#method.services_iter) instead.
+    pub fn services(&self) -> &Vec<Service> {
+        &self.services
+    }
+
+    /// Returns an Iterator of all services that can be used from this device.
+    pub fn services_iter(&self) -> impl Iterator<Item = &Service> {
+        self.services().iter().chain(self.devices().iter().flat_map(
+            |device| -> Box<dyn Iterator<Item = &Service>> { Box::new(device.services_iter()) },
+        ))
+    }
+    pub fn find_service(&self, service_type: &URN) -> Option<&Service> {
+        self.services_iter()
+            .find(|s| s.service_type() == service_type)
+    }
+
+    /// Returns an Iterator of all devices that can be used from this device.
+    pub fn devices_iter(&self) -> impl Iterator<Item = &DeviceSpec> {
+        self.devices().iter().chain(self.devices().iter().flat_map(
+            |device| -> Box<dyn Iterator<Item = &DeviceSpec>> { Box::new(device.devices_iter()) },
+        ))
+    }
+    pub fn find_device(&self, device_type: &URN) -> Option<&DeviceSpec> {
+        self.devices_iter().find(|d| &d.device_type == device_type)
+    }
+}

third-party/rupnp/src/discovery.rs

@@ -0,0 +1,74 @@
+use crate::{Device, Error, Result};
+use futures_util::stream::{Stream, StreamExt, TryStreamExt};
+use ssdp_client::SearchTarget;
+use std::time::Duration;
+
+/// Discovers UPnP devices on the network.
+///
+/// # Example usage:
+/// ```rust,no_run
+/// use futures::prelude::*;
+/// use std::time::Duration;
+/// use rupnp::ssdp::SearchTarget;
+///
+/// # async fn discover() -> Result<(), rupnp::Error> {
+/// let devices = rupnp::discover(&SearchTarget::RootDevice, Duration::from_secs(3), None).await?;
+/// let mut devices = std::pin::pin!(devices);
+///
+/// while let Some(device) = devices.try_next().await? {
+///     println!(
+///         "{} - {} @ {}",
+///         device.device_type(),
+///         device.friendly_name(),
+///         device.url()
+///     );
+/// }
+///
+/// # Ok(())
+/// # }
+/// ```
+// TODO: doc include once stable
+pub async fn discover(
+    search_target: &SearchTarget,
+    timeout: Duration,
+    ttl: Option<u32>,
+) -> Result<impl Stream<Item = Result<Device>>> {
+    return discover_with_properties(search_target, timeout, ttl, &[]).await;
+}
+
+/// Discovers UPnP devices on the network and saves extra_fields in device descriptions
+///
+/// # Example usage:
+/// ```rust,no_run
+/// use futures::prelude::*;
+/// use std::time::Duration;
+/// use rupnp::ssdp::SearchTarget;
+///
+/// # async fn discover_with_properties() -> Result<(), rupnp::Error> {
+/// let devices = rupnp::discover_with_properties(&SearchTarget::RootDevice, Duration::from_secs(3), None, &["manufacturer", "manufacturerURL"]).await?;
+/// let mut devices = std::pin::pin!(devices);
+///
+/// while let Some(device) = devices.try_next().await? {
+///     println!(
+///         "{} - {} @ {}",
+///         device.device_type(),
+///         device.get_extra_property("manufacturer").unwrap_or_default(),
+///         device.get_extra_property("manufacturerURL").unwrap_or_default()
+///     );
+/// }
+///
+/// # Ok(())
+/// # }
+/// ```
+pub async fn discover_with_properties<'a>(
+    search_target: &SearchTarget,
+    timeout: Duration,
+    ttl: Option<u32>,
+    extra_keys: &'a [&'a str],
+) -> Result<impl Stream<Item = Result<Device>> + 'a> {
+    Ok(ssdp_client::search(search_target, timeout, 3, ttl)
+        .await?
+        .map_err(Error::SSDPError)
+        .map(|res| Ok(res?.location().parse()?))
+        .and_then(move |url| Device::from_url_and_properties(url, extra_keys)))
+}

third-party/rupnp/src/error.rs

@@ -0,0 +1,180 @@
+use crate::find_in_xml;
+use std::{fmt, str::Utf8Error};
+
+/// The UPnP Error type.
+#[derive(Debug)]
+#[non_exhaustive]
+pub enum Error {
+    UPnPError(UPnPError),
+    SSDPError(ssdp_client::Error),
+    NetworkError(hyper::Error),
+    NetworkClientError(hyper_util::client::legacy::Error),
+    NoLocalInterfaceOpen,
+    IO(std::io::Error),
+    InvalidUrl(http::uri::InvalidUri),
+    InvalidUtf8(Utf8Error),
+    ParseError(&'static str),
+    HttpErrorCode(http::StatusCode),
+    XmlError(roxmltree::Error),
+    XmlMissingElement(String, String),
+    InvalidResponse(Box<dyn std::error::Error + Send + Sync + 'static>),
+}
+impl Error {
+    pub fn invalid_response<E: std::error::Error + Send + Sync + 'static>(err: E) -> Error {
+        Error::InvalidResponse(Box::new(err))
+    }
+}
+
+impl fmt::Display for Error {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Error::UPnPError(err) => write!(f, "{err}"),
+            Error::SSDPError(err) => write!(f, "error trying to discover devices: {err}"),
+            Error::IO(err) => write!(f, "error reading response: {err}"),
+            Error::NetworkError(err) => {
+                write!(f, "An error occurred trying to connect to device: {err}")
+            }
+            Error::NetworkClientError(err) => {
+                write!(f, "An error occurred trying to connect to device: {err}")
+            }
+            Error::NoLocalInterfaceOpen => write!(
+                f,
+                "could not subscribe to events: no local ipv4 interface open"
+            ),
+            Error::InvalidUrl(err) => write!(f, "invalid url: {err}"),
+            Error::InvalidUtf8(err) => write!(f, "invalid utf8: {err}"),
+            Error::ParseError(err) => write!(f, "{err}"),
+            Error::InvalidResponse(err) => write!(f, "Invalid response: {err}"),
+            Error::HttpErrorCode(code) => {
+                write!(f, "The control point responded with status code {code}")
+            }
+            Error::XmlError(err) => write!(f, "failed to parse xml: {err}"),
+            Error::XmlMissingElement(parent, child) => write!(
+                f,
+                "`{parent}` does not contain a `{child}` element or attribute"
+            ),
+        }
+    }
+}
+impl std::error::Error for Error {
+    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
+        match self {
+            Error::UPnPError(err) => Some(err),
+            Error::SSDPError(err) => Some(err),
+            Error::NetworkError(err) => Some(err),
+            Error::IO(err) => Some(err),
+            Error::InvalidUrl(err) => Some(err),
+            Error::InvalidUtf8(err) => Some(err),
+            Error::XmlError(err) => Some(err),
+            _ => None,
+        }
+    }
+}
+impl From<hyper::Error> for Error {
+    fn from(err: hyper::Error) -> Self {
+        Error::NetworkError(err)
+    }
+}
+impl From<hyper_util::client::legacy::Error> for Error {
+    fn from(err: hyper_util::client::legacy::Error) -> Self {
+        Error::NetworkClientError(err)
+    }
+}
+impl From<roxmltree::Error> for Error {
+    fn from(err: roxmltree::Error) -> Self {
+        Error::XmlError(err)
+    }
+}
+impl From<std::io::Error> for Error {
+    fn from(err: std::io::Error) -> Self {
+        Error::IO(err)
+    }
+}
+impl From<http::uri::InvalidUri> for Error {
+    fn from(err: http::uri::InvalidUri) -> Self {
+        Error::InvalidUrl(err)
+    }
+}
+impl From<ssdp_client::Error> for Error {
+    fn from(err: ssdp_client::Error) -> Self {
+        Error::SSDPError(err)
+    }
+}
+impl From<UPnPError> for Error {
+    fn from(err: UPnPError) -> Self {
+        Error::UPnPError(err)
+    }
+}
+impl From<Utf8Error> for Error {
+    fn from(err: Utf8Error) -> Self {
+        Error::InvalidUtf8(err)
+    }
+}
+
+#[derive(Debug)]
+pub struct UPnPError {
+    #[allow(unused)]
+    fault_code: String,
+    fault_string: String,
+    err_code: u16,
+}
+
+impl std::error::Error for UPnPError {}
+impl fmt::Display for UPnPError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(
+            f,
+            "{} {}: {}",
+            self.fault_string,
+            self.err_code,
+            self.err_code_description()
+        )
+    }
+}
+
+impl UPnPError {
+    pub fn err_code(&self) -> u16 {
+        self.err_code
+    }
+
+    pub fn err_code_description(&self) -> &str {
+        match self.err_code {
+            401 => "No action by that name at this service.",
+            402 => "Invalid Arguments",
+            403 => "(deprecated error code)",
+            501 => "Action failed",
+            600 => "Argument value invalid",
+            601 => "Argument Value Out of Range",
+            602 => "Optional Action Not Implemented",
+            603 => "Out of Memory",
+            604 => "Human Intervention Required",
+            605 => "String Argument Too Long",
+            606..=612 => "(error code reserved for UPnP DeviceSecurity)",
+            613..=699 => "Common action error. Defined by UPnP Forum Technical Committee.",
+            700..=799 => "Action-specific error defined by UPnP Forum working committee.",
+            800..=899 => "Action-specific error for non-standard actions. Defined by UPnP vendor.",
+            _ => "Invalid Error Code",
+        }
+    }
+
+    pub(crate) fn from_fault_node(node: roxmltree::Node<'_, '_>) -> Result<UPnPError, Error> {
+        let (fault_code, fault_string, detail) =
+            find_in_xml! { node => faultcode, faultstring, detail };
+        let fault_code = fault_code.text().unwrap_or_default().to_string();
+        let fault_string = fault_string.text().unwrap_or_default().to_string();
+        let err_code = detail
+            .descendants()
+            .find(|n| n.tag_name().name().eq_ignore_ascii_case("errorCode"))
+            .ok_or_else(|| Error::XmlMissingElement("detail".to_string(), "errorCode".to_string()))?
+            .text()
+            .unwrap_or_default()
+            .parse()
+            .map_err(Error::invalid_response)?;
+
+        Ok(UPnPError {
+            fault_code,
+            fault_string,
+            err_code,
+        })
+    }
+}

third-party/rupnp/src/lib.rs

@@ -0,0 +1,67 @@
+#![warn(
+    nonstandard_style,
+    rust_2018_idioms,
+    future_incompatible,
+    missing_debug_implementations
+)]
+
+//! An asynchronous library for finding UPnP control points, performing actions on them
+//! and reading their service descriptions.
+//! UPnP stand for `Universal Plug and Play` and is widely used for routers, WiFi-enabled speakers
+//! and media servers.
+//!
+//! Specification:
+//! [http://upnp.org/specs/arch/UPnP-arch-DeviceArchitecture-v2.0.pdf](http://upnp.org/specs/arch/UPnP-arch-DeviceArchitecture-v2.0.pdf)
+//!
+//! # Example usage:
+//! The following code searches for devices that have a `RenderingControl` service
+//! and prints their names along with their current volume.
+//!
+//! ```rust,no_run
+//! use futures::prelude::*;
+//! use std::time::Duration;
+//! use rupnp::ssdp::{SearchTarget, URN};
+//!
+//! const RENDERING_CONTROL: URN = URN::service("schemas-upnp-org", "RenderingControl", 1);
+//!
+//! #[tokio::main]
+//! async fn main() -> Result<(), rupnp::Error> {
+//!     let search_target = SearchTarget::URN(RENDERING_CONTROL);
+//!     let devices = rupnp::discover(&search_target, Duration::from_secs(3), None).await?;
+//!     let mut devices = std::pin::pin!(devices);
+//!
+//!     while let Some(device) = devices.try_next().await? {
+//!         let service = device
+//!             .find_service(&RENDERING_CONTROL)
+//!             .expect("searched for RenderingControl, got something else");
+//!
+//!         let args = "<InstanceID>0</InstanceID><Channel>Master</Channel>";
+//!         let response = service.action(device.url(), "GetVolume", args).await?;
+//!
+//!         let volume = response.get("CurrentVolume").unwrap();
+//!
+//!         println!("'{}' is at volume {}", device.friendly_name(), volume);
+//!     }
+//!
+//!     Ok(())
+//! }
+//! ```
+// TODO: doc include when it gets stable
+
+mod device;
+mod discovery;
+mod error;
+/// Service Control Protocol Description.
+pub mod scpd;
+mod service;
+pub mod utils;
+
+pub use device::{Device, DeviceSpec};
+pub use discovery::{discover, discover_with_properties};
+pub use error::Error;
+pub use service::Service;
+
+pub use http;
+pub use ssdp_client as ssdp;
+
+pub(crate) type Result<T, E = Error> = std::result::Result<T, E>;

third-party/rupnp/src/scpd/action.rs

@@ -0,0 +1,160 @@
+use crate::{
+    find_in_xml,
+    scpd::{StateVariable, StateVariableKind},
+    utils, Error, Result,
+};
+use roxmltree::Node;
+use std::{fmt, rc::Rc};
+
+/// An SCPD action.
+/// The action consists of its name used in the services
+/// [`action`](../struct.Service.html#method.action) function and a List of
+/// [`Argument`](struct.Argument.html)s
+#[derive(Debug, Clone)]
+pub struct Action {
+    name: String,
+    arguments: Vec<Argument>,
+}
+impl fmt::Display for Action {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "{}(", self.name())?;
+
+        for e in self.input_arguments().take(1) {
+            write!(f, "{e}")?;
+        }
+        for e in self.input_arguments().skip(1) {
+            write!(f, ", {e}")?;
+        }
+        write!(f, ")")?;
+
+        let mut output_arguments = self.output_arguments();
+        if let Some(arg) = output_arguments.next() {
+            write!(f, " -> ({arg}")?;
+            for arg in output_arguments {
+                write!(f, ", {arg}")?;
+            }
+            write!(f, ")")?;
+        }
+
+        Ok(())
+    }
+}
+
+impl Action {
+    pub(crate) fn from_xml(
+        node: Node<'_, '_>,
+        state_variables: &[Rc<StateVariable>],
+    ) -> Result<Self> {
+        #[allow(non_snake_case)]
+        let (name, arguments) = find_in_xml! { node => name, ?argumentList };
+
+        let arguments = arguments
+            .map(|args| {
+                args.children()
+                    .filter(Node::is_element)
+                    .map(|node| Argument::from_xml(node, state_variables))
+                    .collect::<Result<_>>()
+            })
+            .transpose()?
+            .unwrap_or_default();
+
+        Ok(Self {
+            name: utils::parse_node_text(name)?,
+            arguments,
+        })
+    }
+
+    pub fn name(&self) -> &String {
+        &self.name
+    }
+
+    pub fn input_arguments(&self) -> impl Iterator<Item = &Argument> {
+        self.arguments.iter().filter(|a| a.is_input())
+    }
+    pub fn output_arguments(&self) -> impl Iterator<Item = &Argument> {
+        self.arguments.iter().filter(|a| a.is_output())
+    }
+}
+
+/// Every argument has its associated [`StateVariable`](struct.StateVariable.html), which contains
+/// more information about its possible values/range/etc.
+#[derive(Debug, Clone)]
+pub struct Argument {
+    name: String,
+    // if not input, it is an output
+    is_input: bool,
+    state_var: Rc<StateVariable>,
+}
+impl fmt::Display for Argument {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "{}: ", &self.name)?;
+
+        match self.state_var.kind() {
+            StateVariableKind::Simple(datatype) => write!(f, "{datatype}")?,
+            StateVariableKind::Enum(variants) => {
+                write!(f, "[{}]", variants.join(", "))?;
+            }
+            StateVariableKind::Range(range) => write!(f, "{range}")?,
+        }
+
+        if let Some(default) = self.state_var.default() {
+            write!(f, " = {default}")?;
+        }
+
+        Ok(())
+    }
+}
+
+impl Argument {
+    fn from_xml(node: Node<'_, '_>, state_variables: &[Rc<StateVariable>]) -> Result<Self> {
+        #[allow(non_snake_case)]
+        let (name, direction, related_statevar) =
+            find_in_xml! { node => name, direction, relatedStateVariable };
+
+        let related_statevar = related_statevar
+            .text()
+            .ok_or_else(|| Error::ParseError("`relatedStateVariable` is empty"))?;
+
+        let state_var = state_variables
+            .iter()
+            .find(|sv| sv.name().eq_ignore_ascii_case(related_statevar))
+            .ok_or_else(|| Error::ParseError("found argument without related state variable"))?
+            .clone();
+
+        let direction = direction.text().unwrap_or_default().to_ascii_lowercase();
+        let is_input = match direction.as_str() {
+            "in" => Ok(true),
+            "out" => Ok(false),
+            _ => Err(Error::invalid_response(ParseDirectionErr(direction))),
+        }?;
+
+        Ok(Self {
+            name: utils::parse_node_text(name)?,
+            is_input,
+            state_var,
+        })
+    }
+    pub fn name(&self) -> &str {
+        &self.name
+    }
+
+    pub fn is_input(&self) -> bool {
+        self.is_input
+    }
+    pub fn is_output(&self) -> bool {
+        !self.is_input
+    }
+
+    pub fn related_state_variable(&self) -> &Rc<StateVariable> {
+        &self.state_var
+    }
+}
+
+#[derive(Debug)]
+pub struct ParseDirectionErr(String);
+impl std::fmt::Display for ParseDirectionErr {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "invalid direction: `{}`", &self.0)
+    }
+}
+impl std::error::Error for ParseDirectionErr {}

third-party/rupnp/src/scpd/mod.rs

@@ -0,0 +1,76 @@
+use crate::{
+    find_in_xml,
+    utils::{self, HttpResponseExt, HyperBodyExt},
+    Error,
+};
+use bytes::Bytes;
+use http::Uri;
+use http_body_util::Empty;
+use hyper_util::rt::TokioExecutor;
+use roxmltree::{Document, Node};
+use ssdp_client::URN;
+use std::rc::Rc;
+
+mod action;
+mod state_variable;
+pub use action::*;
+pub use state_variable::*;
+
+/// Service Control Protocol Description.
+/// It contains information about a particular service, more specifically its actions and state
+/// variables.
+#[derive(Debug)]
+pub struct SCPD {
+    urn: URN,
+    state_variables: Vec<Rc<StateVariable>>,
+    actions: Vec<Action>,
+}
+impl SCPD {
+    pub fn urn(&self) -> &URN {
+        &self.urn
+    }
+    pub fn state_variables(&self) -> &[Rc<StateVariable>] {
+        &self.state_variables
+    }
+    pub fn actions(&self) -> &[Action] {
+        &self.actions
+    }
+
+    /// Fetches the SCPD description.
+    /// The `urn` has to be provided because it isn't included in the description.
+    pub(crate) async fn from_url(url: &Uri, urn: URN) -> Result<Self, Error> {
+        let body = hyper_util::client::legacy::Client::builder(TokioExecutor::new())
+            .build_http::<Empty<Bytes>>()
+            .get(url.clone())
+            .await?
+            .err_if_not_200()?
+            .into_body()
+            .bytes()
+            .await?;
+        let body = std::str::from_utf8(&body)?;
+
+        let document = Document::parse(body)?;
+        let scpd = utils::find_root(&document, "scpd", "Service Control Point Definition")?;
+
+        #[allow(non_snake_case)]
+        let (state_variables, actions) = find_in_xml! { scpd => serviceStateTable, actionList };
+
+        let state_variables: Vec<_> = state_variables
+            .children()
+            .filter(Node::is_element)
+            .map(StateVariable::from_xml)
+            .map(|sv| sv.map(Rc::new))
+            .collect::<Result<_, _>>()?;
+        let actions = actions
+            .children()
+            .filter(Node::is_element)
+            .map(|node| Action::from_xml(node, &state_variables))
+            .collect::<Result<_, _>>()?;
+
+        Ok(Self {
+            urn,
+            state_variables,
+            actions,
+        })
+    }
+}

third-party/rupnp/src/scpd/state_variable.rs

@@ -0,0 +1,231 @@
+use crate::{find_in_xml, utils, Error, Result};
+use roxmltree::Node;
+use std::fmt;
+use std::fmt::Formatter;
+
+/// A `StateVariable` is the type of every [Argument](struct.Argument.html) in UPnP Actions.
+/// It is either a single value, an enumeration of strings or an integer range: see
+/// [StateVariableKind](enum.StateVariableKind.html).
+#[derive(Debug)]
+pub struct StateVariable {
+    name: String,
+    default: Option<String>,
+    kind: StateVariableKind,
+    send_events: bool,
+    multicast: bool,
+    optional: bool,
+}
+
+/// The range of a StateVariable
+#[derive(Debug)]
+pub struct StateVariableRange {
+    minimum: String,
+    maximum: String,
+    step: Option<String>,
+}
+
+impl fmt::Display for StateVariableRange {
+    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
+        write!(f, "{}..={}", self.minimum(), self.maximum())?;
+        if let Some(step) = self.step() {
+            write!(f, ":{step}")?;
+        }
+
+        Ok(())
+    }
+}
+
+impl StateVariableRange {
+    pub fn minimum(&self) -> &str {
+        &self.minimum
+    }
+
+    pub fn maximum(&self) -> &str {
+        &self.maximum
+    }
+
+    pub fn step(&self) -> Option<&str> {
+        self.step.as_deref()
+    }
+}
+
+/// The type of a state variable.
+#[derive(Debug)]
+pub enum StateVariableKind {
+    /// Just a value of some datatype
+    Simple(DataType),
+    /// An enumeration of possible strings. Can have a default value.
+    Enum(Vec<String>),
+    /// A Range of values.
+    Range(StateVariableRange),
+}
+
+impl fmt::Display for StateVariable {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "{}", self.name())
+    }
+}
+
+impl StateVariable {
+    pub(crate) fn from_xml(node: Node<'_, '_>) -> Result<Self> {
+        #[allow(non_snake_case)]
+        let (name, datatype, default, variants, range, optional) = find_in_xml! { node => name, dataType, ?defaultValue, ?allowedValueList, ?allowedValueRange, ?optional };
+
+        let variants = variants
+            .map(|a| {
+                a.children()
+                    .filter(Node::is_element)
+                    .map(utils::parse_node_text)
+                    .collect()
+            })
+            .transpose()?;
+
+        let default = default.map(utils::parse_node_text).transpose()?;
+        let range = range.map(range_from_xml).transpose()?;
+
+        let name = utils::parse_node_text(name)?;
+        let datatype = utils::parse_node_text(datatype)?;
+        let optional = optional.is_some();
+
+        let send_events = utils::find_node_attribute(node, "sendEvents")
+            .map(|val| val == "yes")
+            .unwrap_or(true);
+        let multicast = utils::find_node_attribute(node, "multicast")
+            .map(|val| val == "yes")
+            .unwrap_or(false);
+
+        let kind = match (variants, range) {
+            (None, None) => Ok(StateVariableKind::Simple(datatype)),
+            (Some(variants), None) => Ok(StateVariableKind::Enum(variants)),
+            (None, Some(range)) => Ok(StateVariableKind::Range(range)),
+            (Some(_), Some(_)) => Err(Error::ParseError(
+                "both `AllowedValues` and `AllowedValueRange` is set",
+            )),
+        }?;
+
+        Ok(StateVariable {
+            name,
+            kind,
+            default,
+            optional,
+            send_events,
+            multicast,
+        })
+    }
+
+    pub fn name(&self) -> &str {
+        &self.name
+    }
+
+    pub fn default(&self) -> Option<&str> {
+        self.default.as_deref()
+    }
+
+    pub fn optional(&self) -> bool {
+        self.optional
+    }
+
+    pub fn sends_events(&self) -> bool {
+        self.send_events
+    }
+
+    pub fn is_multicast(&self) -> bool {
+        self.multicast
+    }
+
+    pub fn kind(&self) -> &StateVariableKind {
+        &self.kind
+    }
+}
+
+#[derive(Debug, Copy, Clone, Eq, PartialEq)]
+#[allow(non_camel_case_types)]
+pub enum DataType {
+    ui1,
+    ui2,
+    ui4,
+    ui8,
+    i1,
+    i2,
+    i4,
+    int,
+    r4,
+    r8,
+    Number,
+    Float,
+    Fixed14_4,
+    Char,
+    String,
+    Date,
+    DateTime,
+    DateTimeTz,
+    Time,
+    TimeTz,
+    Boolean,
+    BinBase64,
+    BinHex,
+    Uri,
+}
+
+impl fmt::Display for DataType {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        fmt::Debug::fmt(self, f)
+    }
+}
+
+#[derive(Debug)]
+pub struct ParseDataTypeErr(String);
+impl fmt::Display for ParseDataTypeErr {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "invalid data type: {}", self.0)
+    }
+}
+impl std::error::Error for ParseDataTypeErr {}
+impl std::str::FromStr for DataType {
+    type Err = ParseDataTypeErr;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        match s {
+            "ui1" => Ok(DataType::ui1),
+            "ui2" => Ok(DataType::ui2),
+            "ui4" => Ok(DataType::ui4),
+            "ui8" => Ok(DataType::ui8),
+            "i1" => Ok(DataType::i1),
+            "i2" => Ok(DataType::i2),
+            "i4" => Ok(DataType::i4),
+            "int" => Ok(DataType::int),
+            "r4" => Ok(DataType::r4),
+            "r8" => Ok(DataType::r8),
+            "number" => Ok(DataType::Number),
+            "float" => Ok(DataType::Float),
+            "fixed14_4" => Ok(DataType::Fixed14_4),
+            "char" => Ok(DataType::Char),
+            "string" => Ok(DataType::String),
+            "date" => Ok(DataType::Date),
+            "dateTime" => Ok(DataType::DateTime),
+            "dateTimeTz" => Ok(DataType::DateTimeTz),
+            "time" => Ok(DataType::Time),
+            "timeTz" => Ok(DataType::TimeTz),
+            "boolean" => Ok(DataType::Boolean),
+            "bin.base64" => Ok(DataType::BinBase64),
+            "bin.hex" => Ok(DataType::BinHex),
+            "uri" => Ok(DataType::Uri),
+            _ => Err(ParseDataTypeErr(s.to_string())),
+        }
+    }
+}
+
+fn range_from_xml(node: Node<'_, '_>) -> Result<StateVariableRange> {
+    #[allow(non_snake_case)]
+    let (minimum, maximum, step) = find_in_xml! { node => minimum, maximum, ?step };
+
+    let step = step.map(utils::parse_node_text).transpose()?;
+    let minimum = utils::parse_node_text(minimum)?;
+    let maximum = utils::parse_node_text(maximum)?;
+
+    Ok(StateVariableRange {
+        minimum,
+        maximum,
+        step,
+    })
+}

third-party/rupnp/src/service.rs

@@ -0,0 +1,381 @@
+use crate::{
+    error::{Error, UPnPError},
+    find_in_xml,
+    scpd::SCPD,
+    utils::{self, HttpResponseExt},
+    Result,
+};
+
+use bytes::Bytes;
+#[cfg(feature = "subscribe")]
+use futures_core::stream::Stream;
+#[cfg(feature = "subscribe")]
+use genawaiter::sync::{Co, Gen};
+use http_body_util::Empty;
+use hyper_util::{client::legacy::Client, rt::TokioExecutor};
+#[cfg(feature = "subscribe")]
+use tokio::{
+    io::{AsyncBufReadExt, AsyncWriteExt, BufReader},
+    net::TcpListener,
+};
+
+use http::{uri::PathAndQuery, Request, Uri};
+use roxmltree::{Document, Node};
+use ssdp_client::URN;
+
+use std::collections::HashMap;
+use utils::HyperBodyExt;
+
+/// A UPnP Service is the description of endpoints on a device for performing actions and reading
+/// the service definition.
+/// For a list of actions and state variables the service provides, take a look at [`scpd`](struct.Service.html#method.scpd).
+#[derive(Debug, Clone)]
+pub struct Service {
+    service_type: URN,
+    service_id: String,
+    scpd_endpoint: PathAndQuery,
+    control_endpoint: PathAndQuery,
+    event_sub_endpoint: PathAndQuery,
+}
+
+impl Service {
+    pub(crate) fn from_xml(node: Node<'_, '_>) -> Result<Self> {
+        #[allow(non_snake_case)]
+        let (service_type, service_id, scpd_endpoint, control_endpoint, event_sub_endpoint) =
+            find_in_xml! { node => serviceType, serviceId, SCPDURL, controlURL, eventSubURL };
+
+        Ok(Self {
+            service_type: utils::parse_node_text(service_type)?,
+            service_id: utils::parse_node_text(service_id)?,
+            scpd_endpoint: parse_service_endpoint(scpd_endpoint)?,
+            control_endpoint: parse_service_endpoint(control_endpoint)?,
+            event_sub_endpoint: parse_service_endpoint(event_sub_endpoint)?,
+        })
+    }
+
+    /// Returns the [URN](ssdp_client::URN) of this service.
+    pub fn service_type(&self) -> &URN {
+        &self.service_type
+    }
+
+    /// Returns the `Service Identifier`.
+    pub fn service_id(&self) -> &str {
+        &self.service_id
+    }
+
+    pub(crate) fn control_url(&self, url: &Uri) -> Uri {
+        replace_url_path(url, &self.control_endpoint)
+    }
+    pub(crate) fn scpd_url(&self, url: &Uri) -> Uri {
+        replace_url_path(url, &self.scpd_endpoint)
+    }
+    pub(crate) fn event_sub_url(&self, url: &Uri) -> Uri {
+        replace_url_path(url, &self.event_sub_endpoint)
+    }
+
+    /// Fetches the [`SCPD`](scpd/struct.SCPD.html) of this service.
+    pub async fn scpd(&self, url: &Uri) -> Result<SCPD> {
+        SCPD::from_url(&self.scpd_url(url), self.service_type().clone()).await
+    }
+
+    /// Execute some UPnP Action on this service.
+    /// The URL is usually obtained by the device this service was found on.
+    /// The payload is xml-formatted data.
+    ///
+    /// # Example usage:
+    ///
+    /// ```rust,no_run
+    /// # use ssdp_client::URN;
+    /// # async fn rendering_control_example() -> Result<(), rupnp::Error> {
+    /// # let some_url = unimplemented!();
+    /// use rupnp::ssdp::URN;
+    /// use rupnp::Device;
+    ///
+    /// let urn = URN::service("schemas-upnp-org", "RenderingControl", 1);
+    ///
+    /// let device = Device::from_url( some_url ).await?;
+    /// let service = device.find_service(&urn)
+    ///     .expect("service exists");
+    ///
+    /// let args = "<InstanceID>0</InstanceID><Channel>Master</Channel>";
+    /// let response = service.action(device.url(), "GetVolume", args).await?;
+    ///
+    /// let volume = response
+    ///     .get("CurrentVolume")
+    ///     .expect("exists");
+    ///
+    /// println!("Volume: {}", volume);
+    /// # Ok(())
+    /// # }
+    /// ```
+    pub async fn action(
+        &self,
+        url: &Uri,
+        action: &str,
+        payload: &str,
+    ) -> Result<HashMap<String, String>> {
+        let body = format!(
+            r#"
+            <s:Envelope xmlns:s="http://schemas.xmlsoap.org/soap/envelope/"
+                s:encodingStyle="http://schemas.xmlsoap.org/soap/encoding/">
+                <s:Body>
+                    <u:{action} xmlns:u="{service}">
+                        {payload}
+                    </u:{action}>
+                </s:Body>
+            </s:Envelope>"#,
+            service = &self.service_type,
+            action = action,
+            payload = payload
+        );
+
+        let soap_action = format!("\"{}#{}\"", &self.service_type, action);
+
+        let request = Request::post(self.control_url(url))
+            .header("CONTENT-TYPE", "text/xml; charset=\"utf-8\"")
+            .header("SOAPAction", soap_action)
+            .body(body)
+            .expect("infallible");
+        let doc = Client::builder(TokioExecutor::new())
+            .build_http()
+            .request(request)
+            .await?
+            .err_if_not_200()?
+            .into_body()
+            .bytes()
+            .await?;
+        let doc = std::str::from_utf8(&doc)?;
+
+        let document = Document::parse(doc)?;
+        let response = utils::find_root(&document, "Body", "UPnP Response")?
+            .first_element_child()
+            .ok_or_else(|| {
+                Error::XmlMissingElement("Body".to_string(), format!("{action}Response"))
+            })?;
+
+        if response.tag_name().name().eq_ignore_ascii_case("Fault") {
+            return Err(UPnPError::from_fault_node(response)?.into());
+        }
+
+        let values: HashMap<_, _> = response
+            .children()
+            .filter(Node::is_element)
+            .filter_map(|node| -> Option<(String, String)> {
+                node.text()
+                    .map(|text| (node.tag_name().name().to_string(), text.to_string()))
+            })
+            .collect();
+
+        Ok(values)
+    }
+
+    #[cfg(feature = "subscribe")]
+    async fn make_subscribe_request(
+        &self,
+        url: &Uri,
+        callback: &str,
+        timeout_secs: u32,
+    ) -> Result<String> {
+        use bytes::Bytes;
+        use http_body_util::Empty;
+
+        let req = Request::builder()
+            .uri(self.event_sub_url(url))
+            .method("SUBSCRIBE")
+            .header("CALLBACK", format!("<{callback}>"))
+            .header("NT", "upnp:event")
+            .header("TIMEOUT", format!("Second-{timeout_secs}"))
+            .body(Empty::<Bytes>::new())
+            .expect("infallible");
+
+        let response = Client::builder(TokioExecutor::new())
+            .build_http()
+            .request(req)
+            .await?
+            .err_if_not_200()?;
+
+        let sid = response
+            .headers()
+            .get("sid")
+            .ok_or_else(|| Error::ParseError("missing http header `SID`"))?
+            .to_str()
+            .map_err(|_| Error::ParseError("SID header contained non-visible ASCII bytes"))?
+            .to_string();
+
+        Ok(sid)
+    }
+
+    /// Subscribe for state variable changes.
+    ///
+    /// It returns the SID which can be used to unsubscribe to the service and a stream of
+    /// responses.
+    ///
+    /// Each response is a [HashMap](std::collections::HashMap) of the state variables.
+    ///
+    /// # Example usage:
+    /// ```rust,no_run
+    /// # use futures::prelude::*;
+    /// # async fn subscribe_example() -> Result<(), rupnp::Error> {
+    /// # let device: rupnp::Device = unimplemented!();
+    /// # let service: rupnp::Service = unimplemented!();
+    /// let (_sid, stream) = service.subscribe(device.url(), 300).await?;
+    ///
+    /// while let Some(state_vars) = stream.try_next().await? {
+    ///     for (key, value) in state_vars {
+    ///         println!("{} => {}", key, value);
+    ///     }
+    /// }
+    /// # Ok(())
+    /// # }
+    /// ```
+    #[cfg(feature = "subscribe")]
+    pub async fn subscribe(
+        &self,
+        url: &Uri,
+        timeout_secs: u32,
+    ) -> Result<(String, impl Stream<Item = Result<HashMap<String, String>>>)> {
+        let addr = utils::get_local_addr()?;
+        let listener = TcpListener::bind(addr).await?;
+
+        let addr = format!("http://{}", listener.local_addr()?);
+
+        let sid = self
+            .make_subscribe_request(url, &addr, timeout_secs)
+            .await?;
+
+        let stream = Gen::new(move |co: Co<Result<_>>| subscribe_stream(listener, co));
+
+        Ok((sid, stream))
+    }
+
+    /// Renew a subscription made with the [subscribe](struct.Service.html#method.subscribe) method.
+    ///
+    /// When the sid is invalid, the control point will respond with a `412 Preconditition failed`.
+    pub async fn renew_subscription(&self, url: &Uri, sid: &str, timeout_secs: u32) -> Result<()> {
+        let req = Request::builder()
+            .uri(self.event_sub_url(url))
+            .method("SUBSCRIBE")
+            .header("SID", sid)
+            .header("TIMEOUT", format!("Second-{timeout_secs}"))
+            .body(Empty::<Bytes>::new())
+            .expect("infallible");
+        Client::builder(TokioExecutor::new())
+            .build_http()
+            .request(req)
+            .await?
+            .err_if_not_200()?;
+
+        Ok(())
+    }
+
+    /// Unsubscribe from further event notifications.
+    ///
+    /// The SID is usually obtained by the [subscribe](struct.Service.html#method.subscribe) method.
+    ///
+    /// When the sid is invalid, the control point will respond with a `412 Preconditition failed`.
+    pub async fn unsubscribe(&self, url: &Uri, sid: &str) -> Result<()> {
+        let req = Request::builder()
+            .uri(self.event_sub_url(url))
+            .method("UNSUBSCRIBE")
+            .header("SID", sid)
+            .body(Empty::<Bytes>::new())
+            .expect("infallible");
+
+        Client::builder(TokioExecutor::new())
+            .build_http()
+            .request(req)
+            .await?
+            .err_if_not_200()?;
+
+        Ok(())
+    }
+}
+
+#[cfg(feature = "subscribe")]
+macro_rules! yield_try {
+    ( $co:expr => $expr:expr ) => {
+        match $expr {
+            Ok(val) => val,
+            Err(e) => {
+                $co.yield_(Err(e.into())).await;
+                continue;
+            }
+        }
+    };
+}
+
+#[cfg(feature = "subscribe")]
+fn propertyset_to_map(input: &str) -> Result<HashMap<String, String>, roxmltree::Error> {
+    let doc = Document::parse(input)?;
+    let hashmap: HashMap<String, String> = doc
+        .root_element() // <e:propertyset />
+        .children() // <e:property />
+        .filter_map(|child| child.first_element_child()) // actual tag
+        .filter_map(|node| {
+            node.text()
+                .map(|text| (node.tag_name().name().to_string(), text.to_string()))
+        })
+        .collect();
+
+    Ok(hashmap)
+}
+
+#[cfg(feature = "subscribe")]
+async fn subscribe_stream(listener: TcpListener, co: Co<Result<HashMap<String, String>>>) {
+    loop {
+        let (mut stream, _) = yield_try!(co => listener.accept().await);
+        let mut lines = BufReader::new(&mut stream).lines();
+
+        let mut input = String::new();
+        let mut is_xml = false;
+
+        // sometimes the xml is on one line, sometimes on multiple ones.
+        // we dont care about the http stuff before the "<e:propertyset>"
+        while let Ok(Some(line)) = lines.next_line().await {
+            if is_xml || line.starts_with("<e:propertyset") {
+                input.push_str(&line);
+                is_xml = true;
+            }
+
+            if line.ends_with("</e:propertyset>") {
+                break;
+            };
+        }
+
+        let response = "HTTP/1.1 200 OK\r\n\r\n";
+        let _ = stream.write_all(response.as_bytes()).await;
+
+        let hashmap = yield_try!(co => propertyset_to_map(&input));
+
+        co.yield_(Ok(hashmap)).await;
+    }
+}
+
+fn replace_url_path(url: &Uri, path: &PathAndQuery) -> Uri {
+    let mut parts = url.clone().into_parts();
+    parts.path_and_query = Some(path.clone());
+    Uri::from_parts(parts).expect("infallible")
+}
+
+// QBZ #745: preserve the wire paths accepted by http 1.4.0 without downgrading
+// http for the whole application. Valid PathAndQuery values take the original
+// parser unchanged. Empty paths previously represented `/`. http 1.4.0 sent
+// relative endpoints without a leading slash even in the HTTP request target;
+// http 1.5.0 rejects them entirely. Normalize only those rejected endpoints to
+// valid origin-form paths. Do not change URLBase/description-directory handling
+// or any endpoint already accepted by the released 2.1.0 parser.
+fn parse_service_endpoint(node: Node<'_, '_>) -> Result<PathAndQuery> {
+    let text = node.text().unwrap_or("");
+    if text.is_empty() {
+        return Ok(PathAndQuery::from_static("/"));
+    }
+    match text.parse::<PathAndQuery>() {
+        Ok(path) => Ok(path),
+        Err(error) if !text.starts_with('/') => {
+            // Re-parse after adding only the missing slash: invalid characters
+            // and the http crate's size limit remain enforced.
+            format!("/{text}").parse().map_err(|_| Error::invalid_response(error))
+        }
+        Err(error) => Err(Error::invalid_response(error)),
+    }
+}

third-party/rupnp/src/utils.rs

@@ -0,0 +1,193 @@
+use crate::{Error, Result};
+use http_body_util::BodyExt;
+#[cfg(feature = "subscribe")]
+use if_addrs::{get_if_addrs, Interface};
+use roxmltree::{Document, Node};
+#[cfg(feature = "subscribe")]
+use std::net::{IpAddr, SocketAddrV4};
+
+pub(crate) trait HttpResponseExt: Sized {
+    fn err_if_not_200(self) -> Result<Self>;
+}
+impl<B> HttpResponseExt for hyper::Response<B> {
+    fn err_if_not_200(self) -> Result<Self> {
+        if self.status() != 200 {
+            Err(Error::HttpErrorCode(self.status()))
+        } else {
+            Ok(self)
+        }
+    }
+}
+pub(crate) trait HyperBodyExt: Sized {
+    async fn bytes(self) -> Result<bytes::Bytes>;
+}
+impl HyperBodyExt for hyper::body::Incoming {
+    async fn bytes(self) -> Result<bytes::Bytes> {
+        Ok(self.collect().await?.to_bytes())
+    }
+}
+
+#[macro_export]
+#[doc(hidden)]
+macro_rules! find_in_xml {
+    ( $node:expr => $( $($var:ident)? $(?$var_opt:ident)? ),+ $(#$var_hash_opt:ident)? ) => { {
+        let node = $node;
+        $(
+            $(let mut $var = None;)?
+            $(let mut $var_opt = None;)?
+        )*
+        $(
+            let mut $var_hash_opt: HashMap<String, Option<String>> = $var_hash_opt
+                .iter()
+                .map(|k| (k.to_string(), None))
+                .collect();
+        )?
+        for child in node.children().filter(roxmltree::Node::is_element) {
+            match child.tag_name().name() {
+                $(
+                    $(stringify!($var) => $var = Some(child),)?
+                    $(stringify!($var_opt) => $var_opt = Some(child),)?
+                )*
+                _ => (),
+            }
+            $(
+                if $var_hash_opt.contains_key(child.tag_name().name()) {
+                    $var_hash_opt.insert(
+                        child.tag_name().name().to_string(),
+                        $crate::utils::parse_node_text(child).ok());
+                }
+            )?
+        }
+
+        $($(
+            let $var = $var.ok_or_else(|| $crate::Error::XmlMissingElement(
+                node.tag_name().name().to_string(),
+                stringify!($var).to_string(),
+            ))?;
+        )?)*
+
+        (
+            $(
+                $($var)?
+                $($var_opt)?
+            ),*
+            $( $var_hash_opt )?
+        )
+    } }
+}
+
+pub fn parse_node_text<T, E>(node: Node<'_, '_>) -> Result<T>
+where
+    T: std::str::FromStr<Err = E>,
+    E: std::error::Error + Send + Sync + 'static,
+{
+    node.text()
+        .unwrap_or_default()
+        .parse()
+        .map_err(Error::invalid_response)
+}
+
+pub fn find_root<'a, 'input: 'a>(
+    document: &'input Document<'_>,
+    element: &str,
+    docname: &str,
+) -> Result<Node<'a, 'input>> {
+    document
+        .descendants()
+        .filter(Node::is_element)
+        .find(|n| n.tag_name().name().eq_ignore_ascii_case(element))
+        .ok_or_else(|| Error::XmlMissingElement(docname.to_string(), element.to_string()))
+}
+
+pub fn find_node_attribute<'n, 'd: 'n>(node: Node<'d, 'n>, attr: &str) -> Option<&'n str> {
+    node.attributes()
+        .find(|a| a.name().eq_ignore_ascii_case(attr))
+        .map(|a| a.value())
+}
+
+#[cfg(feature = "subscribe")]
+pub fn get_local_addr() -> Result<SocketAddrV4> {
+    get_if_addrs()?
+        .iter()
+        .map(Interface::ip)
+        .filter_map(|addr| match addr {
+            IpAddr::V4(addr) => Some(addr),
+            IpAddr::V6(_) => None,
+        })
+        .find(|x| x.is_private())
+        .ok_or(Error::NoLocalInterfaceOpen)
+        .map(|addr| SocketAddrV4::new(addr, 0))
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::find_in_xml;
+    use roxmltree::Document;
+    use std::collections::HashMap;
+
+    #[test]
+    fn test_find_in_xml_macro() -> Result<(), Box<dyn std::error::Error>> {
+        let xml = r#"
+        <device>
+            <deviceType>urn:schemas-upnp-org:device:MediaServer:1</deviceType>
+            <friendlyName>My Media Server</friendlyName>
+            <manufacturer>ACME Corp</manufacturer>
+            <modelName>MediaBox 3000</modelName>
+            <serviceList>
+                <service>
+                    <serviceType>urn:schemas-upnp-org:service:ContentDirectory:1</serviceType>
+                </service>
+            </serviceList>
+            <deviceList>
+                <device>
+                    <deviceType>urn:schemas-upnp-org:device:MediaRenderer:1</deviceType>
+                </device>
+            </deviceList>
+            <extraInfo>Some extra information</extraInfo>
+        </device>
+        "#;
+
+        let doc = Document::parse(xml).unwrap();
+        let device_node = doc.root_element();
+
+        let extra_element_keys = &["extraInfo"];
+
+        #[rustfmt::skip]
+        #[allow(non_snake_case)]
+        let (device_type, friendly_name, services, devices, extra_elements) = 
+            find_in_xml! { device_node => deviceType, friendlyName, ?serviceList, ?deviceList, #extra_element_keys };
+
+        // Test required elements
+        assert_eq!(
+            device_type.text(),
+            Some("urn:schemas-upnp-org:device:MediaServer:1")
+        );
+        assert_eq!(friendly_name.text(), Some("My Media Server"));
+
+        // Test optional elements
+        assert!(services.is_some());
+        assert!(devices.is_some());
+
+        // Test extra elements
+        let mut expected_extra = HashMap::new();
+        expected_extra.insert(
+            "extraInfo".to_string(),
+            Some("Some extra information".to_string()),
+        );
+        assert_eq!(extra_elements, expected_extra);
+
+        // Test non-existent optional element
+        let extra_element_keys = &["nonExistentElement"];
+
+        #[rustfmt::skip]
+        #[allow(non_snake_case)]
+        let (_, _, _, _, missing_extra) =
+            find_in_xml! { device_node => deviceType, friendlyName, ?serviceList, ?deviceList, #extra_element_keys };
+
+        let mut expected_missing = HashMap::new();
+        expected_missing.insert("nonExistentElement".to_string(), None);
+        assert_eq!(missing_extra, expected_missing);
+
+        Ok(())
+    }
+}