mptcp-pm 0.1.1

Linux kernel MPTCP path manager netlink Library
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

// SPDX-License-Identifier: MIT

use futures::{future::Either, FutureExt, Stream, StreamExt, TryStream};
use genetlink::GenetlinkHandle;
use netlink_packet_core::{NetlinkMessage, NLM_F_DUMP, NLM_F_REQUEST};
use netlink_packet_generic::GenlMessage;
use netlink_packet_utils::DecodeError;

use crate::{
    try_mptcp,
    MptcpPathManagerAddressHandle,
    MptcpPathManagerCmd,
    MptcpPathManagerError,
    MptcpPathManagerLimitsHandle,
    MptcpPathManagerMessage,
};

#[derive(Clone, Debug)]
pub struct MptcpPathManagerHandle {
    pub handle: GenetlinkHandle,
}

impl MptcpPathManagerHandle {
    pub(crate) fn new(handle: GenetlinkHandle) -> Self {
        MptcpPathManagerHandle { handle }
    }

    // equivalent to `ip mptcp endpoint` command
    // Instead of using `endpoint`, we are aligning with kernel netlink name
    // `address` here.
    pub fn address(&self) -> MptcpPathManagerAddressHandle {
        MptcpPathManagerAddressHandle::new(self.clone())
    }

    // equivalent to `ip mptcp limits` command
    pub fn limits(&self) -> MptcpPathManagerLimitsHandle {
        MptcpPathManagerLimitsHandle::new(self.clone())
    }

    pub async fn request(
        &mut self,
        message: NetlinkMessage<GenlMessage<MptcpPathManagerMessage>>,
    ) -> Result<
        impl Stream<Item = Result<NetlinkMessage<GenlMessage<MptcpPathManagerMessage>>, DecodeError>>,
        MptcpPathManagerError,
    > {
        self.handle.request(message).await.map_err(|e| {
            MptcpPathManagerError::RequestFailed(format!("BUG: Request failed with {}", e))
        })
    }
}

pub(crate) async fn mptcp_execute(
    handle: &mut MptcpPathManagerHandle,
    mptcp_msg: MptcpPathManagerMessage,
) -> impl TryStream<Ok = GenlMessage<MptcpPathManagerMessage>, Error = MptcpPathManagerError> {
    let nl_header_flags = match mptcp_msg.cmd {
        MptcpPathManagerCmd::AddressGet => NLM_F_REQUEST | NLM_F_DUMP,
        MptcpPathManagerCmd::LimitsGet => NLM_F_REQUEST,
    };

    let mut nl_msg = NetlinkMessage::from(GenlMessage::from_payload(mptcp_msg));

    nl_msg.header.flags = nl_header_flags;

    match handle.request(nl_msg).await {
        Ok(response) => Either::Left(response.map(move |msg| Ok(try_mptcp!(msg)))),
        Err(e) => Either::Right(
            futures::future::err::<GenlMessage<MptcpPathManagerMessage>, MptcpPathManagerError>(e)
                .into_stream(),
        ),
    }
}