commonware-p2p 2026.4.0

Communicate with authenticated peers over encrypted connections.
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

use commonware_macros::select;
use commonware_utils::channel::mpsc::{self, error::TrySendError};

#[derive(Clone, Debug)]
pub struct Relay<T> {
    low: mpsc::Sender<T>,
    high: mpsc::Sender<T>,
}

impl<T> Relay<T> {
    pub const fn new(low: mpsc::Sender<T>, high: mpsc::Sender<T>) -> Self {
        Self { low, high }
    }

    /// Sends the given `message` to the appropriate channel based on `priority`.
    ///
    /// Uses non-blocking `try_send` to avoid blocking the caller when the
    /// channel buffer is full. Returns an error if the channel is full or
    /// disconnected.
    pub fn send(&self, message: T, priority: bool) -> Result<(), TrySendError<T>> {
        let sender = if priority { &self.high } else { &self.low };
        sender.try_send(message)
    }
}

/// Message received from one of the prioritized relay channels.
pub enum Prioritized<C, D> {
    /// Control message received from the control channel.
    Control(C),
    /// Data message received from either the high- or low-priority data channel.
    Data(D),
    /// One of the relay channels closed before yielding a message.
    Closed,
}

/// Awaits a message from control, high, or low priority receivers.
pub async fn recv_prioritized<C, D>(
    control: &mut mpsc::Receiver<C>,
    high: &mut mpsc::Receiver<D>,
    low: &mut mpsc::Receiver<D>,
) -> Prioritized<C, D> {
    select! {
        msg = control.recv() => msg.map_or(Prioritized::Closed, Prioritized::Control),
        msg = high.recv() => msg.map_or(Prioritized::Closed, Prioritized::Data),
        msg = low.recv() => msg.map_or(Prioritized::Closed, Prioritized::Data),
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_relay_content_priority() {
        let (low_sender, mut low_receiver) = mpsc::channel(1);
        let (high_sender, mut high_receiver) = mpsc::channel(1);
        let relay = Relay::new(low_sender, high_sender);

        // Send a high priority message
        let data = 123;
        relay.send(data, true).unwrap();
        match high_receiver.try_recv() {
            Ok(received_data) => {
                assert_eq!(data, received_data);
            }
            _ => panic!("Expected high priority message"),
        }
        assert!(low_receiver.try_recv().is_err());

        // Send a low priority message
        let data = 456;
        relay.send(data, false).unwrap();
        match low_receiver.try_recv() {
            Ok(received_data) => {
                assert_eq!(data, received_data);
            }
            _ => panic!("Expected low priority message"),
        }
        assert!(high_receiver.try_recv().is_err());
    }
}