Simultaneous HTTP downloads

CHANGELOG.md

@@ -2,6 +2,10 @@
 
 All notable changes to this project will be documented in this file.
 
+## Unreleased
+
+ * Tiles are now downloaded in parallel.
+
 ## 0.17.0
 
  * `egui` updated to 0.25.

Cargo.toml

@@ -1,5 +1,12 @@
 [workspace]
-members = ["walkers", "demo", "demo_native", "demo_web", "demo_android/rust"]
+members = [
+ "walkers",
+ "demo",
+ "demo_native",
+ "demo_web",
+ "demo_android/rust",
+ "hypermocker",
+]
 resolver = "2"
 
 [workspace.package]

hypermocker/Cargo.toml

@@ -0,0 +1,17 @@
+[package]
+name = "hypermocker"
+edition = "2021"
+version = "0.1.0"
+publish = false
+
+[dependencies]
+hyper = { version = "1.1.0", features = ["full"] }
+tokio = { version = "1.28", features = ["macros"] }
+hyper-util = { version = "0.1", features = ["full"] }
+http-body-util = "0.1"
+log = "0.4"
+
+[dev-dependencies]
+reqwest = "0.11"
+futures = "0.3.28"
+env_logger = "0.10"

hypermocker/src/lib.rs

@@ -0,0 +1,189 @@
+use http_body_util::Full;
+use hyper::{server::conn::http1, Response};
+use hyper_util::rt::TokioIo;
+use std::{
+ collections::HashMap,
+ future::Future,
+ net::SocketAddr,
+ pin::Pin,
+ sync::{Arc, Mutex},
+};
+use tokio::net::TcpListener;
+use tokio::sync::oneshot;
+
+pub use hyper::body::Bytes;
+
+struct Expectation {
+ payload_rx: oneshot::Receiver<Bytes>,
+ happened_tx: oneshot::Sender<()>,
+}
+
+#[derive(Default)]
+struct State {
+ /// Anticipations made by [`Mock::anticipate`].
+ expectations: HashMap<String, Expectation>,
+
+ /// Requests that were unexpected.
+ unexpected: Vec<String>,
+}
+
+pub struct Server {
+ port: u16,
+ state: Arc<Mutex<State>>,
+}
+
+impl Server {
+ /// Create new [`Mock`], and bind it to a random port.
+ pub async fn bind() -> Server {
+ let state = Arc::new(Mutex::new(State::default()));
+
+ let addr = SocketAddr::from(([127, 0, 0, 1], 0));
+ let listener = TcpListener::bind(addr).await.unwrap();
+ let port = listener.local_addr().unwrap().port();
+
+ let state_clone = state.clone();
+ tokio::spawn(async move {
+ loop {
+ let (stream, _) = listener.accept().await.unwrap();
+ let io = TokioIo::new(stream);
+
+ let state = state_clone.clone();
+ tokio::task::spawn(async move {
+ http1::Builder::new()
+ .serve_connection(io, Service { state })
+ .await
+ .unwrap();
+ });
+ }
+ });
+
+ Server { port, state }
+ }
+
+ /// Port, which this server listens on.
+ pub fn port(&self) -> u16 {
+ self.port
+ }
+
+ /// Anticipate a HTTP request, but do not respond to it yet.
+ pub async fn anticipate(&self, url: String) -> AnticipatedRequest {
+ log::info!("Anticipating '{}'.", url);
+ let (payload_tx, payload_rx) = oneshot::channel();
+ let (happened_tx, happened_rx) = oneshot::channel();
+ if self
+ .state
+ .lock()
+ .unwrap()
+ .expectations
+ .insert(
+ url.to_owned(),
+ Expectation {
+ payload_rx,
+ happened_tx,
+ },
+ )
+ .is_some()
+ {
+ panic!("already anticipating");
+ };
+ AnticipatedRequest {
+ url,
+ payload_tx,
+ happened_rx: Some(happened_rx),
+ }
+ }
+}
+
+impl Drop for Server {
+ fn drop(&mut self) {
+ if !self.state.lock().unwrap().unexpected.is_empty() {
+ panic!("there are unexpected requests");
+ }
+ }
+}
+
+/// HTTP request that was anticipated to arrive.
+pub struct AnticipatedRequest {
+ url: String,
+ payload_tx: tokio::sync::oneshot::Sender<Bytes>,
+ happened_rx: Option<oneshot::Receiver<()>>,
+}
+
+impl AnticipatedRequest {
+ /// Respond to this request with the given body.
+ pub async fn respond(self, payload: Bytes) {
+ log::info!("Responding to '{}'.", self.url);
+ self.payload_tx.send(payload).unwrap();
+ }
+
+ /// Expect the request to come, but still do not respond to it yet.
+ pub async fn expect(&mut self) {
+ log::info!("Expecting '{}'.", self.url);
+ if let Some(happened_tx) = self.happened_rx.take() {
+ happened_tx.await.unwrap();
+ } else {
+ panic!("this request was already expected");
+ }
+ }
+}
+
+struct Service {
+ state: Arc<Mutex<State>>,
+}
+
+impl hyper::service::Service<hyper::Request<hyper::body::Incoming>> for Service {
+ type Response = Response<Full<Bytes>>;
+ type Error = hyper::Error;
+ type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;
+
+ fn call(&self, request: hyper::Request<hyper::body::Incoming>) -> Self::Future {
+ log::info!("Incoming request '{}'.", request.uri());
+ let state = self.state.clone();
+ Box::pin(async move {
+ let expectation = state
+ .lock()
+ .unwrap()
+ .expectations
+ .remove(&request.uri().path().to_string());
+
+ if let Some(expectation) = expectation {
+ // [`AnticipatedRequest`] might be dropped by now, and there is no one to receive it,
+ // but that is OK.
+ let _ = expectation.happened_tx.send(());
+
+ match expectation.payload_rx.await {
+ Ok(payload) => {
+ log::debug!(
+ "Proper responding to '{}' with {} bytes.",
+ request.uri(),
+ payload.len()
+ );
+ Ok(Response::new(Full::new(payload)))
+ }
+ Err(_) => {
+ log::error!(
+ "AnticipatedRequest for '{}' was dropped before responding.",
+ request.uri()
+ );
+ // TODO: This panic will be ignored by hyper/tokio stack.
+ panic!(
+ "AnticipatedRequest for '{}' was dropped before responding.",
+ request.uri()
+ );
+ }
+ }
+ } else {
+ log::warn!("Unexpected '{}'.", request.uri());
+ state
+ .lock()
+ .unwrap()
+ .unexpected
+ .push(request.uri().to_string());
+ Ok(Response::builder()
+ .status(418)
+ .body(Full::new(Bytes::from_static(b"unexpected")))
+ .unwrap())
+ }
+ })
+ }
+}

