Simultaneous HTTP downloads

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,169 @@
+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,
+                Expectation {
+                    payload_rx,
+                    happened_tx,
+                },
+            )
+            .is_some()
+        {
+            panic!("already anticipating");
+        };
+        AnticipatedRequest {
+            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 {
+    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.");
+        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) {
+        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 {
+                log::debug!("Responding.");
+
+                // [`AnticipatedRequest`] might be dropped by now, and there is no one to receive it,
+                // but that is OK.
+                let _ = expectation.happened_tx.send(());
+
+                let payload = expectation.payload_rx.await.unwrap();
+                Ok(Response::new(Full::new(payload)))
+            } 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" }