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   async {
  47     loop {
  48       select! {
  49         packet = tx_stream.next_segment(),
  50         if tx_defer.is_none() &&
  51            reqs.len() < ic.max_requests_outstanding.sat() =>
  52         {
  53           let mut upbound_total = 0;
  54           let mut upbound = vec![];
  55           let mut to_process: Option<Result<Option<Vec<u8>>,_>>
  56             = Some(packet);
  57           while let Some(packet) = to_process.take() {
  58             let mut packet =
  59               packet.context("read from ipif")?
  60               .ok_or_else(|| io::Error::from(io::ErrorKind::UnexpectedEof))?;
  61             if let Ok(()) = slip::check_checkmtu_mimeify(&mut packet, &ic) {
  62               let new_upbound_total = packet.len() + upbound_total + 1;
  63               if new_upbound_total > ic.max_batch_up.sat() {
  64                 tx_defer = Some(packet);
  65                 break;
  66               }
  67               upbound_total = new_upbound_total;
  68               upbound.push(packet);
  69               // we rely on `next_segment` being cancellation-safe,
  70               // which isn't documented as true but seems reasonably safe
  71               pin!{ let next_segment = tx_stream.next_segment(); }
  72               to_process = match poll!(next_segment) {
  73                 Poll::Ready(p) => Some(p),
  74                 Poll::Pending => None,
  75               };
  76             }
  77           }
  78           assert!( to_process.is_none() );
  79           dbg!(&reqs.len(), &upbound_total, &upbound.len());
  80
  81           let req = hyper::Request::post(&ic.url)
  82             .body(
  83               Itertools::intersperse(
  84                 upbound.into_iter().map(|u| Cow::from(u)),
  85                 Cow::from(&[SLIP_END] as &'static [u8])
  86               )
  87             ).context("construct request")?;
  88
  89 //          dbg!(&req);
  90
  91 //          hclient.request
  92
  93 /*
  94           Body
  95             made out of Stream
  96             made out of futures::stream::iter
  97
  98
  99           let datalen =
 100
 101           let o = 0;
 102           let i =
 103 */
 104           reqs.push(());
 105           // xxx make new request
 106         }
 107       }
 108     }
 109   }.await
 110 }
 111
 112 #[tokio::main]
 113 async fn main() -> Result<(), AE> {
 114   let ics = config::read(LinkEnd::Client)?;
 115   if ics.is_empty() { throw!(anyhow!("no associations with server(s)")); }
 116
 117   env_logger::init();
 118
 119   let https = HttpsConnector::new();
 120   let hclient = hyper::Client::builder().build::<_, hyper::Body>(https);
 121   let hclient = Arc::new(hclient);
 122
 123   info!("starting");
 124   let () = future::select_all(
 125     ics.into_iter().map(|ic| Box::pin(async {
 126       let assocname = ic.to_string();
 127       info!("{} starting", &assocname);
 128       let hclient = hclient.clone();
 129       let join = task::spawn(async {
 130         run_client(ic, hclient).await.void_unwrap_err()
 131       });
 132       match join.await {
 133         Ok(e) => {
 134           error!("{} failed: {:?}", &assocname, e);
 135         },
 136         Err(je) => {
 137           error!("{} panicked!", &assocname);
 138           panic::resume_unwind(je.into_panic());
 139         },
 140       }
 141     }))
 142   ).await.0;
 143
 144   error!("quitting because one of your client connections crashed");
 145   process::exit(16);
 146 }