pub async fn spawn(
        tenant: impl Into<String>,
        mut channel: SessionChannel,
        cmd: impl Into<String>,
        args: Vec<String>,
        detached: bool,
        pty: Option<PtySize>,
    ) -> Result<Self, RemoteProcessError> {
        let tenant = tenant.into();
        let cmd = cmd.into();

        // Submit our run request and get back a mailbox for responses
        let mut mailbox = channel
            .mail(Request::new(
                tenant.as_str(),
                vec![RequestData::ProcSpawn {
                    cmd,
                    args,
                    detached,
                    pty,
                }],
            ))
            .await?;

        // Wait until we get the first response, and get id from proc started
        let (id, origin_id) = match mailbox.next().await {
            Some(res) if res.payload.len() != 1 => {
                return Err(RemoteProcessError::TransportError(TransportError::IoError(
                    io::Error::new(io::ErrorKind::InvalidData, "Got wrong payload size"),
                )));
            }
            Some(res) => {
                let origin_id = res.origin_id;
                match res.payload.into_iter().next().unwrap() {
                    ResponseData::ProcSpawned { id } => (id, origin_id),
                    ResponseData::Error(x) => {
                        return Err(RemoteProcessError::TransportError(TransportError::IoError(
                            x.into(),
                        )))
                    }
                    x => {
                        return Err(RemoteProcessError::TransportError(TransportError::IoError(
                            io::Error::new(
                                io::ErrorKind::InvalidData,
                                format!("Got response type of {}", x.as_ref()),
                            ),
                        )))
                    }
                }
            }
            None => {
                return Err(RemoteProcessError::TransportError(TransportError::IoError(
                    io::Error::from(io::ErrorKind::ConnectionAborted),
                )))
            }
        };

        // Create channels for our stdin/stdout/stderr
        let (stdin_tx, stdin_rx) = mpsc::channel(CLIENT_PIPE_CAPACITY);
        let (stdout_tx, stdout_rx) = mpsc::channel(CLIENT_PIPE_CAPACITY);
        let (stderr_tx, stderr_rx) = mpsc::channel(CLIENT_PIPE_CAPACITY);

        // Used to terminate request task, either explicitly by the process or internally
        // by the response task when it terminates
        let (kill_tx, kill_rx) = mpsc::channel(1);
        let kill_tx_2 = kill_tx.clone();

        // Now we spawn a task to handle future responses that are async
        // such as ProcStdout, ProcStderr, and ProcDone
        let (abort_res_task_tx, mut abort_res_task_rx) = mpsc::channel::<()>(1);
        let res_task = tokio::spawn(async move {
            tokio::select! {
                _ = abort_res_task_rx.recv() => {
                    panic!("killed");
                }
                res = process_incoming_responses(id, mailbox, stdout_tx, stderr_tx, kill_tx_2) => {
                    res
                }
            }
        });

        // Spawn a task that takes stdin from our channel and forwards it to the remote process
        let (abort_req_task_tx, mut abort_req_task_rx) = mpsc::channel::<()>(1);
        let req_task = tokio::spawn(async move {
            tokio::select! {
                _ = abort_req_task_rx.recv() => {
                    panic!("killed");
                }
                res = process_outgoing_requests(tenant, id, channel, stdin_rx, kill_rx) => {
                    res
                }
            }
        });

        let status = Arc::new(RwLock::new(None));
        let status_2 = Arc::clone(&status);
        let wait_task = tokio::spawn(async move {
            let res = match tokio::try_join!(req_task, res_task) {
                Ok((_, res)) => res,
                Err(x) => Err(RemoteProcessError::from(x)),
            };
            status_2.write().await.replace(res);
        });

        Ok(Self {
            id,
            origin_id,
            abort_req_task_tx,
            abort_res_task_tx,
            stdin: Some(RemoteStdin(stdin_tx)),
            stdout: Some(RemoteStdout(stdout_rx)),
            stderr: Some(RemoteStderr(stderr_rx)),
            kill: kill_tx,
            wait_task,
            status,
        })
    }
