diff --git a/example/examples/echo-dialer.rs b/example/examples/echo-dialer.rs index 7e7730b6..611b6fe3 100644 --- a/example/examples/echo-dialer.rs +++ b/example/examples/echo-dialer.rs @@ -26,15 +26,11 @@ extern crate libp2p_tcp_transport as tcp; extern crate tokio_core; extern crate tokio_io; -use bytes::Bytes; -use futures::future::{Future, FutureResult, IntoFuture}; -use futures::{Stream, Sink}; -use std::io::Error as IoError; -use std::iter; -use swarm::{Transport, ConnectionUpgrade}; +use bytes::BytesMut; +use futures::{Stream, Sink, Future}; +use swarm::Transport; use tcp::TcpConfig; use tokio_core::reactor::Core; -use tokio_io::{AsyncRead, AsyncWrite}; use tokio_io::codec::length_delimited; fn main() { @@ -51,11 +47,12 @@ fn main() { } }); - let with_echo = with_secio.with_upgrade(Echo); + let with_echo = with_secio.with_simple_protocol_upgrade("/echo/1.0.0", |socket| { + Ok(length_delimited::Framed::<_, BytesMut>::new(socket)) + }); let dialer = with_echo.dial(swarm::multiaddr::Multiaddr::new("/ip4/127.0.0.1/tcp/10333").unwrap()) - .map_err(|_| panic!()) - .unwrap() + .unwrap_or_else(|_| panic!()) .and_then(|f| { f.send("hello world".into()) }) @@ -70,26 +67,3 @@ fn main() { core.run(dialer).unwrap(); } - -// TODO: copy-pasted from echo-server -#[derive(Debug, Copy, Clone)] -pub struct Echo; -impl ConnectionUpgrade for Echo - where C: AsyncRead + AsyncWrite -{ - type NamesIter = iter::Once<(Bytes, Self::UpgradeIdentifier)>; - type UpgradeIdentifier = (); - - #[inline] - fn protocol_names(&self) -> Self::NamesIter { - iter::once(("/echo/1.0.0".into(), ())) - } - - type Output = length_delimited::Framed; - type Future = FutureResult; - - #[inline] - fn upgrade(self, socket: C, _: Self::UpgradeIdentifier) -> Self::Future { - Ok(length_delimited::Framed::new(socket)).into_future() - } -} diff --git a/example/examples/echo-server.rs b/example/examples/echo-server.rs index 12e63f6b..233376b8 100644 --- a/example/examples/echo-server.rs +++ b/example/examples/echo-server.rs @@ -26,15 +26,14 @@ extern crate libp2p_tcp_transport as tcp; extern crate tokio_core; extern crate tokio_io; -use bytes::Bytes; -use futures::future::{Future, FutureResult, IntoFuture, loop_fn, Loop}; +use bytes::BytesMut; +use futures::future::{Future, IntoFuture, loop_fn, Loop}; use futures::{Stream, Sink}; use std::io::Error as IoError; use std::iter; use swarm::{Transport, ConnectionUpgrade}; use tcp::TcpConfig; use tokio_core::reactor::Core; -use tokio_io::{AsyncRead, AsyncWrite}; use tokio_io::codec::length_delimited; fn main() { @@ -51,11 +50,12 @@ fn main() { } }); - let with_echo = with_secio.with_upgrade(Echo); + let with_echo = with_secio.with_simple_protocol_upgrade("/echo/1.0.0", |socket| { + Ok(length_delimited::Framed::<_, BytesMut>::new(socket)) + }); let future = with_echo.listen_on(swarm::multiaddr::Multiaddr::new("/ip4/0.0.0.0/tcp/10333").unwrap()) - .map_err(|_| panic!()) - .unwrap().0 + .unwrap_or_else(|_| panic!()).0 .for_each(|(socket, _)| { loop_fn(socket, |socket| { socket.into_future() @@ -72,26 +72,3 @@ fn main() { core.run(future).unwrap(); } - -// TODO: copy-pasted from echo-dialer -#[derive(Debug, Copy, Clone)] -pub struct Echo; -impl ConnectionUpgrade for Echo - where C: AsyncRead + AsyncWrite -{ - type NamesIter = iter::Once<(Bytes, Self::UpgradeIdentifier)>; - type UpgradeIdentifier = (); - - #[inline] - fn protocol_names(&self) -> Self::NamesIter { - iter::once(("/echo/1.0.0".into(), ())) - } - - type Output = length_delimited::Framed; - type Future = FutureResult; - - #[inline] - fn upgrade(self, socket: C, _: Self::UpgradeIdentifier) -> Self::Future { - Ok(length_delimited::Framed::new(socket)).into_future() - } -} diff --git a/libp2p-swarm/src/transport.rs b/libp2p-swarm/src/transport.rs index 87c5ae0e..877cf29b 100644 --- a/libp2p-swarm/src/transport.rs +++ b/libp2p-swarm/src/transport.rs @@ -31,11 +31,12 @@ use bytes::Bytes; use futures::{Stream, Poll, Async}; -use futures::future::{IntoFuture, Future, ok as future_ok, FutureResult}; +use futures::future::{IntoFuture, Future, ok as future_ok, FutureResult, FromErr}; use multiaddr::Multiaddr; use multistream_select; use std::io::{Cursor, Error as IoError, Read, Write}; use std::iter; +use std::sync::Arc; use tokio_io::{AsyncRead, AsyncWrite}; /// A transport is an object that can be used to produce connections by listening or dialing a @@ -101,6 +102,25 @@ pub trait Transport { upgrade: upgrade, } } + + /// Wraps this transport inside an upgrade. Similar to `with_upgrade`, but more convenient to + /// use for small protocols. + #[inline] + fn with_simple_protocol_upgrade(self, name: N, upgrade: F) + -> UpgradedNode> + where Self: Sized, + N: Into, + F: Fn(Self::RawConn) -> O, + O: IntoFuture, + { + UpgradedNode { + transports: self, + upgrade: SimpleProtocolUpgrade { + name: name.into(), + upgrade: Arc::new(upgrade), + }, + } + } } /// Dummy implementation of `Transport` that just denies every single attempt. @@ -164,6 +184,48 @@ impl Transport for OrTransport } } +/// Implementation of `ConnetionUpgrade`. See `with_simple_protocol_upgrade`. +#[derive(Debug)] +pub struct SimpleProtocolUpgrade { + name: Bytes, + // Note: we put the closure `F` in an `Arc` because Rust closures aren't automatically clonable + // yet. + upgrade: Arc, +} + +impl Clone for SimpleProtocolUpgrade { + #[inline] + fn clone(&self) -> Self { + SimpleProtocolUpgrade { + name: self.name.clone(), + upgrade: self.upgrade.clone(), + } + } +} + +impl ConnectionUpgrade for SimpleProtocolUpgrade + where C: AsyncRead + AsyncWrite, + F: Fn(C) -> O, + O: IntoFuture, +{ + type NamesIter = iter::Once<(Bytes, ())>; + type UpgradeIdentifier = (); + + #[inline] + fn protocol_names(&self) -> Self::NamesIter { + iter::once(("/echo/1.0.0".into(), ())) + } + + type Output = O::Item; + type Future = FromErr; + + #[inline] + fn upgrade(self, socket: C, _: ()) -> Self::Future { + let upgrade = &self.upgrade; + upgrade(socket).into_future().from_err() + } +} + /// Implements `Stream` and dispatches all method calls to either `First` or `Second`. #[derive(Debug, Copy, Clone)] pub enum EitherStream {