server: collect body

[hippotat.git] / src / bin / server.rs

diff --git a/src/bin/server.rs b/src/bin/server.rs
index 3e71b4134776e76778cb4373c1712fd2c3e3e50e..b956101e681a2fba8c20dc0f5652fc80a1e15ba4 100644 (file)
--- a/src/bin/server.rs
+++ b/src/bin/server.rs
@@ -22,10 +22,28 @@ struct ClientHandles {
   web: tokio::sync::mpsc::Sender<WebRequest>,
 }
 
+/// Sent from hyper worker pool task to client task
+#[allow(dead_code)] // xxx
 struct WebRequest {
-  reply: tokio::sync::oneshot::Sender<()>,
+  // initial part of body
+  // used up to and including first 2 lines of metadata
+  // end delimiter for the metadata not yet located, but in here somewhere
+  initial: Box<[u8]>,
+  initial_remaining: usize,
+  body: hyper::body::Body,
+  reply_to: tokio::sync::oneshot::Sender<WebResponse>,
+  warnings: Warnings,
 }
 
+/// Reply from client task to hyper worker pool task
+#[allow(dead_code)] // xxx
+struct WebResponse {
+  warnings: Warnings,
+  data: Result<WebResponseData, AE>,
+}
+
+type WebResponseData = ();
+
 async fn handle(
   all_clients: Arc<AllClients>,
   req: hyper::Request<hyper::Body>
@@ -122,6 +140,8 @@ async fn handle(
     let hmac_exp = token_hmac(secret, client_time_s.as_bytes());
     // We also definitely want a consttime memeq for the hmac value
     let hmac_ok = hmac_got.ct_eq(&hmac_exp);
+    //dbg!(DumpHex(&hmac_exp), client.is_some());
+    //dbg!(DumpHex(hmac_got), hmac_ok, client_exists);
     if ! bool::from(hmac_ok & client_exists) {
       throw!(anyhow!("xxx should be a 403 error"));
     }
@@ -139,11 +159,29 @@ async fn handle(
     };
     chk_skew(client_time, now, "ahead")?;
     chk_skew(now, client_time, "behind")?;
-    
-    eprintln!("boundary={:?} start={} name={:?} client={}",
-              boundary, start, &comp.name, &client.ic);
 
-    Ok::<_,AE>(())
+    let initial_remaining = meta.remaining_bytes_len();
+
+    //eprintln!("boundary={:?} start={} name={:?} client={}",
+    // boundary, start, &comp.name, &client.ic);
+
+    let (reply_to, reply_recv) = tokio::sync::oneshot::channel();
+    let wreq = WebRequest {
+      initial,
+      initial_remaining,
+      body,
+      warnings: mem::take(&mut warnings),
+      reply_to
+    };
+    trace!("{} request", &client.ic);
+
+    client.web.try_send(wreq)
+      .map_err(|_| anyhow!("client task shut down!"))?;
+
+    let reply: WebResponse = reply_recv.await?;
+    warnings = reply.warnings;
+
+    reply.data
   }.await {
     Ok(()) => {
     },
@@ -157,11 +195,86 @@ async fn handle(
   Ok(hyper::Response::new(hyper::Body::from("Hello World")))
 }
 
-async fn run_client(_ic: Arc<InstanceConfig>, _web: mpsc::Receiver<WebRequest>)
+#[allow(unused_variables)] // xxx
+async fn run_client(_ic: Arc<InstanceConfig>,
+                    mut web: mpsc::Receiver<WebRequest>)
                     -> Result<Void, AE>
 {
-  tokio::time::sleep(Duration::from_secs(1_000_000_000)).await;
-  Err(anyhow!("xxx"))
+  struct Outstanding {
+    reply_to: tokio::sync::oneshot::Sender<WebResponse>,
+    max_requests_outstanding: u32,
+  }
+  let mut outstanding: VecDeque<Outstanding> = default();
+  let  downbound: VecDeque<(/*xxx*/)> = default();
+
+  let try_send_response = |
+    reply_to: tokio::sync::oneshot::Sender<WebResponse>,
+    response: WebResponse
+  | {
+    reply_to.send(response)
+      .unwrap_or_else(|_: WebResponse| () /* oh dear */ /* xxx trace? */);
+  };
+
+  loop {
+    if let Some(ret) = {
+      if ! downbound.is_empty() {
+        outstanding.pop_front()
+      } else if let Some((i,_)) = outstanding.iter().enumerate().find({
+        |(_,o)| outstanding.len() > o.max_requests_outstanding.sat()
+      }) {
+        Some(outstanding.remove(i).unwrap())
+      } else {
+        None
+      }
+    } {
+      let response = WebResponse {
+        data: Ok(()),
+        warnings: default(),
+      };
+
+      try_send_response(ret.reply_to, response);
+    }
+
+    select!{
+      req = web.recv() =>
+      {
+        let WebRequest {
+          initial, initial_remaining, body,
+          reply_to, warnings,
+        } = req.ok_or_else(|| anyhow!("webservers all shut down!"))?;
+
+        match async {
+
+          let whole_request = body.try_fold(
+            initial.into_vec(),
+            |mut w, by| async move { w.extend_from_slice(&by); Ok(w) },
+          ).await.context("read request body")?;
+
+          dbg!(whole_request.len());
+
+/*          
+
+          multipart::ComponentIterator::resume_mid_component(
+            &initial[initial_remaining..],
+  */          
+
+          Ok::<_,AE>(())
+        }.await {
+          Ok(()) => outstanding.push_back(Outstanding {
+            reply_to: reply_to,
+            max_requests_outstanding: 42, // xxx
+          }),
+          Err(e) => {
+            try_send_response(reply_to, WebResponse {
+              data: Err(e),
+              warnings,
+            });
+          },
+        }
+      }
+    }
+  }
+  //Err(anyhow!("xxx"))
 }
 
 #[tokio::main]