aggligator 0.9.11

Aggregates multiple links (TCP or similar) into one connection having their combined bandwidth and provides resiliency against failure of individual links.
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

//! Peekable MPSC wrapper.

use tokio::sync::mpsc;

/// Receiver that allows peeking at next message.
#[derive(Debug)]
pub struct PeekableReceiver<T> {
    rx: mpsc::Receiver<T>,
    peeked: Option<T>,
}

impl<T> From<mpsc::Receiver<T>> for PeekableReceiver<T> {
    fn from(rx: mpsc::Receiver<T>) -> Self {
        Self::new(rx)
    }
}

impl<T> PeekableReceiver<T> {
    /// Creates a new peekable receiver.
    pub fn new(rx: mpsc::Receiver<T>) -> Self {
        Self { rx, peeked: None }
    }

    /// Receive next message.
    #[allow(dead_code)]
    pub async fn recv(&mut self) -> Option<T> {
        match self.peeked.take() {
            Some(msg) => Some(msg),
            None => self.rx.recv().await,
        }
    }

    /// Receives next message, if one is immediately available.
    pub fn try_recv(&mut self) -> Result<T, mpsc::error::TryRecvError> {
        match self.peeked.take() {
            Some(msg) => Ok(msg),
            None => self.rx.try_recv(),
        }
    }

    /// Peeks at the next message.
    pub async fn peek(&mut self) -> Option<&T> {
        if self.peeked.is_none() {
            self.peeked = self.rx.recv().await;
        }

        self.peeked.as_ref()
    }

    /// Peeks at the next message, if one is available.
    pub fn try_peek(&mut self) -> Result<&T, mpsc::error::TryRecvError> {
        if self.peeked.is_none() {
            self.peeked = Some(self.rx.try_recv()?);
        }

        Ok(self.peeked.as_ref().unwrap())
    }

    /// Receives the next messages if the condition is fulfilled.
    pub async fn recv_if(&mut self, cond: impl FnOnce(&T) -> bool) -> Result<T, RecvIfError> {
        match self.peek().await {
            Some(msg) if cond(msg) => Ok(self.try_recv().unwrap()),
            Some(_) => Err(RecvIfError::NoMatch),
            None => Err(RecvIfError::Disconnected),
        }
    }

    /// Receives the next messages if it is immediately available and the condition is fulfilled.
    pub fn try_recv_if(&mut self, cond: impl FnOnce(&T) -> bool) -> Result<T, TryRecvIfError> {
        if cond(self.try_peek()?) {
            Ok(self.try_recv().unwrap())
        } else {
            Err(TryRecvIfError::NoMatch)
        }
    }
}

/// Receive if condition fulfilled error.
#[derive(Debug, Clone)]
pub enum RecvIfError {
    /// Sender has been closed.
    Disconnected,
    /// Condition is not fulfilled.
    NoMatch,
}

/// Receive if condition fulfilled error.
#[derive(Debug, Clone)]
pub enum TryRecvIfError {
    /// No message is present.
    Empty,
    /// Sender has been closed.
    Disconnected,
    /// Condition is not fulfilled.
    NoMatch,
}

impl From<mpsc::error::TryRecvError> for TryRecvIfError {
    fn from(err: mpsc::error::TryRecvError) -> Self {
        match err {
            mpsc::error::TryRecvError::Empty => Self::Empty,
            mpsc::error::TryRecvError::Disconnected => Self::Disconnected,
        }
    }
}