pipeterminal

Cargo.toml

[package]
name = "pipeterminal"
version = "0.1.0"
authors = ["yskszk63 <yskszk63@gmail.com>"]
edition = "2018"
license = "MIT or Apache-2.0"

# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html

[dependencies]
futures = "0.3.15"
libc = "0.2.94"
log = "0.4.14"
nix = "0.20.0"

[dependencies.tokio]
version = "1.5.0"
features = [
    "io-util",
    "io-std",
    "net",
    "rt-multi-thread",
    "macros",
    "process",
]

main.rs

// src/main.rs
// pty with tokio example.
//
// Error handling is omitted for simplicity.

mod pipestream;

use nix::{pty, fcntl};
use nix::fcntl::FcntlArg;
use std::os::unix::io::FromRawFd;
use log::warn;
use tokio::io::copy;

use crate::pipestream::PipeStream;
use nix::pty::OpenptyResult;
use tokio::process::Child;
use nix::sys::termios;
use nix::fcntl::OFlag;

#[tokio::main]
async fn main() {

    let pseudo = pty::openpty(None, None).unwrap();

    // needs master side pty
    // turn on Non-blocking of fd flags on master side pty
    let flags = OFlag::from_bits_truncate(fcntl::fcntl(libc::STDIN_FILENO, FcntlArg::F_GETFL).unwrap());
    let _ = fcntl::fcntl(pseudo.master, FcntlArg::F_SETFL(flags | OFlag::O_NONBLOCK | OFlag::O_CLOEXEC))
        .map_err(|e| warn!("fnctl pseudo.master {:?}", e)).unwrap();
    //let _ = fcntl::fcntl(pseudo.slave, FcntlArg::F_SETFD(FdFlag::FD_CLOEXEC))
    //    .map_err(|e| warn!("fcntl pseudo.slave {:?}", e)).unwrap();



    // Use the TOKIO version of Command.
    // It's not required, but it's not good to mix asynchronous and sync code
    //let mut child = std::process::Command::new("bash")
    let mut child = tokio::process::Command::new("bash")
        .spawn_pty_async(&pseudo).unwrap();

    let stream = PipeStream::from_fd(pseudo.master);
    let (mut r,mut w) = tokio::io::split(stream);

    // Note: This crate is deprecated in tokio 0.2.x and has been moved into tokio_util::codec of the tokio-util crate behind the codec feature flag.
    //https://docs.rs/tokio-codec/0.1.2/tokio_codec/
    //tokio_codec::FramedRead::new(r,tokio_codec::BytesCodec::new());

    // A terminal connected to this program raw mode, off the echo back, etc.
    let original = termios::tcgetattr(libc::STDIN_FILENO).unwrap(); // FIXME
    let mut rawmode = original.clone();
    termios::cfmakeraw(&mut rawmode);
    termios::tcsetattr(libc::STDIN_FILENO, termios::SetArg::TCSANOW, &rawmode).unwrap();

    tokio::spawn(async move{
        // The stdin / stdout of the std crate is line buffered, etc. Therefore, fd of stdin / stdout is used directly.
        let flags = OFlag::from_bits_truncate(fcntl::fcntl(libc::STDOUT_FILENO, FcntlArg::F_GETFL).unwrap());
        fcntl::fcntl(libc::STDOUT_FILENO, FcntlArg::F_SETFL(flags | OFlag::O_NONBLOCK)).unwrap();
        let mut stdout = PipeStream::from_fd(libc::STDOUT_FILENO);
        copy(&mut r, &mut stdout).await.unwrap(); // maybe I/O error occurred when reach slaves EOF
        //copy(&mut r,&mut tokio::io::stdout()).await;
    });

    tokio::spawn(async move{
        let flags = OFlag::from_bits_truncate(fcntl::fcntl(libc::STDIN_FILENO, FcntlArg::F_GETFL).unwrap());
        fcntl::fcntl(libc::STDIN_FILENO, FcntlArg::F_SETFL(flags | OFlag::O_NONBLOCK)).unwrap();
        let mut stdin = PipeStream::from_fd(libc::STDIN_FILENO);
        copy(&mut stdin,&mut w).await.unwrap();
        //copy(&mut tokio::io::stdin(),&mut w).await;
    });

    child.wait().await.unwrap();
    // Restore the state of the terminal
    termios::tcsetattr(libc::STDIN_FILENO, termios::SetArg::TCSANOW, &original).unwrap();
}



