rns_net/common/link_manager/
channel_handling.rs

1use super::*;
2
3impl LinkManager {
4    /// Flush the channel TX ring for a link, clearing outstanding messages.
5    /// Called after holepunch completion where signaling messages are fire-and-forget.
6    pub fn flush_channel_tx(&mut self, link_id: &LinkId) {
7        if let Some(link) = self.links.get_mut(link_id) {
8            if let Some(ref mut channel) = link.channel {
9                channel.flush_tx();
10            }
11        }
12    }
13
14    /// Send a channel message on a link.
15    pub fn send_channel_message(
16        &mut self,
17        link_id: &LinkId,
18        msgtype: u16,
19        payload: &[u8],
20        rng: &mut dyn Rng,
21    ) -> Result<Vec<LinkManagerAction>, String> {
22        let link = match self.links.get_mut(link_id) {
23            Some(l) => l,
24            None => return Err("unknown link".to_string()),
25        };
26
27        let channel = match link.channel {
28            Some(ref mut ch) => ch,
29            None => return Err("link has no active channel".to_string()),
30        };
31
32        let link_mdu = link.engine.mdu();
33        let now = time::now();
34        let chan_actions = match channel.send(msgtype, payload, now, link_mdu) {
35            Ok(a) => {
36                link.channel_send_ok += 1;
37                a
38            }
39            Err(e) => {
40                log::debug!("Channel send failed: {:?}", e);
41                match e {
42                    rns_core::channel::ChannelError::NotReady => link.channel_send_not_ready += 1,
43                    rns_core::channel::ChannelError::MessageTooBig => {
44                        link.channel_send_too_big += 1;
45                    }
46                    rns_core::channel::ChannelError::InvalidEnvelope => {
47                        link.channel_send_other_error += 1;
48                    }
49                }
50                return Err(e.to_string());
51            }
52        };
53
54        let _ = link;
55        Ok(self.process_channel_actions(link_id, chan_actions, rng))
56    }
57
58    /// Convert ChannelActions to LinkManagerActions.
59    pub(super) fn process_channel_actions(
60        &mut self,
61        link_id: &LinkId,
62        actions: Vec<rns_core::channel::ChannelAction>,
63        rng: &mut dyn Rng,
64    ) -> Vec<LinkManagerAction> {
65        let mut result = Vec::new();
66        for action in actions {
67            match action {
68                rns_core::channel::ChannelAction::SendOnLink { raw, sequence } => {
69                    // Encrypt and send as CHANNEL context. If the packet cannot be
70                    // emitted, remove the reserved channel sequence so receivers do
71                    // not stall on a gap that never reached the wire.
72                    let encrypted = match self.links.get(link_id) {
73                        Some(link) => match link.engine.encrypt(&raw, rng) {
74                            Ok(encrypted) => encrypted,
75                            Err(_) => {
76                                if let Some(link_mut) = self.links.get_mut(link_id) {
77                                    if let Some(channel) = link_mut.channel.as_mut() {
78                                        channel.cancel_send(sequence);
79                                    }
80                                }
81                                continue;
82                            }
83                        },
84                        None => continue,
85                    };
86                    let flags = PacketFlags {
87                        header_type: constants::HEADER_1,
88                        context_flag: constants::FLAG_UNSET,
89                        transport_type: constants::TRANSPORT_BROADCAST,
90                        destination_type: constants::DESTINATION_LINK,
91                        packet_type: constants::PACKET_TYPE_DATA,
92                    };
93                    match RawPacket::pack_raw_with_hash(
94                        flags,
95                        0,
96                        link_id,
97                        None,
98                        constants::CONTEXT_CHANNEL,
99                        &encrypted,
100                    ) {
101                        Ok((raw_bytes, packet_hash)) => {
102                            if let Some(link_mut) = self.links.get_mut(link_id) {
103                                link_mut
104                                    .pending_channel_packets
105                                    .insert(packet_hash, sequence);
106                            }
107                            result.push(LinkManagerAction::SendPacket {
108                                raw: raw_bytes,
109                                dest_type: constants::DESTINATION_LINK,
110                                attached_interface: None,
111                            });
112                        }
113                        Err(_) => {
114                            if let Some(link_mut) = self.links.get_mut(link_id) {
115                                if let Some(channel) = link_mut.channel.as_mut() {
116                                    channel.cancel_send(sequence);
117                                }
118                            }
119                        }
120                    }
121                }
122                rns_core::channel::ChannelAction::MessageReceived {
123                    msgtype, payload, ..
124                } => {
125                    result.push(LinkManagerAction::ChannelMessageReceived {
126                        link_id: *link_id,
127                        msgtype,
128                        payload,
129                    });
130                }
131                rns_core::channel::ChannelAction::TeardownLink => {
132                    result.push(LinkManagerAction::LinkClosed {
133                        link_id: *link_id,
134                        reason: Some(TeardownReason::Timeout),
135                    });
136                }
137            }
138        }
139        result
140    }
141}
rns_net/common/link_manager/channel_handling.rs

rns_net/common/link_manager/
channel_handling.rs
