rust-ipfs 0.3.0-alpha

IPFS node implementation
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
110
111
112
113
114
115
116
117
118

use libp2p::core::muxing::StreamMuxerBox;
use libp2p::core::transport::timeout::TransportTimeout;
use libp2p::core::transport::upgrade::Version;
use libp2p::core::transport::{Boxed, OrTransport};
use libp2p::core::upgrade::SelectUpgrade;
use libp2p::dns::TokioDnsConfig;
use libp2p::identity;
use libp2p::mplex::MplexConfig;
use libp2p::noise::{self, NoiseConfig};
use libp2p::quic::tokio::Transport as TokioQuicTransport;
use libp2p::quic::Config as QuicConfig;
use libp2p::relay::v2::client::transport::ClientTransport;
use libp2p::swarm::derive_prelude::EitherOutput;
use libp2p::tcp::{tokio::Transport as TokioTcpTransport, Config as GenTcpConfig};
use libp2p::yamux::YamuxConfig;
use libp2p::{PeerId, Transport};
use std::io::{self, Error, ErrorKind};
use std::time::Duration;

/// Transport type.
pub(crate) type TTransport = Boxed<(PeerId, StreamMuxerBox)>;

#[derive(Debug, Clone)]
pub struct TransportConfig {
    pub yamux_config: YamuxConfig,
    pub mplex_config: MplexConfig,
    pub no_delay: bool,
    pub port_reuse: bool,
    pub timeout: Duration,
}

impl Default for TransportConfig {
    fn default() -> Self {
        Self {
            yamux_config: YamuxConfig::default(),
            mplex_config: MplexConfig::new(),
            no_delay: true,
            port_reuse: true,
            timeout: Duration::from_secs(30),
        }
    }
}

/// Builds the transport that serves as a common ground for all connections.
///
/// Set up an encrypted TCP transport over the Yamux and Mplex protocol.
pub fn build_transport(
    keypair: identity::Keypair,
    relay: Option<ClientTransport>,
    config: TransportConfig,
) -> io::Result<TTransport> {
    let TransportConfig {
        yamux_config,
        mplex_config,
        no_delay,
        port_reuse,
        timeout,
    } = config;
    let xx_keypair = noise::Keypair::<noise::X25519Spec>::new()
        .into_authentic(&keypair)
        .unwrap();
    let noise_config = NoiseConfig::xx(xx_keypair).into_authenticated();

    let multiplex_upgrade = SelectUpgrade::new(yamux_config, mplex_config);

    //TODO: Cleanup
    let tcp_config = GenTcpConfig::default()
        .nodelay(no_delay)
        .port_reuse(port_reuse);

    let mut quic_config = QuicConfig::new(&keypair);
    quic_config.handshake_timeout = Duration::from_secs(1);
    let quic_transport = TokioQuicTransport::new(quic_config);

    let tcp_transport = TokioTcpTransport::new(tcp_config.clone());
    let ws_transport = libp2p::websocket::WsConfig::new(TokioTcpTransport::new(tcp_config));

    let transport = tcp_transport.or_transport(ws_transport);

    let transport_timeout = TransportTimeout::new(transport, Duration::from_secs(30));
    #[cfg(not(target_os = "android"))]
    let transport = TokioDnsConfig::system(transport_timeout)?;
    #[cfg(target_os = "android")]
    let transport =
        TokioDnsConfig::custom(transport_timeout, Default::default(), Default::default())?;

    let transport = match relay {
        Some(relay) => {
            let transport = OrTransport::new(relay, transport);

            transport
                .upgrade(Version::V1)
                .authenticate(noise_config)
                .multiplex(multiplex_upgrade)
                .timeout(timeout)
                .map(|(peer_id, muxer), _| (peer_id, StreamMuxerBox::new(muxer)))
                .map_err(|err| Error::new(ErrorKind::Other, err))
                .boxed()
        }
        None => transport
            .upgrade(Version::V1)
            .authenticate(noise_config)
            .multiplex(multiplex_upgrade)
            .timeout(timeout)
            .map(|(peer_id, muxer), _| (peer_id, StreamMuxerBox::new(muxer)))
            .map_err(|err| Error::new(ErrorKind::Other, err))
            .boxed(),
    };

    let transport = OrTransport::new(quic_transport, transport)
        .map(|either_output, _| match either_output {
            EitherOutput::First((peer_id, muxer)) => (peer_id, StreamMuxerBox::new(muxer)),
            EitherOutput::Second((peer_id, muxer)) => (peer_id, StreamMuxerBox::new(muxer)),
        })
        .boxed();

    Ok(transport)
}