src/ipif.rs

   1 // Copyright 2021 Ian Jackson and contributors to Hippotat
   2 // SPDX-License-Identifier: GPL-3.0-or-later
   3 // There is NO WARRANTY.
   4
   5 use crate::prelude::*;
   6
   7 pub struct Ipif {
   8   pub tx: t_io::Split<t_io::BufReader<t_proc::ChildStdout>>,
   9   pub rx: t_proc::ChildStdin,
  10   stderr_task: JoinHandle<io::Result<()>>,
  11   child: t_proc::Child,
  12 }
  13
  14 impl Ipif {
  15   #[throws(AE)]
  16   pub fn start(cmd: &str, ic_name: Option<String>) -> Self {
  17     let mut child = tokio::process::Command::new("sh")
  18       .args(&["-c", cmd])
  19       .stdin (process::Stdio::piped())
  20       .stdout(process::Stdio::piped())
  21       .stderr(process::Stdio::piped())
  22       .kill_on_drop(true)
  23       .spawn().context("spawn ipif")?;
  24
  25     let stderr = child.stderr.take().unwrap();
  26
  27     let stderr_task = task::spawn(async move {
  28       let mut stderr = t_io::BufReader::new(stderr).lines();
  29       while let Some(l) = stderr.next_line().await? {
  30         error!("{}ipif stderr: {}",
  31                OptionPrefixColon(ic_name.as_ref()),
  32                l.trim_end());
  33       }
  34       Ok::<_,io::Error>(())
  35     });
  36
  37     let tx = child.stdout.take().unwrap();
  38     let rx = child.stdin .take().unwrap();
  39     let tx = t_io::BufReader::new(tx).split(SLIP_END);
  40
  41     Ipif {
  42       tx,
  43       rx,
  44       stderr_task,
  45       child,
  46     }
  47   }
  48
  49   pub async fn quitting(mut self, ic: Option<&InstanceConfig>) {
  50     let icd = OptionPrefixColon(ic);
  51     drop(self.rx);
  52
  53     match self.child.wait().await {
  54       Err(e) => error!("{}also, failed to await ipif child: {}", icd, e),
  55       Ok(st) => {
  56         let stderr_timeout = Duration::from_millis(1000);
  57         match tokio::time::timeout(stderr_timeout, self.stderr_task).await {
  58           Err::<_,tokio::time::error::Elapsed>(_)
  59             => warn!("{}ipif stderr task continues!", icd),
  60           Ok(Err(e)) => error!("{}ipif stderr task crashed: {}", icd, e),
  61           Ok(Ok(Err(e))) => error!("{}ipif stderr read failed: {}", icd, e),
  62           Ok(Ok(Ok(()))) => { },
  63         }
  64         if ! st.success() {
  65           error!("{}ipif process failed: {}", icd, st);
  66         }
  67       }
  68     }
  69
  70     drop(self.tx);
  71   }
  72 }