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>
};
let mut body = req.into_body();
- let initial = match read_limited_bytes(METADATA_MAX_LEN, &mut body).await {
+ let initial = match read_limited_bytes(
+ METADATA_MAX_LEN, default(), default(), &mut body
+ ).await {
Ok(all) => all,
Err(ReadLimitedError::Truncated { sofar,.. }) => sofar,
Err(ReadLimitedError::Hyper(e)) => throw!(e),
let client: ClientName = meta.need_parse().context("client addr")?;
- let (client_time, hmac_got) = (||{
+ let mut hmac_got = [0; HMAC_L];
+ let (client_time, hmac_got_l) = (||{
let token: &str = meta.need_next().context(r#"find in "m""#)?;
- let (time_t, hmac_got) = token.split_once(' ')
+ let (time_t, hmac_b64) = token.split_once(' ')
.ok_or_else(|| anyhow!("split"))?;
- let time_t: u64 = time_t.parse().context("parse time_t")?;
- Ok::<_,AE>((time_t, hmac_got))
+ let time_t = u64::from_str_radix(time_t, 16).context("parse time_t")?;
+ let l = io::copy(
+ &mut base64::read::DecoderReader::new(&mut hmac_b64.as_bytes(),
+ BASE64_CONFIG),
+ &mut &mut hmac_got[..]
+ ).context("parse b64 token")?;
+ let l = l.try_into()?;
+ Ok::<_,AE>((time_t, l))
})().context("token")?;
+ let hmac_got = &hmac_got[0..hmac_got_l];
let client = all_clients.get(&client);
let client_time_s = format!("{:x}", client_time);
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.as_bytes().ct_eq(&hmac_exp);
+ 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"));
}
};
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(()) => {
},
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 {
+
+ // xxx size limit
+
+ 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]
~ian / hippotat.git / blobdiff