/// Spawns the specified process on the remote machine using the given session
Spawns the specified process on the remote machine using the given session
Spawns the specified process on the remote machine using the given session
pub async fn spawn(
        tenant: impl Into<String>,
        mut channel: SessionChannel,
        cmd: impl Into<String>,
        args: Vec<String>,
        detached: bool,
        pty: Option<PtySize>,
    ) -> Result<Self, RemoteProcessError> {
        let tenant = tenant.into();
        let cmd = cmd.into();
        // Submit our run request and get back a mailbox for responses
        let mut mailbox = channel
            .mail(Request::new(
                tenant.as_str(),
                vec![RequestData::ProcSpawn {
                    cmd,
                    args,
                    detached,
                    pty,
                }],
            ))
            .await?;
        // Wait until we get the first response, and get id from proc started
        let (id, origin_id) = match mailbox.next().await {
            Some(res) if res.payload.len() != 1 => {
                return Err(RemoteProcessError::TransportError(TransportError::IoError(
                    io::Error::new(io::ErrorKind::InvalidData, "Got wrong payload size"),
                )));
            }
            Some(res) => {
                let origin_id = res.origin_id;
                match res.payload.into_iter().next().unwrap() {
                    ResponseData::ProcSpawned { id } => (id, origin_id),
                    ResponseData::Error(x) => {
                        return Err(RemoteProcessError::TransportError(TransportError::IoError(
                            x.into(),
                        )))
                    }
                    x => {
                        return Err(RemoteProcessError::TransportError(TransportError::IoError(
                            io::Error::new(
                                io::ErrorKind::InvalidData,
                                format!("Got response type of {}", x.as_ref()),
                            ),
                        )))
                    }
                }
            }
            None => {
                return Err(RemoteProcessError::TransportError(TransportError::IoError(
                    io::Error::from(io::ErrorKind::ConnectionAborted),
                )))
            }
        };
        // Create channels for our stdin/stdout/stderr
        let (stdin_tx, stdin_rx) = mpsc::channel(CLIENT_PIPE_CAPACITY);
        let (stdout_tx, stdout_rx) = mpsc::channel(CLIENT_PIPE_CAPACITY);
        let (stderr_tx, stderr_rx) = mpsc::channel(CLIENT_PIPE_CAPACITY);
        // Used to terminate request task, either explicitly by the process or internally
        // by the response task when it terminates
        let (kill_tx, kill_rx) = mpsc::channel(1);
        let kill_tx_2 = kill_tx.clone();
        // Now we spawn a task to handle future responses that are async
        // such as ProcStdout, ProcStderr, and ProcDone
        let (abort_res_task_tx, mut abort_res_task_rx) = mpsc::channel::<()>(1);
        let res_task = tokio::spawn(async move {
            tokio::select! {
                _ = abort_res_task_rx.recv() => {
                    panic!("killed");
                }
                res = process_incoming_responses(id, mailbox, stdout_tx, stderr_tx, kill_tx_2) => {
                    res
                }
            }
        });
        // Spawn a task that takes stdin from our channel and forwards it to the remote process
        let (abort_req_task_tx, mut abort_req_task_rx) = mpsc::channel::<()>(1);
        let req_task = tokio::spawn(async move {
            tokio::select! {
                _ = abort_req_task_rx.recv() => {
                    panic!("killed");
                }
                res = process_outgoing_requests(tenant, id, channel, stdin_rx, kill_rx) => {
                    res
                }
            }
        });
        let status = Arc::new(RwLock::new(None));
        let status_2 = Arc::clone(&status);
        let wait_task = tokio::spawn(async move {
            let res = match tokio::try_join!(req_task, res_task) {
                Ok((_, res)) => res,
                Err(x) => Err(RemoteProcessError::from(x)),
            };
            status_2.write().await.replace(res);
        });
        Ok(Self {
            id,
            origin_id,
            abort_req_task_tx,
            abort_res_task_tx,
            stdin: Some(RemoteStdin(stdin_tx)),
            stdout: Some(RemoteStdout(stdout_rx)),
            stderr: Some(RemoteStderr(stderr_rx)),
            kill: kill_tx,
            wait_task,
            status,
        })
    }