hypermocker/tests/integration_test.rs

@@ -0,0 +1,76 @@
+use hyper::body::Bytes;
+use hypermocker::Server;
+use std::time::Duration;
+
+#[tokio::test]
+async fn anticipate_then_request() {
+ let _ = env_logger::try_init();
+
+ let mock = Server::bind().await;
+ let url = format!("http://localhost:{}/foo", mock.port());
+ let request = mock.anticipate("/foo".to_string()).await;
+
+ // Make sure that mock's internals kick in.
+ tokio::time::sleep(Duration::from_secs(1)).await;
+
+ futures::future::join(
+ async {
+ let response = reqwest::get(url).await.unwrap();
+ let bytes = response.bytes().await.unwrap();
+ assert_eq!(&bytes[..], b"hello");
+ },
+ async {
+ request.respond(Bytes::from_static(b"hello")).await;
+ },
+ )
+ .await;
+}
+
+#[tokio::test]
+async fn anticipate_expect_then_request() {
+ let _ = env_logger::try_init();
+
+ let mock = Server::bind().await;
+ let url = format!("http://localhost:{}/foo", mock.port());
+ let mut request = mock.anticipate("/foo".to_string()).await;
+
+ // Make sure that mock's internals kick in.
+ tokio::time::sleep(Duration::from_secs(1)).await;
+
+ futures::future::join(
+ async {
+ let response = reqwest::get(url).await.unwrap();
+ let bytes = response.bytes().await.unwrap();
+ assert_eq!(&bytes[..], b"hello");
+ },
+ async {
+ request.expect().await;
+ request.respond(Bytes::from_static(b"hello")).await;
+ },
+ )
+ .await;
+}
+
+#[tokio::test]
+#[should_panic(expected = "there are unexpected requests")]
+async fn unanticipated_request() {
+ let _ = env_logger::try_init();
+
+ let mock = Server::bind().await;
+ let url = format!("http://localhost:{}/foo", mock.port());
+
+ let response = reqwest::get(url).await.unwrap();
+ let bytes = response.bytes().await.unwrap();
+ assert_eq!(&bytes[..], b"unexpected");
+}
+
+#[tokio::test]
+#[should_panic(expected = "already anticipating")]
+async fn can_not_anticipate_twice() {
+ let _ = env_logger::try_init();
+
+ let mock = Server::bind().await;
+
+ mock.anticipate("/foo".to_string()).await;
+ mock.anticipate("/foo".to_string()).await;
+}

walkers/Cargo.toml

@@ -34,3 +34,4 @@ eframe.workspace = true
 env_logger = "0.10"
 approx = "0.5"
 mockito = "1.1"
+hypermocker = { path = "../hypermocker" }