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 stream_for_rx = ipif.stdin .take().unwrap();
  41   let mut reqs = Vec::with_capacity(ic.max_requests_outstanding.sat());
  42
  43   async {
  44     loop {
  45       select! {
  46         packet = tx_stream.next_segment(),
  47         if reqs.len() < ic.max_requests_outstanding.sat() => {
  48           let packet = packet.context("read from ipif")?;
  49           reqs.push(());
  50           // xxx make new request
  51         }
  52       }
  53     }
  54   }.await
  55 }
  56
  57 #[tokio::main]
  58 async fn main() -> Result<(), AE> {
  59   let ics = config::read(LinkEnd::Client)?;
  60   if ics.is_empty() { throw!(anyhow!("no associations with server(s)")); }
  61
  62   env_logger::init();
  63
  64   let https = HttpsConnector::new();
  65   let hclient = hyper::Client::builder().build::<_, hyper::Body>(https);
  66   let hclient = Arc::new(hclient);
  67
  68   info!("starting");
  69   let () = future::select_all(
  70     ics.into_iter().map(|ic| Box::pin(async {
  71       let assocname = ic.to_string();
  72       info!("{} starting", &assocname);
  73       let hclient = hclient.clone();
  74       let join = task::spawn(async {
  75         run_client(ic, hclient).await.void_unwrap_err()
  76       });
  77       match join.await {
  78         Ok(e) => {
  79           error!("{} failed: {:?}", &assocname, e);
  80         },
  81         Err(je) => {
  82           error!("{} panicked!", &assocname);
  83           panic::resume_unwind(je.into_panic());
  84         },
  85       }
  86     }))
  87   ).await.0;
  88
  89   error!("quitting because one of your client connections crashed");
  90   process::exit(16);
  91 }