pub async fn spawn ( tenant : impl Into < String > , mut channel : SessionChannel , cmd : impl Into < String > , args : Vec < String > , detached : bool , pty : Option < PtySize > , ) -> Result < Self , RemoteProcessError > { let tenant = tenant . into ( ) ; let cmd = cmd . into ( ) ; // Submit our run request and get back a mailbox for responses let mut mailbox = channel . mail ( Request :: new ( tenant . as_str ( ) , vec ! [ RequestData ProcSpawn { cmd args detached pty } ] , ) ) . await ? ; // Wait until we get the first response, and get id from proc started let ( id , origin_id ) = match mailbox . next ( ) . await { Some ( res ) if res . payload . len ( ) != 1 => { return Err ( RemoteProcessError :: TransportError ( TransportError :: IoError ( io :: Error :: new ( io :: ErrorKind :: InvalidData , " " ) , ) ) ) ; } Some ( res ) => { let origin_id = res . origin_id ; match res . payload . into_iter ( ) . next ( ) . unwrap ( ) { ResponseData :: ProcSpawned { id } => ( id , origin_id ) , ResponseData :: Error ( x ) => { return Err ( RemoteProcessError :: TransportError ( TransportError :: IoError ( x . into ( ) , ) ) ) } x => { return Err ( RemoteProcessError :: TransportError ( TransportError :: IoError ( io :: Error :: new ( io :: ErrorKind :: InvalidData , format ! ( " " x as_ref ( ) ) , ) , ) ) ) } } } None => { return Err ( RemoteProcessError :: TransportError ( TransportError :: IoError ( io :: Error :: from ( io :: ErrorKind :: ConnectionAborted ) , ) ) ) } } ; // Create channels for our stdin/stdout/stderr let ( stdin_tx , stdin_rx ) = mpsc :: channel ( CLIENT_PIPE_CAPACITY ) ; let ( stdout_tx , stdout_rx ) = mpsc :: channel ( CLIENT_PIPE_CAPACITY ) ; let ( stderr_tx , stderr_rx ) = mpsc :: channel ( CLIENT_PIPE_CAPACITY ) ; // Used to terminate request task, either explicitly by the process or internally // by the response task when it terminates let ( kill_tx , kill_rx ) = mpsc :: channel ( 1 ) ; let kill_tx_2 = kill_tx . clone ( ) ; // Now we spawn a task to handle future responses that are async // such as ProcStdout, ProcStderr, and ProcDone let ( abort_res_task_tx , mut abort_res_task_rx ) = mpsc :: channel :: < ( ) > ( 1 ) ; let res_task = tokio :: spawn ( async move { tokio :: select ! { abort_res_task_rx recv ( ) { panic ( " " ) } res process_incoming_responses ( id mailbox stdout_tx stderr_tx kill_tx_2 ) { res } } } ) ; // Spawn a task that takes stdin from our channel and forwards it to the remote process let ( abort_req_task_tx , mut abort_req_task_rx ) = mpsc :: channel :: < ( ) > ( 1 ) ; let req_task = tokio :: spawn ( async move { tokio :: select ! { abort_req_task_rx recv ( ) { panic ( " " ) } res process_outgoing_requests ( tenant id channel stdin_rx kill_rx ) { res } } } ) ; let status = Arc :: new ( RwLock :: new ( None ) ) ; let status_2 = Arc :: clone ( & status ) ; let wait_task = tokio :: spawn ( async move { let res = match tokio :: try_join ! ( req_task res_task ) { Ok ( ( _ , res ) ) => res , Err ( x ) => Err ( RemoteProcessError :: from ( x ) ) , } ; status_2 . write ( ) . await . replace ( res ) ; } ) ; Ok ( Self { id , origin_id , abort_req_task_tx , abort_res_task_tx , stdin : Some ( RemoteStdin ( stdin_tx ) ) , stdout : Some ( RemoteStdout ( stdout_rx ) ) , stderr : Some ( RemoteStderr ( stderr_rx ) ) , kill : kill_tx , wait_task , status , } ) }
Spawns the specified process on the remote machine using the given session
Spawns the specified process on the remote machine using the given session
tenant
impl Into<String>
channel
SessionChannel
cmd
impl Into<String>
args
Vec<String>
detached
bool
pty
Option<PtySize>