pub trait CommandExt {
    fn spawn_pty_async(&mut self, pty: &OpenptyResult)-> std::io::Result<Child>  ;
}

impl CommandExt for tokio::process::Command {
    fn spawn_pty_async(&mut self, pty: &OpenptyResult) -> std::io::Result<Child> {

        let child_stdin = unsafe { std::process::Stdio::from_raw_fd(pty.slave) };
        let child_stdout = unsafe { std::process::Stdio::from_raw_fd(pty.slave) };
        let child_stderr = unsafe { std::process::Stdio::from_raw_fd(pty.slave) };

        let masterpty = pty.master;
        self.stdout(child_stdout);
        self.stdin(child_stdin);
        self.stderr(child_stderr);

        unsafe {
        self.pre_exec(move || {
            if libc::close(masterpty) != 0 {
                return Err(std::io::Error::last_os_error());
            }

            if libc::setsid() < 0 {
                return Err(std::io::Error::last_os_error());
            }

            if libc::ioctl(0, libc::TIOCSCTTY.into(), 1) != 0 {
                return Err(std::io::Error::last_os_error());
            }
            Ok(())
        });
        }

        self.spawn()
    }
}

pipestream.rs

// src/pipestream.rs
use std::os::unix::io::RawFd;
use std::task::{Context, Poll};
use std::pin::Pin;

use tokio::io::{self, AsyncRead, AsyncWrite, ReadBuf};
use tokio::io::unix::AsyncFd;
use futures::ready;
use nix::{unistd, Error as NixError};

fn into_io_err(err: NixError) -> io::Error {
    match err {
        NixError::Sys(errno) => io::Error::from_raw_os_error(errno as i32),
        err => io::Error::new(io::ErrorKind::Other, err),
    }
}

fn sys_read(fd: RawFd, buf: &mut ReadBuf<'_>) -> io::Result<usize> {
    let buf = buf.initialize_unfilled();
    unistd::read(fd, buf).map_err(into_io_err)
}

fn sys_write(fd: RawFd, buf: &[u8]) -> io::Result<usize> {
    unistd::write(fd, buf).map_err(into_io_err)
}

pub struct PipeStream {
    inner: AsyncFd<RawFd>,
}

impl PipeStream {
    pub fn from_fd(fd: RawFd) -> Self {
        Self {
            inner: AsyncFd::new(fd).unwrap(), // FIXME
        }
    }
}

impl AsyncRead for PipeStream {
    fn poll_read(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<io::Result<()>> {
        let Self { inner, } = self.get_mut();
        loop {
            let mut guard = ready!(inner.poll_read_ready_mut(cx))?;
            let result = guard.try_io(|inner| { sys_read(*inner.get_ref(), buf) });
            if let Ok(result) = result {
                buf.advance(result?);
                return Poll::Ready(Ok(()))
            }
        }
    }
}

impl AsyncWrite for PipeStream {
    fn poll_write(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8]) -> Poll<io::Result<usize>> {
        let Self { inner, } = self.get_mut();
        loop {
            let mut guard = ready!(inner.poll_write_ready_mut(cx))?;
            let result = guard.try_io(|inner| { sys_write(*inner.get_ref(), buf) });
            if let Ok(result) = result {
                return Poll::Ready(result)
            }
        }
    }
    fn poll_flush(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<io::Result<()>> {
        Poll::Ready(Ok(()))
    }
    fn poll_shutdown(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<io::Result<()>> {
        Poll::Ready(Ok(()))
    }
}