zbus 5.3.0

API for D-Bus communication
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
129
130
131
132
133
134
135
136
137

use std::{io, os::fd::BorrowedFd};

#[cfg(not(feature = "tokio"))]
use async_process::{Child, ChildStdin, ChildStdout};

#[cfg(feature = "tokio")]
use tokio::{
    io::{AsyncReadExt, ReadBuf},
    process::{Child, ChildStdin, ChildStdout},
};

use super::{ReadHalf, RecvmsgResult, Socket, Split, WriteHalf};

/// A Command stream socket.
///
/// This socket communicates with a spawned child process via its standard input
/// and output streams.
#[derive(Debug)]
pub(crate) struct Command {
    stdin: ChildStdin,
    stdout: ChildStdout,
}

impl Command {
    fn into_split(self) -> (ChildStdout, ChildStdin) {
        (self.stdout, self.stdin)
    }
}

impl Socket for Command {
    type ReadHalf = ChildStdout;
    type WriteHalf = ChildStdin;

    fn split(self) -> Split<Self::ReadHalf, Self::WriteHalf> {
        let (read, write) = self.into_split();

        Split { read, write }
    }
}

impl TryFrom<&mut Child> for Command {
    type Error = crate::Error;

    fn try_from(child: &mut Child) -> Result<Self, Self::Error> {
        let stdin = child
            .stdin
            .take()
            .ok_or(crate::Error::Failure("child stdin not found".into()))?;

        let stdout = child
            .stdout
            .take()
            .ok_or(crate::Error::Failure("child stdout not found".into()))?;

        Ok(Command { stdin, stdout })
    }
}

#[cfg(not(feature = "tokio"))]
#[async_trait::async_trait]
impl ReadHalf for ChildStdout {
    async fn recvmsg(&mut self, buf: &mut [u8]) -> RecvmsgResult {
        match futures_util::AsyncReadExt::read(&mut self, buf).await {
            Err(e) => Err(e),
            Ok(len) => {
                #[cfg(unix)]
                let ret = (len, vec![]);
                #[cfg(not(unix))]
                let ret = len;
                Ok(ret)
            }
        }
    }
}

#[cfg(feature = "tokio")]
#[async_trait::async_trait]
impl ReadHalf for ChildStdout {
    async fn recvmsg(&mut self, buf: &mut [u8]) -> RecvmsgResult {
        let mut read_buf = ReadBuf::new(buf);
        self.read_buf(&mut read_buf).await.map(|_| {
            let ret = read_buf.filled().len();
            #[cfg(unix)]
            let ret = (ret, vec![]);

            ret
        })
    }
}

#[cfg(not(feature = "tokio"))]
#[async_trait::async_trait]
impl WriteHalf for ChildStdin {
    async fn sendmsg(
        &mut self,
        buf: &[u8],
        #[cfg(unix)] fds: &[BorrowedFd<'_>],
    ) -> io::Result<usize> {
        #[cfg(unix)]
        if !fds.is_empty() {
            return Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                "fds cannot be sent with a command stream",
            ));
        }

        futures_util::AsyncWriteExt::write(&mut self, buf).await
    }

    async fn close(&mut self) -> io::Result<()> {
        futures_util::AsyncWriteExt::close(&mut self).await
    }
}

#[cfg(feature = "tokio")]
#[async_trait::async_trait]
impl WriteHalf for ChildStdin {
    async fn sendmsg(
        &mut self,
        buf: &[u8],
        #[cfg(unix)] fds: &[BorrowedFd<'_>],
    ) -> io::Result<usize> {
        #[cfg(unix)]
        if !fds.is_empty() {
            return Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                "fds cannot be sent with a command stream",
            ));
        }

        tokio::io::AsyncWriteExt::write(&mut self, buf).await
    }

    async fn close(&mut self) -> io::Result<()> {
        tokio::io::AsyncWriteExt::shutdown(&mut self).await
    }
}