rsynko_process/
run.rs

1use crate::ProcessHold;
2use ambassador::Delegate;
3use rsynko_manager::ChangeKind;
4use rsynko_memory::{PlannedChange, RsyncEndpoint, RsyncSyntax, SyncCommand, SyncObservation};
5use rsynko_rsync::*;
6use std::io::{BufReader, Read};
7use std::process::{Child, ChildStdout, Command, Stdio};
8use std::sync::mpsc::Sender;
9
10/// Runs folder transfers as operating-system processes.
11// Several capabilities delegate to the same component, which Clippy reads as a repeated attribute.
12#[allow(clippy::duplicated_attributes, reason = "one delegation per capability, not per target")]
13#[derive(Debug, Delegate)]
14#[delegate(RsyncEndpointAlg, target = "syntax")]
15#[delegate(RsyncEndpointViewAlg, target = "syntax")]
16#[delegate(SyncCommandAlg, target = "syntax")]
17#[delegate(SyncCommandViewAlg, target = "syntax")]
18#[delegate(SyncChangeAlg, target = "syntax")]
19#[delegate(SyncObservationAlg, target = "syntax")]
20#[delegate(SyncObservationViewAlg, target = "syntax")]
21pub struct ProcessSyncEnv {
22    syntax: RsyncSyntax,
23    observations: Sender<SyncObservation>,
24    hold: ProcessHold,
25}
26
27impl RsyncSorts for ProcessSyncEnv {
28    type Endpoint = RsyncEndpoint;
29    type Command = SyncCommand;
30    type Observation = SyncObservation;
31    type Change = PlannedChange;
32}
33
34/// Carries one running transfer process and the output it has not been read out of yet.
35#[derive(Debug)]
36pub struct SyncRun {
37    child: Child,
38    output: BufReader<ChildStdout>,
39}
40
41/// Denotes failure to run one folder transfer.
42#[derive(Debug, thiserror::Error)]
43pub enum ProcessSyncError {
44    /// Denotes a transfer program that could not be started at all.
45    #[error("the transfer program could not be started: {0}")]
46    Unstartable(std::io::Error),
47    /// Denotes failure to read what a running transfer wrote.
48    #[error("the transfer could not be read: {0}")]
49    Unreadable(std::io::Error),
50    /// Denotes a transfer that ran and refused what it was asked to do.
51    #[error("the transfer refused: {0}")]
52    Refused(String),
53}
54
55impl ProcessSyncEnv {
56    /// Runs transfers, stating every observation it reads to the given receiver.
57    #[must_use]
58    pub fn new(observations: Sender<SyncObservation>) -> Self {
59        Self::held(observations, ProcessHold::default())
60    }
61
62    /// Runs transfers that whoever holds this handle can hold still and let go again.
63    #[must_use]
64    pub const fn held(observations: Sender<SyncObservation>, hold: ProcessHold) -> Self {
65        Self { syntax: RsyncSyntax, observations, hold }
66    }
67}
68
69impl SyncRunAlg for ProcessSyncEnv {
70    type Error = ProcessSyncError;
71    type Run = SyncRun;
72
73    fn start_sync(&self, command: &Self::Command) -> Result<Self::Run, Self::Error> {
74        let mut started = Command::new(self.command_program(command));
75        started
76            .args(self.command_arguments(command).collect::<Vec<_>>())
77            .stdin(Stdio::null())
78            .stdout(Stdio::piped())
79            .stderr(Stdio::piped());
80        // The transfer is given a family of its own, so holding it still holds all of it and
81        // nothing else.
82        #[cfg(unix)]
83        std::os::unix::process::CommandExt::process_group(&mut started, 0);
84        let mut child = started.spawn().map_err(ProcessSyncError::Unstartable)?;
85        self.hold.running(Some(child.id()));
86        let stdout =
87            child.stdout.take().ok_or_else(|| ProcessSyncError::Refused("the transfer wrote nothing".to_owned()))?;
88        Ok(SyncRun { child, output: BufReader::new(stdout) })
89    }
90
91    fn next_sync_line(&self, run: &mut Self::Run) -> Result<Option<String>, Self::Error> {
92        let mut line = Vec::new();
93        let mut byte = [0_u8; 1];
94        loop {
95            let read = run.output.read(&mut byte).map_err(ProcessSyncError::Unreadable)?;
96            if read == 0 {
97                break;
98            }
99            // Progress is written over itself with carriage returns, so either ends a line.
100            if byte[0] == b'\n' || byte[0] == b'\r' {
101                if line.is_empty() {
102                    continue;
103                }
104                break;
105            }
106            line.push(byte[0]);
107        }
108        if line.is_empty() {
109            return Ok(None);
110        }
111        Ok(Some(String::from_utf8_lossy(&line).into_owned()))
112    }
113
114    fn finish_sync(&self, mut run: Self::Run) -> Result<(), Self::Error> {
115        let status = run.child.wait().map_err(ProcessSyncError::Unreadable)?;
116        self.hold.running(None);
117        if status.success() {
118            return Ok(());
119        }
120        let mut refusal = String::new();
121        if let Some(errors) = run.child.stderr.as_mut() {
122            let _read = errors.read_to_string(&mut refusal);
123        }
124        let refusal = refusal.trim();
125        Err(ProcessSyncError::Refused(if refusal.is_empty() {
126            format!("the transfer ended with {status}")
127        } else {
128            refusal.lines().take(4).collect::<Vec<_>>().join("; ")
129        }))
130    }
131}
132
133impl SyncWatchAlg for ProcessSyncEnv {
134    fn watch_sync(&self, observation: &Self::Observation) {
135        // Nobody watching is not a failure: a transfer states what it does whether or not it is
136        // being read.
137        let _watched = self.observations.send(observation.clone());
138    }
139}
rsynko_process/run.rs

rsynko_process/
run.rs
