distant 0.20.0

Operate on a remote computer through file and process manipulation
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128

use std::io::{self, Write};
use std::thread;

use distant_core::{
    RemoteLspStderr, RemoteLspStdin, RemoteLspStdout, RemoteStderr, RemoteStdin, RemoteStdout,
};
use log::*;
use tokio::task::{JoinError, JoinHandle};

use super::stdin;

/// Represents a link between a remote process' stdin/stdout/stderr and this process'
/// stdin/stdout/stderr
pub struct RemoteProcessLink {
    _stdin_thread: Option<thread::JoinHandle<()>>,
    stdin_task: Option<JoinHandle<io::Result<()>>>,
    stdout_task: JoinHandle<io::Result<()>>,
    stderr_task: JoinHandle<io::Result<()>>,
}

macro_rules! from_pipes {
    ($stdin:expr, $stdout:expr, $stderr:expr, $buffer:expr) => {{
        let mut stdin_thread = None;
        let mut stdin_task = None;
        if let Some(mut stdin_handle) = $stdin {
            let (thread, mut rx) = stdin::spawn_channel($buffer);
            let task = tokio::spawn(async move {
                loop {
                    if let Some(input) = rx.recv().await {
                        trace!("Forwarding stdin: {:?}", String::from_utf8_lossy(&input));
                        if let Err(x) = stdin_handle.write(&*input).await {
                            break Err(x);
                        }
                    } else {
                        break Ok(());
                    }
                }
            });
            stdin_thread = Some(thread);
            stdin_task = Some(task);
        }
        let stdout_task = tokio::spawn(async move {
            let handle = io::stdout();
            loop {
                match $stdout.read().await {
                    Ok(output) => {
                        let mut out = handle.lock();
                        out.write_all(&output)?;
                        out.flush()?;
                    }
                    Err(x) => break Err(x),
                }
            }
        });
        let stderr_task = tokio::spawn(async move {
            let handle = io::stderr();
            loop {
                match $stderr.read().await {
                    Ok(output) => {
                        let mut out = handle.lock();
                        out.write_all(&output)?;
                        out.flush()?;
                    }
                    Err(x) => break Err(x),
                }
            }
        });

        RemoteProcessLink {
            _stdin_thread: stdin_thread,
            stdin_task,
            stdout_task,
            stderr_task,
        }
    }};
}

impl RemoteProcessLink {
    /// Creates a new process link from the pipes of a remote process.
    ///
    /// `max_pipe_chunk_size` represents the maximum size (in bytes) of data that will be read from
    /// stdin at one time to forward to the remote process.
    pub fn from_remote_pipes(
        stdin: Option<RemoteStdin>,
        mut stdout: RemoteStdout,
        mut stderr: RemoteStderr,
        max_pipe_chunk_size: usize,
    ) -> Self {
        from_pipes!(stdin, stdout, stderr, max_pipe_chunk_size)
    }

    /// Creates a new process link from the pipes of a remote LSP server process.
    ///
    /// `max_pipe_chunk_size` represents the maximum size (in bytes) of data that will be read from
    /// stdin at one time to forward to the remote process.
    pub fn from_remote_lsp_pipes(
        stdin: Option<RemoteLspStdin>,
        mut stdout: RemoteLspStdout,
        mut stderr: RemoteLspStderr,
        max_pipe_chunk_size: usize,
    ) -> Self {
        from_pipes!(stdin, stdout, stderr, max_pipe_chunk_size)
    }

    /// Shuts down the link, aborting any running tasks, and swallowing join errors
    pub async fn shutdown(self) {
        self.abort();
        let _ = self.wait().await;
    }

    /// Waits for the stdin, stdout, and stderr tasks to complete
    pub async fn wait(self) -> Result<(), JoinError> {
        if let Some(stdin_task) = self.stdin_task {
            tokio::try_join!(stdin_task, self.stdout_task, self.stderr_task).map(|_| ())
        } else {
            tokio::try_join!(self.stdout_task, self.stderr_task).map(|_| ())
        }
    }

    /// Aborts the link by aborting tasks processing stdin, stdout, and stderr
    pub fn abort(&self) {
        if let Some(stdin_task) = self.stdin_task.as_ref() {
            stdin_task.abort();
        }
        self.stdout_task.abort();
        self.stderr_task.abort();
    }
}