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