aggligator 0.12.1

Aggregates multiple links (TCP, WebSocket, USB or similar) into one connection having their combined bandwidth and provides resiliency against failure of individual links, allowing failover and roaming between networks in user space.
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
106
107
108
109

//! Link aggregation implementation.

use bytes::Bytes;
use futures::{Sink, Stream};
use std::{
    fmt, io,
    sync::{Arc, atomic::AtomicBool},
};
use tokio::sync::{Mutex, mpsc, oneshot, watch};

use crate::{
    TaskError,
    agg::{link_int::LinkInt, task::Task},
    alc::{Channel, RecvError, SendError},
    cfg::{Cfg, ExchangedCfg},
    control::{Control, Direction, Link},
    id::{OwnedConnId, ServerId},
};

#[cfg(feature = "dump")]
pub mod dump;

pub(crate) mod link_int;
pub(crate) mod task;

/// Link aggregator parts.
pub(crate) struct AggParts<TX, RX, TAG> {
    pub task: Task<TX, RX, TAG>,
    pub channel: Channel,
    pub control: Control<TX, RX, TAG>,
    pub connected_rx: oneshot::Receiver<Arc<ExchangedCfg>>,
}

impl<TX, RX, TAG> AggParts<TX, RX, TAG>
where
    RX: Stream<Item = Result<Bytes, io::Error>> + Unpin + Send + 'static,
    TX: Sink<Bytes, Error = io::Error> + Unpin + Send + 'static,
    TAG: fmt::Display + Send + Sync + 'static,
{
    /// Creates a new aggregated connection and returns its parts.
    #[allow(clippy::type_complexity)]
    pub(crate) fn new(
        cfg: Arc<Cfg>, conn_id: OwnedConnId, direction: Direction, server_id: Option<ServerId>,
        remote_server_id: Option<ServerId>, links: Vec<LinkInt<TX, RX, TAG>>,
        link_tx_rx: Option<(mpsc::Sender<LinkInt<TX, RX, TAG>>, mpsc::Receiver<LinkInt<TX, RX, TAG>>)>,
    ) -> Self {
        let (terminate_tx, terminate_rx) = mpsc::channel(1);
        let (read_tx, read_rx) = mpsc::channel(cfg.recv_queue.get());
        let (write_tx, write_rx) = mpsc::channel(cfg.send_queue.get());
        let (read_error_tx, read_error_rx) = watch::channel(Some(RecvError::TaskTerminated));
        let (write_error_tx, write_error_rx) = watch::channel(SendError::TaskTerminated);
        let (read_closed_tx, read_closed_rx) = mpsc::channel(1);
        let (links_tx, links_rx) = watch::channel(links.iter().map(Link::from).collect());
        let (link_tx, link_rx) = link_tx_rx.unwrap_or_else(|| mpsc::channel(cfg.connect_queue.get()));
        let (connected_tx, connected_rx) = oneshot::channel();
        let (stats_tx, stats_rx) = watch::channel(Default::default());
        let (fatal_connect_error_tx, fatal_connect_error_rx) = mpsc::channel(1);
        let (result_tx, result_rx) = watch::channel(Err(TaskError::Terminated));
        let remote_cfg = links.first().as_ref().map(|link| link.remote_cfg());
        let connected = Arc::new(AtomicBool::new(!links.is_empty()));

        Self {
            task: Task::new(
                cfg.clone(),
                remote_cfg.clone(),
                conn_id.clone(),
                direction,
                terminate_rx,
                links_tx,
                link_rx,
                connected_tx,
                read_tx,
                read_closed_rx,
                write_rx,
                read_error_tx,
                write_error_tx,
                stats_tx,
                fatal_connect_error_rx,
                result_tx,
                links,
            ),
            channel: Channel::new(
                cfg.clone(),
                remote_cfg,
                conn_id.get(),
                write_tx,
                write_error_rx,
                read_rx,
                read_closed_tx,
                read_error_rx,
            ),
            control: Control {
                cfg,
                conn_id: conn_id.get(),
                server_id,
                remote_server_id: Arc::new(Mutex::new(remote_server_id)),
                direction,
                terminate_tx,
                link_tx,
                links_rx,
                connected,
                stats_rx,
                fatal_connect_error_tx,
                result_rx,
            },
            connected_rx,
        }
    }
}