strop_remote/exec/
stream.rs

1//! Stream remote stdout without retaining it; keep the same supervised stdin lease.
2use super::{RemoteCommand, RemoteCommandError, RemoteExitStatus, StdinMode};
3use strop_core::process::{self, StreamError, StreamPolicy};
4use strop_core::worker::CancelToken;
5use strop_workspace::RemoteEndpoint;
6
7pub struct RemoteStreamOutput {
8    pub status: RemoteExitStatus,
9    pub stderr: Vec<u8>,
10    pub stderr_dropped: u64,
11}
12#[derive(Debug)]
13pub enum RemoteStreamError<E> {
14    Remote(RemoteCommandError),
15    Consumer(E),
16}
17impl<E> From<RemoteCommandError> for RemoteStreamError<E> {
18    fn from(error: RemoteCommandError) -> Self {
19        Self::Remote(error)
20    }
21}
22
23pub fn stream<E>(
24    endpoint: &RemoteEndpoint,
25    command: &RemoteCommand,
26    token: &CancelToken,
27    consume: impl FnMut(&[u8]) -> Result<(), E>,
28) -> Result<RemoteStreamOutput, RemoteStreamError<E>> {
29    let (mut process, key) = super::run::supervised(endpoint, command, StdinMode::Finite)?;
30    let output = process::stream_with(
31        &mut process,
32        token,
33        &StreamPolicy {
34            stderr_limit: super::run::STDERR_LIMIT,
35            stderr_tail: super::run::STDERR_TAIL,
36            deadline: command.deadline(),
37            hold_stdin: true,
38        },
39        consume,
40    )
41    .map_err(|error| match error {
42        StreamError::Consumer(error) => RemoteStreamError::Consumer(error),
43        StreamError::Spawn(message) => {
44            RemoteStreamError::Remote(RemoteCommandError::Spawn { message })
45        }
46        StreamError::Cancelled => RemoteStreamError::Remote(RemoteCommandError::Cancelled {
47            diagnostics: "stream cancellation observed".into(),
48        }),
49        StreamError::TimedOut(deadline) => RemoteStreamError::Remote(RemoteCommandError::Timeout {
50            seconds: deadline.as_secs(),
51        }),
52        StreamError::Failure(error) => RemoteStreamError::Remote(RemoteCommandError::Local {
53            message: error.message,
54        }),
55    })?;
56    // The shared classifier consumes status and bounded stderr. Stdout has already
57    // been delivered to the caller and is deliberately absent from this result API.
58    let output = super::run::classify(
59        process::CommandOutput {
60            status: output.status,
61            stdout: Vec::new(),
62            stdout_dropped: 0,
63            stderr: output.stderr,
64            stderr_dropped: output.stderr_dropped,
65            stdin_error: None,
66        },
67        &key,
68    )?;
69    Ok(RemoteStreamOutput {
70        status: output.status,
71        stderr: output.stderr,
72        stderr_dropped: output.stderr_dropped,
73    })
74}
strop_remote/exec/stream.rs

strop_remote/exec/
stream.rs
