server/suser.rs

   1 // Copyright 2021 Ian Jackson and contributors to Hippotat
   2 // SPDX-License-Identifier: GPL-3.0-or-later
   3 // There is NO WARRANTY.
   4
   5 use super::*;
   6
   7 #[derive(Debug)]
   8 pub struct User {
   9   pub ic: Arc<InstanceConfig>,
  10   pub web: mpsc::Sender<WebRequest>,
  11   pub route: mpsc::Sender<RoutedPacket>,
  12 }
  13
  14 #[allow(unused_variables)] // xxx
  15 #[allow(unused_mut)] // xxx
  16 pub async fn run(global: Arc<Global>,
  17                  ic: Arc<InstanceConfig>,
  18                  mut web: mpsc::Receiver<WebRequest>,
  19                  mut routed: mpsc::Receiver<RoutedPacket>)
  20                  -> Result<Void, AE>
  21 {
  22   struct Outstanding {
  23     reply_to: oneshot::Sender<WebResponse>,
  24     oi: OutstandingInner,
  25   }
  26   #[derive(Debug)]
  27   struct OutstandingInner {
  28     target_requests_outstanding: u32,
  29   }
  30   let mut outstanding: VecDeque<Outstanding> = default();
  31   let  downbound: VecDeque<(/*xxx*/)> = default();
  32
  33   let try_send_response = |
  34     reply_to: oneshot::Sender<WebResponse>,
  35     response: WebResponse
  36   | {
  37     reply_to.send(response)
  38       .unwrap_or_else(|_: WebResponse| () /* oh dear */ /* xxx trace? */);
  39   };
  40
  41   loop {
  42     if let Some(ret) = {
  43       if ! downbound.is_empty() {
  44         outstanding.pop_front()
  45       } else if let Some((i,_)) = outstanding.iter().enumerate().find({
  46         |(_,o)| outstanding.len() > o.oi.target_requests_outstanding.sat()
  47       }) {
  48         Some(outstanding.remove(i).unwrap())
  49       } else {
  50         None
  51       }
  52     } {
  53       let response = WebResponse {
  54         data: Ok(vec![ /* xxx */ ]),
  55         warnings: default(),
  56       };
  57
  58       try_send_response(ret.reply_to, response);
  59     }
  60
  61     select!{
  62       biased;
  63
  64       // xxx something something routed something
  65
  66       req = web.recv() =>
  67       {
  68         let WebRequest {
  69           initial, initial_remaining, length_hint, mut body,
  70           boundary_finder,
  71           reply_to, conn, mut warnings, may_route,
  72         } = req.ok_or_else(|| anyhow!("webservers all shut down!"))?;
  73
  74         match async {
  75
  76           let initial_used = initial.len() - initial_remaining;
  77
  78           let whole_request = read_limited_bytes(
  79             ic.max_batch_up.sat(),
  80             initial,
  81             length_hint,
  82             &mut body
  83           ).await.context("read request body")?;
  84
  85           let (meta, mut comps) =
  86             multipart::ComponentIterator::resume_mid_component(
  87               &whole_request[initial_used..],
  88               boundary_finder
  89             ).context("resume parsing body, after auth checks")?;
  90
  91           let mut meta = MetadataFieldIterator::new(&meta);
  92
  93           macro_rules! meta {
  94             { $v:ident, ( $( $badcmp:tt )? ), $ret:expr,
  95               let $server:ident, $client:ident $($code:tt)*
  96             } => {
  97               let $v = (||{
  98                 let $server = ic.$v;
  99                 let $client $($code)*
 100                 $(
 101                   if $client $badcmp $server {
 102                     throw!(anyhow!("mismatch: client={:?} {} server={:?}",
 103                                    $client, stringify!($badcmp), $server));
 104                   }
 105                 )?
 106                 Ok::<_,AE>($ret)
 107               })().context(stringify!($v))?;
 108               //dbg!(&$v);
 109             }
 110           }
 111           meta!{
 112             target_requests_outstanding, ( != ), client,
 113             let server, client: u32 = meta.need_parse()?;
 114           }
 115           meta!{
 116             http_timeout, ( > ), client,
 117             let server, client = Duration::from_secs(meta.need_parse()?);
 118           }
 119           meta!{
 120             mtu, ( != ), client,
 121             let server, client: u32 = meta.parse()?.unwrap_or(server);
 122           }
 123           meta!{
 124             max_batch_down, (), min(client, server),
 125             let server, client: u32 = meta.parse()?.unwrap_or(server);
 126           }
 127           meta!{
 128             max_batch_up, ( > ), client,
 129             let server, client = meta.parse()?.unwrap_or(server);
 130           }
 131
 132           while let Some(comp) = comps.next(&mut warnings, PartName::d)? {
 133             if comp.name != PartName::d {
 134               warnings.add(&format_args!("unexpected part {:?}", comp.name))?;
 135             }
 136             slip::processn(Mime2Slip, mtu, comp.payload, |header| {
 137               let saddr = ip_packet_addr::<false>(header)?;
 138               if saddr != ic.link.client.0 { throw!(PE::Src(saddr)) }
 139               let daddr = ip_packet_addr::<true>(header)?;
 140               Ok(daddr)
 141             }, |(daddr,packet)| route_packet(
 142               &global, &conn, Some(&ic.link.client), daddr,
 143               packet, may_route.clone(),
 144             ).map(Ok),
 145               |e| Ok::<_,SlipFramesError<_>>({ warnings.add(&e)?; })
 146             ).await?;
 147           }
 148
 149           let oi = OutstandingInner {
 150             target_requests_outstanding,
 151           };
 152           Ok::<_,AE>(oi)
 153         }.await {
 154           Ok(oi) => outstanding.push_back(Outstanding { reply_to, oi }),
 155           Err(e) => {
 156             try_send_response(reply_to, WebResponse {
 157               data: Err(e),
 158               warnings,
 159             });
 160           },
 161         }
 162       }
 163     }
 164   }
 165   //Err(anyhow!("xxx"))
 166 }