mirrors/gst-plugins-rs
mirrors/gst-plugins-rs
1
1
Fork 0
mirror of https://gitlab.freedesktop.org/gstreamer/gst-plugins-rs.git synced 2024-11-26 13:31:00 +00:00
Code Issues Projects Releases Wiki Activity

reqwest: Update to reqwest 0.12

Browse source 
Part-of: <https://gitlab.freedesktop.org/gstreamer/gst-plugins-rs/-/merge_requests/1510>
This commit is contained in:
Sebastian Dröge 2024-03-20 20:40:24 +02:00
parent 7e1ab086de
commit f97150aa58
3 changed files with 461 additions and 175 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

258
Cargo.lock generated
View file

@ -337,7 +337,7 @@ dependencies = [
"fastrand", "fastrand",
"hex", "hex",
"http 0.2.12", "http 0.2.12",
"hyper", "hyper 0.14.28",
"ring", "ring",
"time", "time",
"tokio", "tokio",
@ -374,7 +374,7 @@ dependencies = [
"bytes", "bytes",
"fastrand", "fastrand",
"http 0.2.12", "http 0.2.12",
"http-body", "http-body 0.4.6",
"percent-encoding", "percent-encoding",
"pin-project-lite", "pin-project-lite",
"tracing", "tracing",
@ -450,7 +450,7 @@ dependencies = [
"hex", "hex",
"hmac 0.12.1", "hmac 0.12.1",
"http 0.2.12", "http 0.2.12",
"http-body", "http-body 0.4.6",
"lru", "lru",
"once_cell", "once_cell",
"percent-encoding", "percent-encoding",
@ -546,7 +546,7 @@ dependencies = [
"aws-types", "aws-types",
"bytes", "bytes",
"http 0.2.12", "http 0.2.12",
"hyper", "hyper 0.14.28",
"once_cell", "once_cell",
"regex-lite", "regex-lite",
"tracing", "tracing",
@ -628,7 +628,7 @@ dependencies = [
"crc32fast", "crc32fast",
"hex", "hex",
"http 0.2.12", "http 0.2.12",
"http-body", "http-body 0.4.6",
"md-5", "md-5",
"pin-project-lite", "pin-project-lite",
"sha1", "sha1",
@ -660,7 +660,7 @@ dependencies = [
"bytes-utils", "bytes-utils",
"futures-core", "futures-core",
"http 0.2.12", "http 0.2.12",
"http-body", "http-body 0.4.6",
"once_cell", "once_cell",
"percent-encoding", "percent-encoding",
"pin-project-lite", "pin-project-lite",
@ -699,10 +699,10 @@ dependencies = [
"aws-smithy-types", "aws-smithy-types",
"bytes", "bytes",
"fastrand", "fastrand",
"h2", "h2 0.3.25",
"http 0.2.12", "http 0.2.12",
"http-body", "http-body 0.4.6",
"hyper", "hyper 0.14.28",
"hyper-rustls", "hyper-rustls",
"once_cell", "once_cell",
"pin-project-lite", "pin-project-lite",
@ -740,7 +740,7 @@ dependencies = [
"bytes-utils", "bytes-utils",
"futures-core", "futures-core",
"http 0.2.12", "http 0.2.12",
"http-body", "http-body 0.4.6",
"itoa", "itoa",
"num-integer", "num-integer",
"pin-project-lite", "pin-project-lite",
@ -2637,15 +2637,18 @@ dependencies = [
name = "gst-plugin-reqwest" name = "gst-plugin-reqwest"
version = "0.13.0-alpha.1" version = "0.13.0-alpha.1"
dependencies = [ dependencies = [
"bytes",
"futures", "futures",
"gst-plugin-version-helper", "gst-plugin-version-helper",
"gstreamer", "gstreamer",
"gstreamer-base", "gstreamer-base",
"headers", "headers 0.4.0",
"hyper", "http-body-util",
"hyper 1.2.0",
"mime", "mime",
"once_cell", "once_cell",
"reqwest", "pin-project-lite",
"reqwest 0.12.1",
"tokio", "tokio",
"url", "url",
] ]
@ -2917,7 +2920,7 @@ dependencies = [
"parse_link_header", "parse_link_header",
"rand", "rand",
"regex", "regex",
"reqwest", "reqwest 0.11.26",
"serde", "serde",
"serde_json", "serde_json",
"thiserror", "thiserror",
@ -2976,7 +2979,7 @@ dependencies = [
"gstreamer-webrtc", "gstreamer-webrtc",
"once_cell", "once_cell",
"parse_link_header", "parse_link_header",
"reqwest", "reqwest 0.11.26",
"tokio", "tokio",
] ]
@ -3446,6 +3449,25 @@ dependencies = [
"tracing", "tracing",
] ]
[[package]]
name = "h2"
version = "0.4.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "51ee2dd2e4f378392eeff5d51618cd9a63166a2513846bbc55f21cfacd9199d4"
dependencies = [
"bytes",
"fnv",
"futures-core",
"futures-sink",
"futures-util",
"http 1.1.0",
"indexmap 2.2.5",
"slab",
"tokio",
"tokio-util",
"tracing",
]
[[package]] [[package]]
name = "hashbrown" name = "hashbrown"
version = "0.12.3" version = "0.12.3"
@ -3470,13 +3492,28 @@ checksum = "06683b93020a07e3dbcf5f8c0f6d40080d725bea7936fc01ad345c01b97dc270"
dependencies = [ dependencies = [
"base64 0.21.7", "base64 0.21.7",
"bytes", "bytes",
"headers-core", "headers-core 0.2.0",
"http 0.2.12", "http 0.2.12",
"httpdate", "httpdate",
"mime", "mime",
"sha1", "sha1",
] ]
[[package]]
name = "headers"
version = "0.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "322106e6bd0cba2d5ead589ddb8150a13d7c4217cf80d7c4f682ca994ccc6aa9"
dependencies = [
"base64 0.21.7",
"bytes",
"headers-core 0.3.0",
"http 1.1.0",
"httpdate",
"mime",
"sha1",
]
[[package]] [[package]]
name = "headers-core" name = "headers-core"
version = "0.2.0" version = "0.2.0"
@ -3486,6 +3523,15 @@ dependencies = [
"http 0.2.12", "http 0.2.12",
] ]
[[package]]
name = "headers-core"
version = "0.3.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "54b4a22553d4242c49fddb9ba998a99962b5cc6f22cb5a3482bec22522403ce4"
dependencies = [
"http 1.1.0",
]
[[package]] [[package]]
name = "heck" name = "heck"
version = "0.4.1" version = "0.4.1"
@ -3602,6 +3648,29 @@ dependencies = [
"pin-project-lite", "pin-project-lite",
] ]
[[package]]
name = "http-body"
version = "1.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1cac85db508abc24a2e48553ba12a996e87244a0395ce011e62b37158745d643"
dependencies = [
"bytes",
"http 1.1.0",
]
[[package]]
name = "http-body-util"
version = "0.1.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0475f8b2ac86659c21b64320d5d653f9efe42acd2a4e560073ec61a155a34f1d"
dependencies = [
"bytes",
"futures-core",
"http 1.1.0",
"http-body 1.0.0",
"pin-project-lite",
]
[[package]] [[package]]
name = "httparse" name = "httparse"
version = "1.8.0" version = "1.8.0"
@ -3636,20 +3705,41 @@ dependencies = [
"futures-channel", "futures-channel",
"futures-core", "futures-core",
"futures-util", "futures-util",
"h2", "h2 0.3.25",
"http 0.2.12", "http 0.2.12",
"http-body", "http-body 0.4.6",
"httparse", "httparse",
"httpdate", "httpdate",
"itoa", "itoa",
"pin-project-lite", "pin-project-lite",
"socket2 0.5.6", "socket2 0.4.10",
"tokio", "tokio",
"tower-service", "tower-service",
"tracing", "tracing",
"want", "want",
] ]
[[package]]
name = "hyper"
version = "1.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "186548d73ac615b32a73aafe38fb4f56c0d340e110e5a200bcadbaf2e199263a"
dependencies = [
"bytes",
"futures-channel",
"futures-util",
"h2 0.4.3",
"http 1.1.0",
"http-body 1.0.0",
"httparse",
"httpdate",
"itoa",
"pin-project-lite",
"smallvec",
"tokio",
"want",
]
[[package]] [[package]]
name = "hyper-proxy" name = "hyper-proxy"
version = "0.9.1" version = "0.9.1"
@ -3658,9 +3748,9 @@ checksum = "ca815a891b24fdfb243fa3239c86154392b0953ee584aa1a2a1f66d20cbe75cc"
dependencies = [ dependencies = [
"bytes", "bytes",
"futures", "futures",
"headers", "headers 0.3.9",
"http 0.2.12", "http 0.2.12",
"hyper", "hyper 0.14.28",
"tokio", "tokio",
"tower-service", "tower-service",
] ]
@ -3673,7 +3763,7 @@ checksum = "ec3efd23720e2049821a693cbc7e65ea87c72f1c58ff2f9522ff332b1491e590"
dependencies = [ dependencies = [
"futures-util", "futures-util",
"http 0.2.12", "http 0.2.12",
"hyper", "hyper 0.14.28",
"log", "log",
"rustls", "rustls",
"rustls-native-certs", "rustls-native-certs",
@ -3688,12 +3778,48 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d6183ddfa99b85da61a140bea0efc93fdf56ceaa041b37d553518030827f9905" checksum = "d6183ddfa99b85da61a140bea0efc93fdf56ceaa041b37d553518030827f9905"
dependencies = [ dependencies = [
"bytes", "bytes",
"hyper", "hyper 0.14.28",
"native-tls", "native-tls",
"tokio", "tokio",
"tokio-native-tls", "tokio-native-tls",
] ]
[[package]]
name = "hyper-tls"
version = "0.6.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "70206fc6890eaca9fde8a0bf71caa2ddfc9fe045ac9e5c70df101a7dbde866e0"
dependencies = [
"bytes",
"http-body-util",
"hyper 1.2.0",
"hyper-util",
"native-tls",
"tokio",
"tokio-native-tls",
"tower-service",
]
[[package]]
name = "hyper-util"
version = "0.1.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ca38ef113da30126bbff9cd1705f9273e15d45498615d138b0c20279ac7a76aa"
dependencies = [
"bytes",
"futures-channel",
"futures-util",
"http 1.1.0",
"http-body 1.0.0",
"hyper 1.2.0",
"pin-project-lite",
"socket2 0.5.6",
"tokio",
"tower",
"tower-service",
"tracing",
]
[[package]] [[package]]
name = "hyphenation" name = "hyphenation"
version = "0.8.4" version = "0.8.4"
@ -4011,7 +4137,7 @@ dependencies = [
"futures-util", "futures-util",
"getopts", "getopts",
"hex", "hex",
"hyper", "hyper 0.14.28",
"librespot-audio", "librespot-audio",
"librespot-connect", "librespot-connect",
"librespot-core", "librespot-core",
@ -4080,7 +4206,7 @@ dependencies = [
"hmac 0.11.0", "hmac 0.11.0",
"http 0.2.12", "http 0.2.12",
"httparse", "httparse",
"hyper", "hyper 0.14.28",
"hyper-proxy", "hyper-proxy",
"librespot-protocol", "librespot-protocol",
"log", "log",
@ -4116,7 +4242,7 @@ dependencies = [
"form_urlencoded", "form_urlencoded",
"futures-core", "futures-core",
"hmac 0.11.0", "hmac 0.11.0",
"hyper", "hyper 0.14.28",
"libmdns", "libmdns",
"librespot-core", "librespot-core",
"log", "log",
@ -4219,7 +4345,7 @@ dependencies = [
"log", "log",
"parking_lot", "parking_lot",
"prost", "prost",
"reqwest", "reqwest 0.11.26",
"scopeguard", "scopeguard",
"serde", "serde",
"sha2", "sha2",
@ -5340,6 +5466,46 @@ name = "reqwest"
version = "0.11.26" version = "0.11.26"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "78bf93c4af7a8bb7d879d51cebe797356ff10ae8516ace542b5182d9dcac10b2" checksum = "78bf93c4af7a8bb7d879d51cebe797356ff10ae8516ace542b5182d9dcac10b2"
dependencies = [
"base64 0.21.7",
"bytes",
"encoding_rs",
"futures-core",
"futures-util",
"h2 0.3.25",
"http 0.2.12",
"http-body 0.4.6",
"hyper 0.14.28",
"hyper-tls 0.5.0",
"ipnet",
"js-sys",
"log",
"mime",
"native-tls",
"once_cell",
"percent-encoding",
"pin-project-lite",
"rustls-pemfile",
"serde",
"serde_json",
"serde_urlencoded",
"sync_wrapper",
"system-configuration",
"tokio",
"tokio-native-tls",
"tower-service",
"url",
"wasm-bindgen",
"wasm-bindgen-futures",
"web-sys",
"winreg",
]
[[package]]
name = "reqwest"
version = "0.12.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e333b1eb9fe677f6893a9efcb0d277a2d3edd83f358a236b657c32301dc6e5f6"
dependencies = [ dependencies = [
"async-compression", "async-compression",
"base64 0.21.7", "base64 0.21.7",
@ -5349,11 +5515,13 @@ dependencies = [
"encoding_rs", "encoding_rs",
"futures-core", "futures-core",
"futures-util", "futures-util",
"h2", "h2 0.4.3",
"http 0.2.12", "http 1.1.0",
"http-body", "http-body 1.0.0",
"hyper", "http-body-util",
"hyper-tls", "hyper 1.2.0",
"hyper-tls 0.6.0",
"hyper-util",
"ipnet", "ipnet",
"js-sys", "js-sys",
"log", "log",
@ -6334,6 +6502,28 @@ dependencies = [
"winnow 0.6.5", "winnow 0.6.5",
] ]
[[package]]
name = "tower"
version = "0.4.13"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b8fa9be0de6cf49e536ce1851f987bd21a43b771b09473c3549a6c853db37c1c"
dependencies = [
"futures-core",
"futures-util",
"pin-project",
"pin-project-lite",
"tokio",
"tower-layer",
"tower-service",
"tracing",
]
[[package]]
name = "tower-layer"
version = "0.3.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c20c8dbed6283a09604c3e69b4b7eeb54e298b8a600d4d5ecb5ad39de609f1d0"
[[package]] [[package]]
name = "tower-service" name = "tower-service"
version = "0.3.2" version = "0.3.2"
@ -6651,9 +6841,9 @@ dependencies = [
"bytes", "bytes",
"futures-channel", "futures-channel",
"futures-util", "futures-util",
"headers", "headers 0.3.9",
"http 0.2.12", "http 0.2.12",
"hyper", "hyper 0.14.28",
"log", "log",
"mime", "mime",
"mime_guess", "mime_guess",

9
net/reqwest/Cargo.toml
View file

@ -10,9 +10,9 @@ rust-version.workspace = true
[dependencies] [dependencies]
url = "2.1" url = "2.1"
reqwest = { version = "0.11", features = ["cookies", "gzip"] } reqwest = { version = "0.12", features = ["cookies", "gzip"] }
futures = "0.3" futures = "0.3"
headers = "0.3" headers = "0.4"
mime = "0.3" mime = "0.3"
gst.workspace = true gst.workspace = true
gst-base.workspace = true gst-base.workspace = true
@ -20,7 +20,10 @@ tokio = { version = "1.0", default-features = false, features = ["time", "rt-mul
once_cell.workspace = true once_cell.workspace = true
[dev-dependencies] [dev-dependencies]
hyper = { version = "0.14", features = ["server"] } hyper = { version = "1.0", features = ["server"] }
http-body-util = "0.1.1"
bytes = "1.0"
pin-project-lite = "0.2"
gst.workspace = true gst.workspace = true
[lib] [lib]

331
net/reqwest/tests/reqwesthttpsrc.rs
View file

@ -10,8 +10,9 @@
#![allow(clippy::single_match)] #![allow(clippy::single_match)]
use gst::glib; use gst::{glib, prelude::*};
use gst::prelude::*; use http_body_util::combinators::BoxBody;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt};
use std::sync::mpsc; use std::sync::mpsc;
@ -33,7 +34,7 @@ struct Harness {
src: gst::Element, src: gst::Element,
pad: gst::Pad, pad: gst::Pad,
receiver: Option<mpsc::Receiver<Message>>, receiver: Option<mpsc::Receiver<Message>>,
rt: Option<tokio::runtime::Runtime>, rt: tokio::runtime::Runtime,
} }
/// Messages sent from our test harness /// Messages sent from our test harness
@ -46,20 +47,34 @@ enum Message {
ServerError(String), ServerError(String),
} }
fn full_body(s: impl Into<bytes::Bytes>) -> BoxBody<bytes::Bytes, hyper::Error> {
use http_body_util::{BodyExt, Full};
Full::new(s.into()).map_err(|never| match never {}).boxed()
}
fn empty_body() -> BoxBody<bytes::Bytes, hyper::Error> {
use http_body_util::{BodyExt, Empty};
Empty::new().map_err(|never| match never {}).boxed()
}
impl Harness { impl Harness {
/// Creates a new HTTP source and test harness around it /// Creates a new HTTP source and test harness around it
/// ///
/// `http_func`: Function to generate HTTP responses based on a request /// `http_func`: Function to generate HTTP responses based on a request
/// `setup_func`: Setup function for the HTTP source, should only set properties and similar /// `setup_func`: Setup function for the HTTP source, should only set properties and similar
fn new< fn new<
F: FnMut(hyper::Request<hyper::Body>) -> hyper::Response<hyper::Body> + Send + 'static, F: FnMut(
hyper::Request<hyper::body::Incoming>,
) -> hyper::Response<BoxBody<bytes::Bytes, hyper::Error>>
+ Send
+ 'static,
G: FnOnce(&gst::Element), G: FnOnce(&gst::Element),
>( >(
http_func: F, http_func: F,
setup_func: G, setup_func: G,
) -> Harness { ) -> Harness {
use hyper::service::{make_service_fn, service_fn}; use hyper::server::conn::http1;
use hyper::Server; use hyper::service::service_fn;
use std::sync::{Arc, Mutex}; use std::sync::{Arc, Mutex};
// Create the HTTP source // Create the HTTP source
@ -112,21 +127,15 @@ impl Harness {
.unwrap(); .unwrap();
// Create an HTTP sever that listens on localhost on some random, free port // Create an HTTP sever that listens on localhost on some random, free port
let addr = ([127, 0, 0, 1], 0).into(); let addr = std::net::SocketAddr::from(([127, 0, 0, 1], 0));
// Whenever a new client is connecting, a new service function is requested. For each // Whenever a new client is connecting, a new service function is requested. For each
// client we use the same service function, which simply calls the function used by the // client we use the same service function, which simply calls the function used by the
// test // test
let http_func = Arc::new(Mutex::new(http_func)); let http_func = Arc::new(Mutex::new(http_func));
let make_service = make_service_fn(move |_ctx| { let service = service_fn(move |req: hyper::Request<hyper::body::Incoming>| {
let http_func = http_func.clone();
async move {
let http_func = http_func.clone();
Ok::<_, hyper::Error>(service_fn(move |req| {
let http_func = http_func.clone(); let http_func = http_func.clone();
async move { Ok::<_, hyper::Error>((*http_func.lock().unwrap())(req)) } async move { Ok::<_, hyper::Error>((*http_func.lock().unwrap())(req)) }
}))
}
}); });
let (local_addr_sender, local_addr_receiver) = tokio::sync::oneshot::channel(); let (local_addr_sender, local_addr_receiver) = tokio::sync::oneshot::channel();
@ -135,13 +144,22 @@ impl Harness {
rt.spawn(async move { rt.spawn(async move {
// Bind the server, retrieve the local port that was selected in the end and set this as // Bind the server, retrieve the local port that was selected in the end and set this as
// the location property on the source // the location property on the source
let server = Server::bind(&addr).serve(make_service); let listener = tokio::net::TcpListener::bind(addr).await.unwrap();
let local_addr = server.local_addr(); let local_addr = listener.local_addr().unwrap();
local_addr_sender.send(local_addr).unwrap(); local_addr_sender.send(local_addr).unwrap();
if let Err(e) = server.await { loop {
let _ = sender.send(Message::ServerError(format!("{e:?}"))); let (stream, _) = listener.accept().await.unwrap();
let io = tokio_io::TokioIo::new(stream);
let service = service.clone();
let sender = sender.clone();
tokio::task::spawn(async move {
let http = http1::Builder::new().serve_connection(io, service);
if let Err(e) = http.await {
let _ = sender.send(Message::ServerError(format!("{e}")));
}
});
} }
}); });
@ -155,7 +173,7 @@ impl Harness {
src, src,
pad, pad,
receiver: Some(receiver), receiver: Some(receiver),
rt: Some(rt), rt,
} }
} }
@ -337,28 +355,25 @@ impl Drop for Harness {
self.pad.set_active(false).unwrap(); self.pad.set_active(false).unwrap();
self.src.set_state(gst::State::Null).unwrap(); self.src.set_state(gst::State::Null).unwrap();
self.rt.take().unwrap();
} }
} }
#[test] #[test]
fn test_basic_request() { fn test_basic_request() {
use std::io::{Cursor, Read}; use std::io::{Cursor, Read};
init(); init();
// Set up a harness that returns "Hello World" for any HTTP request and checks if the // Set up a harness that returns "Hello World" for any HTTP request and checks if the
// default headers are all sent // default headers are all sent
let mut h = Harness::new( let mut h = Harness::new(
|req| { |req| {
use hyper::{Body, Response};
let headers = req.headers(); let headers = req.headers();
assert_eq!(headers.get("connection").unwrap(), "keep-alive"); assert_eq!(headers.get("connection").unwrap(), "keep-alive");
assert_eq!(headers.get("accept-encoding").unwrap(), "identity"); assert_eq!(headers.get("accept-encoding").unwrap(), "identity");
assert_eq!(headers.get("icy-metadata").unwrap(), "1"); assert_eq!(headers.get("icy-metadata").unwrap(), "1");
Response::new(Body::from("Hello World")) hyper::Response::new(full_body("Hello World"))
}, },
|_src| { |_src| {
// No additional setup needed here // No additional setup needed here
@ -399,21 +414,20 @@ fn test_basic_request() {
#[test] #[test]
fn test_basic_request_inverted_defaults() { fn test_basic_request_inverted_defaults() {
use std::io::{Cursor, Read}; use std::io::{Cursor, Read};
init(); init();
// Set up a harness that returns "Hello World" for any HTTP request and override various // Set up a harness that returns "Hello World" for any HTTP request and override various
// default properties to check if the corresponding headers are set correctly // default properties to check if the corresponding headers are set correctly
let mut h = Harness::new( let mut h = Harness::new(
|req| { |req| {
use hyper::{Body, Response};
let headers = req.headers(); let headers = req.headers();
assert_eq!(headers.get("connection").unwrap(), "close"); assert_eq!(headers.get("connection").unwrap(), "close");
assert_eq!(headers.get("accept-encoding").unwrap(), "gzip"); assert_eq!(headers.get("accept-encoding").unwrap(), "gzip");
assert_eq!(headers.get("icy-metadata"), None); assert_eq!(headers.get("icy-metadata"), None);
assert_eq!(headers.get("user-agent").unwrap(), "test user-agent"); assert_eq!(headers.get("user-agent").unwrap(), "test user-agent");
Response::new(Body::from("Hello World")) hyper::Response::new(full_body("Hello World"))
}, },
|src| { |src| {
src.set_property("keep-alive", false); src.set_property("keep-alive", false);
@ -457,14 +471,13 @@ fn test_basic_request_inverted_defaults() {
#[test] #[test]
fn test_extra_headers() { fn test_extra_headers() {
use std::io::{Cursor, Read}; use std::io::{Cursor, Read};
init(); init();
// Set up a harness that returns "Hello World" for any HTTP request and check if the // Set up a harness that returns "Hello World" for any HTTP request and check if the
// extra-headers property works correctly for setting additional headers // extra-headers property works correctly for setting additional headers
let mut h = Harness::new( let mut h = Harness::new(
|req| { |req| {
use hyper::{Body, Response};
let headers = req.headers(); let headers = req.headers();
assert_eq!(headers.get("foo").unwrap(), "bar"); assert_eq!(headers.get("foo").unwrap(), "bar");
assert_eq!(headers.get("baz").unwrap(), "1"); assert_eq!(headers.get("baz").unwrap(), "1");
@ -485,7 +498,7 @@ fn test_extra_headers() {
vec!["1", "2"] vec!["1", "2"]
); );
Response::new(Body::from("Hello World")) hyper::Response::new(full_body("Hello World"))
}, },
|src| { |src| {
src.set_property( src.set_property(
@ -534,18 +547,17 @@ fn test_extra_headers() {
#[test] #[test]
fn test_cookies_property() { fn test_cookies_property() {
use std::io::{Cursor, Read}; use std::io::{Cursor, Read};
init(); init();
// Set up a harness that returns "Hello World" for any HTTP request and check if the // Set up a harness that returns "Hello World" for any HTTP request and check if the
// cookies property can be used to set cookies correctly // cookies property can be used to set cookies correctly
let mut h = Harness::new( let mut h = Harness::new(
|req| { |req| {
use hyper::{Body, Response};
let headers = req.headers(); let headers = req.headers();
assert_eq!(headers.get("cookie").unwrap(), "foo=1; bar=2; baz=3"); assert_eq!(headers.get("cookie").unwrap(), "foo=1; bar=2; baz=3");
Response::new(Body::from("Hello World")) hyper::Response::new(full_body("Hello World"))
}, },
|src| { |src| {
src.set_property( src.set_property(
@ -593,6 +605,7 @@ fn test_cookies_property() {
#[test] #[test]
fn test_iradio_mode() { fn test_iradio_mode() {
use std::io::{Cursor, Read}; use std::io::{Cursor, Read};
init(); init();
// Set up a harness that returns "Hello World" for any HTTP request and check if the // Set up a harness that returns "Hello World" for any HTTP request and check if the
@ -600,18 +613,16 @@ fn test_iradio_mode() {
// and put into caps/tags // and put into caps/tags
let mut h = Harness::new( let mut h = Harness::new(
|req| { |req| {
use hyper::{Body, Response};
let headers = req.headers(); let headers = req.headers();
assert_eq!(headers.get("icy-metadata").unwrap(), "1"); assert_eq!(headers.get("icy-metadata").unwrap(), "1");
Response::builder() hyper::Response::builder()
.header("icy-metaint", "8192") .header("icy-metaint", "8192")
.header("icy-name", "Name") .header("icy-name", "Name")
.header("icy-genre", "Genre") .header("icy-genre", "Genre")
.header("icy-url", "http://www.example.com") .header("icy-url", "http://www.example.com")
.header("Content-Type", "audio/mpeg; rate=44100") .header("Content-Type", "audio/mpeg; rate=44100")
.body(Body::from("Hello World")) .body(full_body("Hello World"))
.unwrap() .unwrap()
}, },
|_src| { |_src| {
@ -677,17 +688,16 @@ fn test_iradio_mode() {
#[test] #[test]
fn test_audio_l16() { fn test_audio_l16() {
use std::io::{Cursor, Read}; use std::io::{Cursor, Read};
init(); init();
// Set up a harness that returns "Hello World" for any HTTP request and check if the // Set up a harness that returns "Hello World" for any HTTP request and check if the
// audio/L16 content type is parsed correctly and put into the caps // audio/L16 content type is parsed correctly and put into the caps
let mut h = Harness::new( let mut h = Harness::new(
|_req| { |_req| {
use hyper::{Body, Response}; hyper::Response::builder()
Response::builder()
.header("Content-Type", "audio/L16; rate=48000; channels=2") .header("Content-Type", "audio/L16; rate=48000; channels=2")
.body(Body::from("Hello World")) .body(full_body("Hello World"))
.unwrap() .unwrap()
}, },
|_src| { |_src| {
@ -741,25 +751,23 @@ fn test_audio_l16() {
#[test] #[test]
fn test_authorization() { fn test_authorization() {
use std::io::{Cursor, Read}; use std::io::{Cursor, Read};
init(); init();
// Set up a harness that returns "Hello World" for any HTTP request // Set up a harness that returns "Hello World" for any HTTP request
// but requires authentication first // but requires authentication first
let mut h = Harness::new( let mut h = Harness::new(
|req| { |req| {
use hyper::{Body, Response};
use reqwest::StatusCode;
let headers = req.headers(); let headers = req.headers();
if let Some(authorization) = headers.get("authorization") { if let Some(authorization) = headers.get("authorization") {
assert_eq!(authorization, "Basic dXNlcjpwYXNzd29yZA=="); assert_eq!(authorization, "Basic dXNlcjpwYXNzd29yZA==");
Response::new(Body::from("Hello World")) hyper::Response::new(full_body("Hello World"))
} else { } else {
Response::builder() hyper::Response::builder()
.status(StatusCode::UNAUTHORIZED.as_u16()) .status(reqwest::StatusCode::UNAUTHORIZED.as_u16())
.header("WWW-Authenticate", "Basic realm=\"realm\"") .header("WWW-Authenticate", "Basic realm=\"realm\"")
.body(Body::empty()) .body(empty_body())
.unwrap() .unwrap()
} }
}, },
@ -802,17 +810,14 @@ fn test_authorization() {
#[test] #[test]
fn test_404_error() { fn test_404_error() {
use reqwest::StatusCode;
init(); init();
// Harness that always returns 404 and we check if that is mapped to the correct error code // Harness that always returns 404 and we check if that is mapped to the correct error code
let mut h = Harness::new( let mut h = Harness::new(
|_req| { |_req| {
use hyper::{Body, Response}; hyper::Response::builder()
.status(reqwest::StatusCode::NOT_FOUND.as_u16())
Response::builder() .body(empty_body())
.status(StatusCode::NOT_FOUND.as_u16())
.body(Body::empty())
.unwrap() .unwrap()
}, },
|_src| {}, |_src| {},
@ -830,17 +835,14 @@ fn test_404_error() {
#[test] #[test]
fn test_403_error() { fn test_403_error() {
use reqwest::StatusCode;
init(); init();
// Harness that always returns 403 and we check if that is mapped to the correct error code // Harness that always returns 403 and we check if that is mapped to the correct error code
let mut h = Harness::new( let mut h = Harness::new(
|_req| { |_req| {
use hyper::{Body, Response}; hyper::Response::builder()
.status(reqwest::StatusCode::FORBIDDEN.as_u16())
Response::builder() .body(empty_body())
.status(StatusCode::FORBIDDEN.as_u16())
.body(Body::empty())
.unwrap() .unwrap()
}, },
|_src| {}, |_src| {},
@ -881,13 +883,12 @@ fn test_network_error() {
#[test] #[test]
fn test_seek_after_ready() { fn test_seek_after_ready() {
use std::io::{Cursor, Read}; use std::io::{Cursor, Read};
init(); init();
// Harness that checks if seeking in Ready state works correctly // Harness that checks if seeking in Ready state works correctly
let mut h = Harness::new( let mut h = Harness::new(
|req| { |req| {
use hyper::{Body, Response};
let headers = req.headers(); let headers = req.headers();
if let Some(range) = headers.get("Range") { if let Some(range) = headers.get("Range") {
if range == "bytes=123-" { if range == "bytes=123-" {
@ -896,11 +897,11 @@ fn test_seek_after_ready() {
*d = ((i + 123) % 256) as u8; *d = ((i + 123) % 256) as u8;
} }
Response::builder() hyper::Response::builder()
.header("content-length", 8192 - 123) .header("content-length", 8192 - 123)
.header("accept-ranges", "bytes") .header("accept-ranges", "bytes")
.header("content-range", "bytes 123-8192/8192") .header("content-range", "bytes 123-8192/8192")
.body(Body::from(data_seek)) .body(full_body(data_seek))
.unwrap() .unwrap()
} else { } else {
panic!("Received an unexpected Range header") panic!("Received an unexpected Range header")
@ -916,10 +917,10 @@ fn test_seek_after_ready() {
*d = (i % 256) as u8; *d = (i % 256) as u8;
} }
Response::builder() hyper::Response::builder()
.header("content-length", 8192) .header("content-length", 8192)
.header("accept-ranges", "bytes") .header("accept-ranges", "bytes")
.body(Body::from(data_full)) .body(full_body(data_full))
.unwrap() .unwrap()
} }
}, },
@ -961,14 +962,13 @@ fn test_seek_after_ready() {
#[test] #[test]
fn test_seek_after_buffer_received() { fn test_seek_after_buffer_received() {
use std::io::{Cursor, Read}; use std::io::{Cursor, Read};
init(); init();
// Harness that checks if seeking in Playing state after having received a buffer works // Harness that checks if seeking in Playing state after having received a buffer works
// correctly // correctly
let mut h = Harness::new( let mut h = Harness::new(
|req| { |req| {
use hyper::{Body, Response};
let headers = req.headers(); let headers = req.headers();
if let Some(range) = headers.get("Range") { if let Some(range) = headers.get("Range") {
if range == "bytes=123-" { if range == "bytes=123-" {
@ -977,11 +977,11 @@ fn test_seek_after_buffer_received() {
*d = ((i + 123) % 256) as u8; *d = ((i + 123) % 256) as u8;
} }
Response::builder() hyper::Response::builder()
.header("content-length", 8192 - 123) .header("content-length", 8192 - 123)
.header("accept-ranges", "bytes") .header("accept-ranges", "bytes")
.header("content-range", "bytes 123-8192/8192") .header("content-range", "bytes 123-8192/8192")
.body(Body::from(data_seek)) .body(full_body(data_seek))
.unwrap() .unwrap()
} else { } else {
panic!("Received an unexpected Range header") panic!("Received an unexpected Range header")
@ -992,10 +992,10 @@ fn test_seek_after_buffer_received() {
*d = (i % 256) as u8; *d = (i % 256) as u8;
} }
Response::builder() hyper::Response::builder()
.header("content-length", 8192) .header("content-length", 8192)
.header("accept-ranges", "bytes") .header("accept-ranges", "bytes")
.body(Body::from(data_full)) .body(full_body(data_full))
.unwrap() .unwrap()
} }
}, },
@ -1038,14 +1038,13 @@ fn test_seek_after_buffer_received() {
#[test] #[test]
fn test_seek_with_stop_position() { fn test_seek_with_stop_position() {
use std::io::{Cursor, Read}; use std::io::{Cursor, Read};
init(); init();
// Harness that checks if seeking in Playing state after having received a buffer works // Harness that checks if seeking in Playing state after having received a buffer works
// correctly // correctly
let mut h = Harness::new( let mut h = Harness::new(
|req| { |req| {
use hyper::{Body, Response};
let headers = req.headers(); let headers = req.headers();
if let Some(range) = headers.get("Range") { if let Some(range) = headers.get("Range") {
if range == "bytes=123-130" { if range == "bytes=123-130" {
@ -1054,11 +1053,11 @@ fn test_seek_with_stop_position() {
*d = ((i + 123) % 256) as u8; *d = ((i + 123) % 256) as u8;
} }
Response::builder() hyper::Response::builder()
.header("content-length", 8) .header("content-length", 8)
.header("accept-ranges", "bytes") .header("accept-ranges", "bytes")
.header("content-range", "bytes 123-130/8192") .header("content-range", "bytes 123-130/8192")
.body(Body::from(data_seek)) .body(full_body(data_seek))
.unwrap() .unwrap()
} else { } else {
panic!("Received an unexpected Range header") panic!("Received an unexpected Range header")
@ -1069,10 +1068,10 @@ fn test_seek_with_stop_position() {
*d = (i % 256) as u8; *d = (i % 256) as u8;
} }
Response::builder() hyper::Response::builder()
.header("content-length", 8192) .header("content-length", 8192)
.header("accept-ranges", "bytes") .header("accept-ranges", "bytes")
.body(Body::from(data_full)) .body(full_body(data_full))
.unwrap() .unwrap()
} }
}, },
@ -1131,11 +1130,9 @@ fn test_cookies() {
// client // client
let mut h = Harness::new( let mut h = Harness::new(
|_req| { |_req| {
use hyper::{Body, Response}; hyper::Response::builder()
Response::builder()
.header("Set-Cookie", "foo=bar") .header("Set-Cookie", "foo=bar")
.body(Body::from("Hello World")) .body(full_body("Hello World"))
.unwrap() .unwrap()
}, },
|_src| { |_src| {
@ -1158,8 +1155,6 @@ fn test_cookies() {
// client provides the cookie that was set in the previous request // client provides the cookie that was set in the previous request
let mut h2 = Harness::new( let mut h2 = Harness::new(
|req| { |req| {
use hyper::{Body, Response};
let headers = req.headers(); let headers = req.headers();
let cookies = headers let cookies = headers
.get("Cookie") .get("Cookie")
@ -1167,8 +1162,8 @@ fn test_cookies() {
.to_str() .to_str()
.unwrap(); .unwrap();
assert!(cookies.split(';').any(|c| c == "foo=bar")); assert!(cookies.split(';').any(|c| c == "foo=bar"));
Response::builder() hyper::Response::builder()
.body(Body::from("Hello again!")) .body(full_body("Hello again!"))
.unwrap() .unwrap()
}, },
|_src| { |_src| {
@ -1224,20 +1219,35 @@ fn test_proxy_prop_souphttpsrc_compatibility() {
fn test_proxy() { fn test_proxy() {
init(); init();
let mut h = Harness::new(
|_req| {
hyper::Response::builder()
.body(full_body("Hello Proxy World"))
.unwrap()
},
|_src| {},
);
// Simplest possible implementation of naive oneshot proxy server? // Simplest possible implementation of naive oneshot proxy server?
// Listen on socket before spawning thread (we won't error out with connection refused). // Listen on socket before spawning thread (we won't error out with connection refused).
let incoming = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); let (proxy_handle, proxy_addr) = {
let proxy_addr = incoming.local_addr().unwrap(); let (proxy_addr_sender, proxy_addr_receiver) = tokio::sync::oneshot::channel();
println!("listening on {proxy_addr}, starting proxy server");
let proxy_server = std::thread::spawn(move || {
use std::io::*;
println!("awaiting connection to proxy server");
let (mut conn, _addr) = incoming.accept().unwrap();
let proxy_handle = h.rt.spawn(async move {
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let proxy_addr = listener.local_addr().unwrap();
println!("listening on {proxy_addr}, starting proxy server");
proxy_addr_sender.send(proxy_addr).unwrap();
println!("awaiting connection to proxy server");
let (conn, _addr) = listener.accept().await.unwrap();
let (conn_reader, mut conn_writer) = tokio::io::split(conn);
println!("client connected, reading request line"); println!("client connected, reading request line");
let mut reader = BufReader::new(conn.try_clone().unwrap()); let mut reader = tokio::io::BufReader::new(conn_reader);
let mut buf = String::new(); let mut buf = String::new();
reader.read_line(&mut buf).unwrap(); reader.read_line(&mut buf).await.unwrap();
let parts: Vec<&str> = buf.split(' ').collect(); let parts: Vec<&str> = buf.split(' ').collect();
let url = reqwest::Url::parse(parts[1]).unwrap(); let url = reqwest::Url::parse(parts[1]).unwrap();
let host = format!( let host = format!(
@ -1247,40 +1257,38 @@ fn test_proxy() {
); );
println!("connecting to target server {host}"); println!("connecting to target server {host}");
let mut server_connection = std::net::TcpStream::connect(host).unwrap(); let mut server_connection = tokio::net::TcpStream::connect(host).await.unwrap();
println!("connected to target server, sending modified request line"); println!("connected to target server, sending modified request line");
server_connection server_connection
.write_all(format!("{} {} {}\r\n", parts[0], url.path(), parts[2]).as_bytes()) .write_all(format!("{} {} {}", parts[0], url.path(), parts[2]).as_bytes())
.await
.unwrap(); .unwrap();
let (mut server_reader, mut server_writer) = tokio::io::split(server_connection);
println!("sent modified request line, forwarding data in both directions"); println!("sent modified request line, forwarding data in both directions");
let send_join_handle = { let send_join_handle = tokio::task::spawn(async move {
let mut server_connection = server_connection.try_clone().unwrap(); tokio::io::copy(&mut reader, &mut server_writer)
std::thread::spawn(move || { .await
copy(&mut reader, &mut server_connection).unwrap(); .unwrap();
}) });
}; tokio::io::copy(&mut server_reader, &mut conn_writer)
copy(&mut server_connection, &mut conn).unwrap(); .await
send_join_handle.join().unwrap(); .unwrap();
send_join_handle.await.unwrap();
println!("shutting down proxy server"); println!("shutting down proxy server");
}); });
let mut h = Harness::new( (
|_req| { proxy_handle,
use hyper::{Body, Response}; futures::executor::block_on(proxy_addr_receiver).unwrap(),
)
Response::builder() };
.body(Body::from("Hello Proxy World"))
.unwrap()
},
|src| {
src.set_property("proxy", proxy_addr.to_string());
},
);
// Set the HTTP source to Playing so that everything can start. // Set the HTTP source to Playing so that everything can start.
h.run(|src| { h.run(move |src| {
src.set_property("proxy", proxy_addr.to_string());
src.set_state(gst::State::Playing).unwrap(); src.set_state(gst::State::Playing).unwrap();
}); });
@ -1292,5 +1300,90 @@ fn test_proxy() {
assert_eq!(num_bytes, "Hello Proxy World".len()); assert_eq!(num_bytes, "Hello Proxy World".len());
// Don't leave threads hanging around. // Don't leave threads hanging around.
proxy_server.join().unwrap(); proxy_handle.abort();
let _ = futures::executor::block_on(proxy_handle);
}
/// Adapter from tokio IO traits to hyper IO traits.
mod tokio_io {
use pin_project_lite::pin_project;
use std::pin::Pin;
use std::task::{Context, Poll};
pin_project! {
#[derive(Debug)]
pub struct TokioIo<T> {
#[pin]
inner: T,
}
}
impl<T> TokioIo<T> {
pub fn new(inner: T) -> Self {
Self { inner }
}
}
impl<T> hyper::rt::Read for TokioIo<T>
where
T: tokio::io::AsyncRead,
{
fn poll_read(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
mut buf: hyper::rt::ReadBufCursor<'_>,
) -> Poll<Result<(), std::io::Error>> {
let n = unsafe {
let mut tbuf = tokio::io::ReadBuf::uninit(buf.as_mut());
match tokio::io::AsyncRead::poll_read(self.project().inner, cx, &mut tbuf) {
Poll::Ready(Ok(())) => tbuf.filled().len(),
other => return other,
}
};
unsafe {
buf.advance(n);
}
Poll::Ready(Ok(()))
}
}
impl<T> hyper::rt::Write for TokioIo<T>
where
T: tokio::io::AsyncWrite,
{
fn poll_write(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &[u8],
) -> Poll<Result<usize, std::io::Error>> {
tokio::io::AsyncWrite::poll_write(self.project().inner, cx, buf)
}
fn poll_flush(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Result<(), std::io::Error>> {
tokio::io::AsyncWrite::poll_flush(self.project().inner, cx)
}
fn poll_shutdown(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Result<(), std::io::Error>> {
tokio::io::AsyncWrite::poll_shutdown(self.project().inner, cx)
}
fn is_write_vectored(&self) -> bool {
tokio::io::AsyncWrite::is_write_vectored(&self.inner)
}
fn poll_write_vectored(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
bufs: &[std::io::IoSlice<'_>],
) -> Poll<Result<usize, std::io::Error>> {
tokio::io::AsyncWrite::poll_write_vectored(self.project().inner, cx, bufs)
}
}
} }