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<Vec<u8>> = 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           //: impl futures::Stream<Cow<&[u8]>>
  87           let body = hyper::body::Body::wrap_stream(
  88             futures::stream::iter(
  89               Itertools::intersperse(
  90                 upbound.into_iter().map(|u| Cow::from(u)),
  91                 Cow::from(&[SLIP_END] as &'static [u8])
  92               ).map(Ok)
  93             )
  94           );
  95
  96           let req = hyper::Request::post(&ic.url).body(body)
  97             .context("construct request")?;
  98
  99           let resp = hclient.request(req);
 100           let fut = Box::pin(async {
 101             let r = async { tokio::time::timeout( ic.http_timeout, async {
 102               let resp = resp.await.context("make request")?;
 103               if ! resp.status().is_success() {
 104                 throw!(anyhow!("HTTP error status {}", &resp.status()));
 105               }
 106               let resp = resp.into_body();
 107               // xxx: some size limit to avoid mallocing the universe
 108               let resp = hyper::body::aggregate(resp).await
 109                 .context("HTTP error fetching response body")?;
 110               Ok::<_,AE>(resp)
 111             }).await? }.await;
 112             if r.is_err() {
 113               tokio::time::sleep(ic.http_retry).await;
 114             }
 115             r
 116           });
 117           reqs.push(fut);
 118         }
 119
 120         (got, goti, _) = future::select_all(&mut reqs) =>
 121         {
 122           reqs.swap_remove(goti);
 123           if let Some(got) = reporter.report(got) {
 124             dbg!(&got.remaining()); // xxx
 125           }
 126         }
 127       }
 128     }
 129   }.await
 130 }
 131
 132 #[tokio::main]
 133 async fn main() -> Result<(), AE> {
 134   let ics = config::read(LinkEnd::Client)?;
 135   if ics.is_empty() { throw!(anyhow!("no associations with server(s)")); }
 136
 137   env_logger::init();
 138
 139   let https = HttpsConnector::new();
 140   let hclient = hyper::Client::builder().build::<_, hyper::Body>(https);
 141   let hclient = Arc::new(hclient);
 142
 143   info!("starting");
 144   let () = future::select_all(
 145     ics.into_iter().map(|ic| Box::pin(async {
 146       let assocname = ic.to_string();
 147       info!("{} starting", &assocname);
 148       let hclient = hclient.clone();
 149       let join = task::spawn(async {
 150         run_client(ic, hclient).await.void_unwrap_err()
 151       });
 152       match join.await {
 153         Ok(e) => {
 154           error!("{} failed: {:?}", &assocname, e);
 155         },
 156         Err(je) => {
 157           error!("{} panicked!", &assocname);
 158           panic::resume_unwind(je.into_panic());
 159         },
 160       }
 161     }))
 162   ).await.0;
 163
 164   error!("quitting because one of your client connections crashed");
 165   process::exit(16);
 166 }