server: wip

Signed-off-by: Ian Jackson <ijackson@chiark.greenend.org.uk>

diff --git a/src/bin/server.rs b/src/bin/server.rs
index 26abf5eb7c72c07e94accd0ce38a9fc92e468adf..764b22f5e2954a98e9d6eede7c034fbb8c54dc56 100644 (file)
--- a/src/bin/server.rs
+++ b/src/bin/server.rs
@@ -22,10 +22,22 @@ 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: tokio::sync::oneshot::Sender<WebResponse>,
 }
 
+/// Reply from client task to hyper worker pool task
+type WebResponse = Result<WebResponseData, AE>;
+type WebResponseData = ();
+
 async fn handle(
   all_clients: Arc<AllClients>,
   req: hyper::Request<hyper::Body>
@@ -141,11 +153,27 @@ 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, reply_recv) = tokio::sync::oneshot::channel();
+    let wreq = WebRequest {
+      initial,
+      initial_remaining,
+      body,
+      reply
+    };
+    trace!("{} request", &client.ic);
+
+    client.web.try_send(wreq)
+      .map_err(|_| anyhow!("client task shut down!"))?;
+
+    let reply: WebResponse = reply_recv.await?;
+
+    reply
   }.await {
     Ok(()) => {
     },
@@ -159,11 +187,45 @@ async fn handle(
   Ok(hyper::Response::new(hyper::Body::from("Hello World")))
 }
 
-async fn run_client(_ic: Arc<InstanceConfig>, _web: mpsc::Receiver<WebRequest>)
+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: tokio::sync::oneshot::Sender<WebResponse>,
+    max_requests_outstanding: u32,
+  }
+  let mut outstanding: VecDeque<Outstanding> = default();
+  let  downbound: VecDeque<(/*xxx*/)> = default();
+
+  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
+      }
+    } {
+      ret.reply.send(Ok(() /* dummy */))
+        .unwrap_or_else(|_: WebResponse| () /* oh dear */ /* xxx trace? */);
+    }
+
+    select!{
+      req = web.recv() =>
+      {
+        let req = req.ok_or_else(|| anyhow!("webservers all shut down!"))?;
+        outstanding.push_back(Outstanding {
+          reply: req.reply,
+          max_requests_outstanding: 42, // xxx
+        });
+      }
+    }
+  }
+  //Err(anyhow!("xxx"))
 }
 
 #[tokio::main]

diff --git a/src/multipart.rs b/src/multipart.rs
index 1c0be403351b4d91fe691afc6d7d39e10b79d8fe..9f3fca9fade6b2533c35161f6e704e7fdcd2f35f 100644 (file)
--- a/src/multipart.rs
+++ b/src/multipart.rs
@@ -109,6 +109,14 @@ impl<'b> MetadataFieldIterator<'b> {
     let s = if let Some(r) = self.next() { r? } else { return None };
     Some(s.parse()?)
   }
+
+  pub fn remaining_bytes_len(&self) -> usize {
+    if let Some(last) = self.last {
+      self.buf.len() - last
+    } else {
+      0
+    }
+  }
 }
                                       
 impl<'b> Iterator for MetadataFieldIterator<'b> {