zbus 4.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

use std::io;

use async_broadcast::{broadcast, Receiver, Sender};

use crate::{fdo::ConnectionCredentials, Message};

/// An in-process channel-based socket.
///
/// This is a pair of two cross-wired channels. Since all communication happens in-process, there is
/// no need for any authentication.
///
/// This type is only available when `p2p` feature is enabled.
#[derive(Debug)]
pub struct Channel {
    writer: Writer,
    reader: Reader,
}

impl Channel {
    /// Create a pair of cross-wired channels.
    ///
    /// Use [`crate::connection::Builder::authenticated_socket`] to create `Connection` instances
    /// from each channel.
    pub fn pair() -> (Self, Self) {
        let (tx1, rx1) = broadcast(CHANNEL_CAPACITY);
        let (tx2, rx2) = broadcast(CHANNEL_CAPACITY);

        (
            Self {
                writer: Writer(tx1),
                reader: Reader(rx2),
            },
            Self {
                writer: Writer(tx2),
                reader: Reader(rx1),
            },
        )
    }
}

impl super::Socket for Channel {
    type ReadHalf = Reader;
    type WriteHalf = Writer;

    fn split(self) -> super::Split<Self::ReadHalf, Self::WriteHalf> {
        super::Split {
            read: self.reader,
            write: self.writer,
        }
    }
}

/// The reader half of a [`Channel`].
///
/// This type is only available when `p2p` feature is enabled.
#[derive(Debug)]
pub struct Reader(Receiver<Message>);

#[async_trait::async_trait]
impl super::ReadHalf for Reader {
    async fn receive_message(
        &mut self,
        _seq: u64,
        _already_received_bytes: &mut Vec<u8>,
        #[cfg(unix)] _already_received_fds: &mut Vec<std::os::fd::OwnedFd>,
    ) -> crate::Result<Message> {
        self.0.recv().await.map_err(|e| {
            crate::Error::InputOutput(io::Error::new(io::ErrorKind::BrokenPipe, e).into())
        })
    }

    async fn peer_credentials(&mut self) -> io::Result<ConnectionCredentials> {
        self_credentials().await
    }
}

/// The writer half of a [`Channel`].
///
/// This type is only available when `p2p` feature is enabled.
#[derive(Debug)]
pub struct Writer(Sender<Message>);

#[async_trait::async_trait]
impl super::WriteHalf for Writer {
    async fn send_message(&mut self, msg: &Message) -> crate::Result<()> {
        self.0
            .broadcast_direct(msg.clone())
            .await
            .map_err(|e| {
                crate::Error::InputOutput(io::Error::new(io::ErrorKind::BrokenPipe, e).into())
            })
            .map(|removed| {
                // We don't enable `overflow` mode so items should never be removed.
                assert!(removed.is_none());
            })
    }

    async fn close(&mut self) -> io::Result<()> {
        self.0.close();

        Ok(())
    }

    async fn peer_credentials(&mut self) -> io::Result<ConnectionCredentials> {
        self_credentials().await
    }
}

/// Return the credentials of the current process.
async fn self_credentials() -> io::Result<ConnectionCredentials> {
    let mut creds = ConnectionCredentials::default().set_process_id(std::process::id());

    #[cfg(unix)]
    {
        use nix::unistd::{Gid, Uid};

        creds = creds
            .set_unix_user_id(Uid::effective().into())
            .add_unix_group_id(Gid::effective().into());
    }
    #[cfg(windows)]
    {
        let sid = crate::win32::ProcessToken::open(None)?.sid()?;
        creds = creds.set_windows_sid(sid);
    }

    Ok(creds)
}

const CHANNEL_CAPACITY: usize = 32;