mptcp_pm/
handle.rs

1// SPDX-License-Identifier: MIT
2
3use futures::{future::Either, FutureExt, Stream, StreamExt, TryStream};
4use genetlink::GenetlinkHandle;
5use netlink_packet_core::DecodeError;
6use netlink_packet_core::{NetlinkMessage, NLM_F_DUMP, NLM_F_REQUEST};
7use netlink_packet_generic::GenlMessage;
8
9use crate::{
10    try_mptcp, MptcpPathManagerAddressHandle, MptcpPathManagerCmd,
11    MptcpPathManagerError, MptcpPathManagerLimitsHandle,
12    MptcpPathManagerMessage,
13};
14
15#[derive(Clone, Debug)]
16pub struct MptcpPathManagerHandle {
17    pub handle: GenetlinkHandle,
18}
19
20impl MptcpPathManagerHandle {
21    pub(crate) fn new(handle: GenetlinkHandle) -> Self {
22        MptcpPathManagerHandle { handle }
23    }
24
25    // equivalent to `ip mptcp endpoint` command
26    // Instead of using `endpoint`, we are aligning with kernel netlink name
27    // `address` here.
28    pub fn address(&self) -> MptcpPathManagerAddressHandle {
29        MptcpPathManagerAddressHandle::new(self.clone())
30    }
31
32    // equivalent to `ip mptcp limits` command
33    pub fn limits(&self) -> MptcpPathManagerLimitsHandle {
34        MptcpPathManagerLimitsHandle::new(self.clone())
35    }
36
37    pub async fn request(
38        &mut self,
39        message: NetlinkMessage<GenlMessage<MptcpPathManagerMessage>>,
40    ) -> Result<
41        impl Stream<
42            Item = Result<
43                NetlinkMessage<GenlMessage<MptcpPathManagerMessage>>,
44                DecodeError,
45            >,
46        >,
47        MptcpPathManagerError,
48    > {
49        self.handle.request(message).await.map_err(|e| {
50            MptcpPathManagerError::RequestFailed(format!(
51                "BUG: Request failed with {e}"
52            ))
53        })
54    }
55}
56
57pub(crate) async fn mptcp_execute(
58    handle: &mut MptcpPathManagerHandle,
59    mptcp_msg: MptcpPathManagerMessage,
60) -> impl TryStream<
61    Ok = GenlMessage<MptcpPathManagerMessage>,
62    Error = MptcpPathManagerError,
63> {
64    let nl_header_flags = match mptcp_msg.cmd {
65        MptcpPathManagerCmd::AddressGet => NLM_F_REQUEST | NLM_F_DUMP,
66        MptcpPathManagerCmd::LimitsGet => NLM_F_REQUEST,
67    };
68
69    let mut nl_msg = NetlinkMessage::from(GenlMessage::from_payload(mptcp_msg));
70
71    nl_msg.header.flags = nl_header_flags;
72
73    match handle.request(nl_msg).await {
74        Ok(response) => {
75            Either::Left(response.map(move |msg| Ok(try_mptcp!(msg))))
76        }
77        Err(e) => Either::Right(
78            futures::future::err::<
79                GenlMessage<MptcpPathManagerMessage>,
80                MptcpPathManagerError,
81            >(e)
82            .into_stream(),
83        ),
84    }
85}
mptcp_pm/handle.rs

mptcp_pm/
handle.rs
