src/bin/client.rs

   1 // Copyright 2021 Ian Jackson and contributors to Hippotat
   2 // SPDX-License-Identifier: AGPL-3.0-or-later
   3 // There is NO WARRANTY.
   4
   5 use hippotat::prelude::*;
   6
   7 /*
   8 struct Client {
   9   requests_outstanding: Vec<>,
  10   tx_read_stream: something,
  11   rx_write_stream: something,
  12 }*/
  13
  14 #[allow(unused_variables)] // xxx
  15 async fn run_client<C>(ic: InstanceConfig, hclient: Arc<hyper::Client<C>>)
  16                        -> Result<Void, AE>
  17 {
  18   debug!("{}: config: {:?}", &ic, &ic);
  19
  20   let mut ipif = tokio::process::Command::new("sh")
  21     .args(&["-c", &ic.ipif])
  22     .stdin (process::Stdio::piped())
  23     .stdout(process::Stdio::piped())
  24     .stderr(process::Stdio::piped())
  25     .kill_on_drop(true)
  26     .spawn().context("spawn ipif")?;
  27
  28   let stderr = ipif.stderr.take().unwrap();
  29   let ic_name = ic.to_string();
  30   let _ = task::spawn(async move {
  31     let mut stderr = tokio::io::BufReader::new(stderr).lines();
  32     while let Some(l) = stderr.next_line().await? {
  33       error!("{}: ipif stderr: {}", &ic_name, l.trim_end());
  34     }
  35     Ok::<_,io::Error>(())
  36   });
  37
  38   let tx_stream = ipif.stdout.take().unwrap();
  39   let mut tx_stream = tokio::io::BufReader::new(tx_stream).split(SLIP_ESC);
  40   let mut tx_defer = None;
  41
  42   let stream_for_rx = ipif.stdin .take().unwrap();
  43
  44   let mut reqs = Vec::with_capacity(ic.max_requests_outstanding.sat());
  45
  46   // xxx check that ic settings are all honoured
  47
  48   async {
  49     loop {
  50       select! {
  51         packet = tx_stream.next_segment(),
  52         if tx_defer.is_none() &&
  53            reqs.len() < ic.max_requests_outstanding.sat() =>
  54         {
  55           let mut upbound_total = 0;
  56           let mut upbound = vec![];
  57           let mut to_process: Option<Result<Option<Vec<u8>>,_>>
  58             = Some(packet);
  59           while let Some(packet) = to_process.take() {
  60             let mut packet =
  61               packet.context("read from ipif")?
  62               .ok_or_else(|| io::Error::from(io::ErrorKind::UnexpectedEof))?;
  63             if let Ok(()) = slip::check_checkmtu_mimeify(&mut packet, &ic) {
  64               let new_upbound_total = packet.len() + upbound_total + 1;
  65               if new_upbound_total > ic.max_batch_up.sat() {
  66                 tx_defer = Some(packet);
  67                 break;
  68               }
  69               upbound_total = new_upbound_total;
  70               upbound.push(packet);
  71               // we rely on `next_segment` being cancellation-safe,
  72               // which isn't documented as true but seems reasonably safe
  73               pin!{ let next_segment = tx_stream.next_segment(); }
  74               to_process = match poll!(next_segment) {
  75                 Poll::Ready(p) => Some(p),
  76                 Poll::Pending => None,
  77               };
  78             }
  79           }
  80           assert!( to_process.is_none() );
  81           dbg!(&reqs.len(), &upbound_total, &upbound.len());
  82
  83           let req = hyper::Request::post(&ic.url)
  84             .body(
  85               Itertools::intersperse(
  86                 upbound.into_iter().map(|u| Cow::from(u)),
  87                 Cow::from(&[SLIP_END] as &'static [u8])
  88               )
  89             ).context("construct request")?;
  90
  91           let resp = hclient.request(req);
  92           let fut = Box::pin(tokio::timeout(
  93             ic.http_timeout,
  94             async {
  95               let r = async {
  96                 let resp = resp.await;
  97                 if ! resp.status().is_success() {
  98                   throw!("HTTP error status {}: {}", &resp.status());
  99                 }
 100                 let resp = resp.into_body();
 101                 // xxx: some size limit to avoid mallocing the universe
 102                 let resp = resp.aggregate().await
 103                   .context("HTTP error fetching response body")?;
 104                 Ok::<_,AE>(resp)
 105               };
 106               if r.is_err() {
 107                 tokio::time::sleep(&ic.http_retry).await;
 108               }
 109               r
 110             }
 111           ));
 112           reqs.push(fut);
 113         }
 114
 115         (got, goti, _) = future::select_all(&mut reqs)
 116         {
 117           reqs.swap_remove(goti);
 118           if let Some(got) = reporter.report(got) {
 119             dbg!(got.len()); // xxx
 120           }
 121         }
 122       }
 123     }
 124   }.await
 125 }
 126
 127 #[tokio::main]
 128 async fn main() -> Result<(), AE> {
 129   let ics = config::read(LinkEnd::Client)?;
 130   if ics.is_empty() { throw!(anyhow!("no associations with server(s)")); }
 131
 132   env_logger::init();
 133
 134   let https = HttpsConnector::new();
 135   let hclient = hyper::Client::builder().build::<_, hyper::Body>(https);
 136   let hclient = Arc::new(hclient);
 137
 138   info!("starting");
 139   let () = future::select_all(
 140     ics.into_iter().map(|ic| Box::pin(async {
 141       let assocname = ic.to_string();
 142       info!("{} starting", &assocname);
 143       let hclient = hclient.clone();
 144       let join = task::spawn(async {
 145         run_client(ic, hclient).await.void_unwrap_err()
 146       });
 147       match join.await {
 148         Ok(e) => {
 149           error!("{} failed: {:?}", &assocname, e);
 150         },
 151         Err(je) => {
 152           error!("{} panicked!", &assocname);
 153           panic::resume_unwind(je.into_panic());
 154         },
 155       }
 156     }))
 157   ).await.0;
 158
 159   error!("quitting because one of your client connections crashed");
 160   process::exit(16);
 161 }