Implement a transport for COMIT

Author: Tobin C. Harding

Cargo.lock

@@ -24,6 +24,7 @@ use libp2p::{
     identity::{self, ed25519},
     PeerId, Swarm,
 };
+use libp2p_comit;
 use rand::rngs::OsRng;
 use std::{
     net::SocketAddr,
@@ -84,7 +85,7 @@ fn main() -> anyhow::Result<()> {
         .compat(),
     )?;
 
-    let transport = libp2p::build_development_transport(local_key_pair);
+    let transport = libp2p_comit::build_comit_transport(local_key_pair);
     let behaviour = network::ComitNode::new(
         ledger_events.clone(),
         Arc::clone(&state_store),

cnd/src/main.rs

@@ -9,6 +9,7 @@ description = "Implementation of the COMIT messaging protocol"
 bytes = "0.4"
 derivative = "1.0.3"
 futures = "0.1"
+libp2p = "0.13"
 libp2p-core = "0.13"
 libp2p-swarm = "0.3"
 log = "0.4"

libp2p-comit/Cargo.toml

@@ -9,6 +9,7 @@ mod protocol;
 mod substream;
 #[cfg(test)]
 pub mod test_harness;
+mod transport;
 
 use serde::{Deserialize, Serialize};
 use serde_json::{self, Value as JsonValue};
@@ -17,6 +18,7 @@ pub use self::{
     behaviour::{BehaviourOutEvent, Comit},
     handler::{ComitHandler, PendingInboundRequest, PendingOutboundRequest},
     protocol::{ComitProtocolConfig, Frames},
+    transport::build_comit_transport,
 };
 use crate::handler::{ProtocolOutEvent, ProtocolOutboundOpenInfo};
 use libp2p_swarm::ProtocolsHandlerEvent;

libp2p-comit/src/lib.rs

@@ -0,0 +1,93 @@
+use libp2p::{
+    dns, identity, mplex, secio, tcp, yamux, Multiaddr, PeerId, Transport, TransportError,
+};
+use libp2p_core::{muxing, upgrade};
+use std::{error, io, time::Duration};
+
+/// Builds a `Transport` that supports the most commonly-used protocols that
+/// libp2p supports.
+pub fn build_comit_transport(
+    keypair: identity::Keypair,
+) -> impl Transport<
+    Output = (
+        PeerId,
+        impl muxing::StreamMuxer<
+                OutboundSubstream = impl Send,
+                Substream = impl Send,
+                Error = impl Into<io::Error>,
+            > + Send
+            + Sync,
+    ),
+    Error = impl error::Error + Send,
+    Listener = impl Send,
+    Dial = impl Send,
+    ListenerUpgrade = impl Send,
+> + Clone {
+    build_tcp_ws_secio_mplex_yamux(keypair)
+}
+
+/// Builds an implementation of `Transport` that is suitable for usage with the
+/// `Swarm`.
+///
+/// The implementation supports TCP/IP, secio as the encryption layer, and mplex
+/// or yamux as the multiplexing layer.
+pub fn build_tcp_ws_secio_mplex_yamux(
+    keypair: identity::Keypair,
+) -> impl Transport<
+    Output = (
+        PeerId,
+        impl muxing::StreamMuxer<
+                OutboundSubstream = impl Send,
+                Substream = impl Send,
+                Error = impl Into<io::Error>,
+            > + Send
+            + Sync,
+    ),
+    Error = impl error::Error + Send,
+    Listener = impl Send,
+    Dial = impl Send,
+    ListenerUpgrade = impl Send,
+> + Clone {
+    ComitTransport::new()
+        .upgrade(upgrade::Version::V1)
+        .authenticate(secio::SecioConfig::new(keypair))
+        .multiplex(upgrade::SelectUpgrade::new(
+            yamux::Config::default(),
+            mplex::MplexConfig::new(),
+        ))
+        .map(|(peer, muxer), _| (peer, muxing::StreamMuxerBox::new(muxer)))
+        .timeout(Duration::from_secs(20))
+}
+
+#[derive(Debug, Clone)]
+struct ComitTransport {
+    inner: InnerImplementation,
+}
+
+type InnerImplementation = dns::DnsConfig<tcp::TcpConfig>;
+
+impl ComitTransport {
+    /// Initializes the `ComitTransport`.
+    pub fn new() -> ComitTransport {
+        let tcp = tcp::TcpConfig::new().nodelay(true);
+        let transport = dns::DnsConfig::new(tcp);
+
+        ComitTransport { inner: transport }
+    }
+}
+
+impl Transport for ComitTransport {
+    type Output = <InnerImplementation as Transport>::Output;
+    type Error = <InnerImplementation as Transport>::Error;
+    type Listener = <InnerImplementation as Transport>::Listener;
+    type ListenerUpgrade = <InnerImplementation as Transport>::ListenerUpgrade;
+    type Dial = <InnerImplementation as Transport>::Dial;
+
+    fn listen_on(self, addr: Multiaddr) -> Result<Self::Listener, TransportError<Self::Error>> {
+        self.inner.listen_on(addr)
+    }
+
+    fn dial(self, addr: Multiaddr) -> Result<Self::Dial, TransportError<Self::Error>> {
+        self.inner.dial(addr)
+    }
+}