ts_netstack_smoltcp_core 0.2.0

command-channel-based userspace netstack built on smoltcp (core functionality)
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
138
139
140
141
142
143
144
145
146
147

//! Helpers to simplify sending requests.

use core::borrow::Borrow;

use smoltcp::iface::SocketHandle;

use crate::{
    Error,
    command::{Command, Request, Response},
};

/// Helper trait to abstract over [`flume::Sender`] and [`flume::WeakSender`].
pub trait UpgradableChannel {
    /// Attempt to upgrade this value to [`flume::Sender`].
    fn upgrade(&self) -> Option<impl Borrow<flume::Sender<Request>>>;

    /// Attempt to upgrade this value to [`flume::Sender`], returning
    /// [`Error::ChannelClosed`] on failure.
    fn upgrade_or_err(&self) -> Result<impl Borrow<flume::Sender<Request>>, Error> {
        self.upgrade().ok_or(Error::ChannelClosed)
    }
}

impl UpgradableChannel for flume::Sender<Request> {
    fn upgrade(&self) -> Option<impl Borrow<flume::Sender<Request>>> {
        Some(self)
    }
}

impl UpgradableChannel for flume::WeakSender<Request> {
    fn upgrade(&self) -> Option<impl Borrow<flume::Sender<Request>>> {
        self.upgrade()
    }
}

impl<T> UpgradableChannel for &T
where
    T: UpgradableChannel + ?Sized,
{
    fn upgrade(&self) -> Option<impl Borrow<flume::Sender<Request>>> {
        T::upgrade(self)
    }
}

/// Synchronously make a request over the given command channel.
///
/// Blocks on command submission and on receipt of a [`Response`].
pub fn request_blocking(
    command_tx: impl UpgradableChannel,
    handle: Option<SocketHandle>,
    command: impl Into<Command>,
) -> Result<Response, Error> {
    // wrapper to minimize monomorphization impact
    fn _request_blocking(
        command_tx: &flume::Sender<Request>,
        handle: Option<SocketHandle>,
        command: Command,
    ) -> Result<Response, Error> {
        let (resp_tx, resp_rx) = flume::bounded(1);

        command_tx.send(Request {
            handle,
            command,
            resp: resp_tx,
        })?;

        resp_rx.recv().map_err(Error::from)
    }

    let ch = command_tx.upgrade_or_err()?;
    let ch = ch.borrow();

    _request_blocking(ch, handle, command.into())
}

/// Make a request over the given command channel.
///
/// Blocks on command submission and the receipt of a [`Response`].
pub fn request(
    command_tx: impl UpgradableChannel,
    handle: Option<SocketHandle>,
    command: impl Into<Command>,
) -> impl Future<Output = Result<Response, Error>> + Send {
    // wrapper to minimize monomorphization impact
    async fn _request(
        command_tx: &flume::Sender<Request>,
        handle: Option<SocketHandle>,
        command: Command,
    ) -> Result<Response, Error> {
        let (resp_tx, resp_rx) = flume::bounded(1);

        command_tx
            .send_async(Request {
                handle,
                command,
                resp: resp_tx,
            })
            .await?;

        resp_rx.recv_async().await.map_err(Error::from)
    }

    // impl Future and the returned async block below are required to do this bit of work upgrading
    // the channel and converting the command outside the async context for lifetime reasons
    let ch = command_tx.upgrade_or_err().map(|x| x.borrow().clone());
    let command = command.into();

    async move {
        let ch = ch?;
        _request(&ch, handle, command).await
    }
}

/// Send a request without blocking on the response.
///
/// Returns an error only if the request can't be sent ([`Error::ChannelClosed`]).
///
/// Mainly intended for use in [`Drop`] implementations, as they can't be `async`
/// and it would be frequently be surprising if they blocked the calling thread.
pub fn request_nonblocking(
    command_tx: impl UpgradableChannel,
    handle: Option<SocketHandle>,
    command: impl Into<Command>,
) -> Result<(), Error> {
    // wrapper to minimize monomorphization impact
    fn _request_nonblocking(
        command_tx: &flume::Sender<Request>,
        handle: Option<SocketHandle>,
        command: Command,
    ) -> Result<(), Error> {
        let (resp_tx, _resp_rx) = flume::bounded(1);

        match command_tx.try_send(Request {
            handle,
            command,
            resp: resp_tx,
        }) {
            Ok(_) | Err(flume::TrySendError::Full(_)) => Ok(()),
            Err(flume::TrySendError::Disconnected(_)) => Err(Error::ChannelClosed),
        }
    }

    let ch = command_tx.upgrade_or_err()?;
    let ch = ch.borrow();

    _request_nonblocking(ch, handle, command.into())
}