server: use meta.need

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

diff --git a/src/bin/server.rs b/src/bin/server.rs
index dfee81960ae90f71238441fe1b627a6d9eeab462..42e3785024f00a39d3892a3cb70d90706c847fd3 100644 (file)
--- a/src/bin/server.rs
+++ b/src/bin/server.rs
@@ -15,11 +15,30 @@ pub struct Opts {
 
 const METADATA_MAX_LEN: usize = MAX_OVERHEAD;
 
+type AllClients = HashMap<ClientName, ClientHandles>;
+
+struct ClientHandles {
+  ic: Arc<InstanceConfig>,
+  web: tokio::sync::mpsc::Sender<WebRequest>,
+}
+
+struct WebRequest {
+  reply: tokio::sync::oneshot::Sender<()>,
+}
+
 async fn handle(
-//    context: (),
-//    addr: SocketAddr,
-    req: hyper::Request<hyper::Body>
-) -> Result<hyper::Response<hyper::Body>, Infallible> {
+  _all_clients: Arc<AllClients>,
+  req: hyper::Request<hyper::Body>
+) -> Result<hyper::Response<hyper::Body>, Void> {
+  if req.method() == Method::GET {
+    let mut resp = hyper::Response::new(hyper::Body::from("hippotat\r\n"));
+    resp.headers_mut().insert(
+      "Content-Type",
+      "text/plain; charset=US-ASCII".try_into().unwrap()
+    );
+    return Ok(resp)
+  }
+
   let mut warnings: Warnings = default();
 
   match async {
@@ -61,17 +80,21 @@ async fn handle(
     else if let Some(start) = find_iter.next() { start + boundary.len() }
     else { throw!(anyhow!("initial boundary not found")) };
 
+    let comp = multipart::process_component
+      (&mut warnings, &initial[start..], PartName::m)?
+      .ok_or_else(|| anyhow!(r#"no "m" component"#))?;
 
-/*
-    let comp = multipart::process_component(
+    if comp.name != PartName::m { throw!(anyhow!(
+      r#"first multipart component must be name="m""#
+    )) }
 
-    if { 
+    let mut meta = MetadataFieldIterator::new(comp.payload_start);
 
-initial[start..].strip_prefixstarts_with(b"\n")
-    || initial[start..].starts_with(b"\r\n")*/
+    let client: ClientName = meta.need().context("client addr")?;
+// let client = all_clients.get(&client).ok_or_else(|| anyhow!(BAD_CLIENT))?;
 
-    eprintln!("boundary={:?} initial={:?} start={}",
-              boundary, initial, start);
+    eprintln!("boundary={:?} start={} name={:?} client={}",
+              boundary, start, &comp.name, client);
 
     Ok::<_,AE>(())
   }.await {
@@ -87,40 +110,81 @@ initial[start..].strip_prefixstarts_with(b"\n")
   Ok(hyper::Response::new(hyper::Body::from("Hello World")))
 }
 
+async fn run_client(_ic: Arc<InstanceConfig>, _web: mpsc::Receiver<WebRequest>)
+                    -> Result<Void, AE>
+{
+  Err(anyhow!("xxx"))
+}
 
 #[tokio::main]
 async fn main() {
   let opts = Opts::from_args();
-  let mut hservers = vec![];
-  let (ics,(global,ipif)) = config::startup(
+  let mut tasks: Vec<(
+    JoinHandle<AE>,
+    String,
+  )> = vec![];
+
+  let (global, ipif) = config::startup(
     "hippotatd", LinkEnd::Server,
     &opts.config, &opts.log, |ics|
   {
     let global = config::InstanceConfigGlobal::from(&ics);
     let ipif = Ipif::start(&global.ipif, None)?;
 
-    let make_service = hyper::service::make_service_fn(|_conn| async {
-      Ok::<_, Infallible>(hyper::service::service_fn(handle))
-    });
+    let all_clients: AllClients = ics.into_iter().map(|ic| {
+      let ic = Arc::new(ic);
+
+      let (web_send, web_recv) = mpsc::channel(
+        5 // xxx should me max_requests_outstanding but that's
+          // marked client-only so needs rework
+      );
+
+      let ic_ = ic.clone();
+      tasks.push((tokio::spawn(async move {
+        run_client(ic_, web_recv).await.void_unwrap_err()
+      }), format!("client {}", &ic)));
+
+      (ic.link.client,
+       ClientHandles {
+         ic,
+         web: web_send,
+       })
+    }).collect();
+    let all_clients = Arc::new(all_clients);
 
     for addr in &global.addrs {
+      let all_clients_ = all_clients.clone();
+      let make_service = hyper::service::make_service_fn(move |_conn| {
+        let all_clients_ = all_clients_.clone();
+        async { Ok::<_, Void>( hyper::service::service_fn(move |req| {
+          handle(all_clients_.clone(), req)
+        }) ) } }
+      );
+
       let addr = SocketAddr::new(*addr, global.port);
       let server = hyper::Server::try_bind(&addr)
         .context("bind")?
         .http1_preserve_header_case(true)
         .serve(make_service);
       info!("listening on {}", &addr);
-      let task = tokio::task::spawn(server);
-      hservers.push(task);
+      let task = tokio::task::spawn(async move {
+        match server.await {
+          Ok(()) => anyhow!("shut down?!"),
+          Err(e) => e.into(),
+        }
+      });
+      tasks.push((task, format!("http server {}", addr)));
     }
-
+    
     Ok((global, ipif))
   });
 
-  let x = future::select_all(&mut hservers).await;
-  error!("xxx hserver {:?}", &x);
+  let died = future::select_all(
+    tasks.iter_mut().map(|e| &mut e.0)
+  ).await;
+  error!("xxx {:?}", &died);
 
   ipif.quitting(None).await;
 
-  dbg!(ics, global);
+  dbg!(global);
 }