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

rsynko_process/
run.rs
