ninep 0.5.0

A bare bones 9p implementation
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

//! Tokio based asynchronous implementation of 9p Servers and Clients
use crate::{
    Result,
    sansio::protocol::{NineP, Rdata, Rmessage, SharedBuf},
};
use simple_coro::CoroState;
use std::{future::Future, io, marker::Unpin};
use tokio::{
    io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt},
    net::{TcpStream, UnixStream},
};

pub mod client;
pub mod server;

/// Asynchronous IO support for reading and writing 9p messages
pub trait AsyncNineP: NineP + Send + Sync {
    /// Encode self as bytes for the 9p protocol and write to the given [AsyncWrite].
    fn write_to<W>(&self, w: &mut W) -> impl Future<Output = io::Result<()>> + Send
    where
        W: AsyncWrite + Unpin + Send,
    {
        write_to(self, w)
    }

    /// Decode self from 9p protocol bytes coming from the given [AsyncRead].
    fn read_from<R>(buf: &SharedBuf, r: &mut R) -> impl Future<Output = io::Result<Self>> + Send
    where
        R: AsyncRead + Unpin + Send,
    {
        read_from(buf, r)
    }
}

impl<T> AsyncNineP for T where T: NineP + Send + Sync {}

// write_to and read_from are written as free functions so we can use async/await here while also
// explicitly requiring a Send bound on the methods of the AsyncNineP trait above.

#[inline(always)]
async fn write_to<T, W>(t: &T, w: &mut W) -> io::Result<()>
where
    T: NineP + Sync,
    W: AsyncWrite + Unpin + Send,
{
    let mut buf = vec![0; t.n_bytes()];
    t.write_bytes(&mut buf)
        .map_err(|e| io::Error::other(e.to_string()))?;

    w.write_all(&buf).await
}

#[inline(always)]
async fn read_from<T, R>(buf: &SharedBuf, r: &mut R) -> io::Result<T>
where
    T: NineP + Send,
    R: AsyncRead + Unpin + Send,
{
    let mut coro = T::read_9p_coro(buf);
    loop {
        coro = match coro.resume() {
            CoroState::Pending(c, n) => {
                // SAFETY: coro is currently suspended and unable to take a reference to buf
                let mut_buf = unsafe { buf.as_inner_mut() };
                mut_buf.resize(n, 0);
                r.read_exact(mut_buf).await?;
                c.send(())
            }

            CoroState::Complete(res) => return res,
        };
    }
}

/// A Stream that makes use of the tokio [AsyncRead] and [AsyncWrite] traits to perform IO
#[allow(async_fn_in_trait)]
pub trait AsyncStream: AsyncRead + AsyncWrite + Unpin + Send + Sized + 'static {
    /// Reply to the specified tag with a given Result. Err's will be converted to 9p error
    /// messages automatically.
    async fn reply(&mut self, tag: u16, resp: Result<Rdata>) {
        let r: Rmessage = (tag, resp).into();
        let _ = r.write_to(self).await;
    }
}

impl AsyncStream for UnixStream {}
impl AsyncStream for TcpStream {}