fluencelabs/rust-libp2p
1
0
Fork 0
mirror of https://github.com/fluencelabs/rust-libp2p synced 2025-05-28 18:21:20 +00:00
Code Issues Projects Releases Wiki Activity

Fix concerns

Browse Source
This commit is contained in:
Pierre Krieger 2017-12-18 12:40:16 +01:00
parent 1da2c5015c
commit 952e3926c8
3 changed files with 31 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

43
example/examples/echo-server.rs
View File

@ -92,35 +92,38 @@ fn main() {
// This closure is called whenever a new connection has been received and successfully
// upgraded to use secio/plaintext and echo.
println!("Successfully negotiated protocol with {}", client_addr);
let client_addr2 = client_addr.clone();
// We loop forever in order to handle all the messages sent by the client.
let client_finished = loop_fn(socket, move |socket| {
let client_finished = {
let client_addr = client_addr.clone();
socket.into_future()
.map_err(|(err, _)| err)
.and_then(move |(msg, rest)| {
if let Some(msg) = msg {
// One message has been received. We send it back to the client.
println!("Received a message from {}: {:?}\n => Sending back \
identical message to remote", client_addr, msg);
Box::new(rest.send(msg).map(|m| Loop::Continue(m)))
as Box<Future<Item = _, Error = _>>
} else {
// End of stream. Connection closed. Breaking the loop.
println!("Received EOF from {}\n => Dropping connection", client_addr);
Box::new(Ok(Loop::Break(())).into_future())
as Box<Future<Item = _, Error = _>>
}
})
});
loop_fn(socket, move |socket| {
let client_addr = client_addr.clone();
socket.into_future()
.map_err(|(err, _)| err)
.and_then(move |(msg, rest)| {
if let Some(msg) = msg {
// One message has been received. We send it back to the client.
println!("Received a message from {}: {:?}\n => Sending back \
identical message to remote", client_addr, msg);
Box::new(rest.send(msg).map(|m| Loop::Continue(m)))
as Box<Future<Item = _, Error = _>>
} else {
// End of stream. Connection closed. Breaking the loop.
println!("Received EOF from {}\n => Dropping connection",
client_addr);
Box::new(Ok(Loop::Break(())).into_future())
as Box<Future<Item = _, Error = _>>
}
})
})
};
// We absorb errors from the `client_finished` future so that an error while processing
// a client (eg. if the client unexpectedly disconnects) doesn't propagate and stop the
// entire server.
client_finished.then(move |res| {
if let Err(err) = res {
println!("Error while processing client {}: {:?}", client_addr2, err);
println!("Error while processing client {}: {:?}", client_addr, err);
}
Ok(())
})

5
libp2p-swarm/src/connection_reuse.rs
View File

@ -164,8 +164,9 @@ where
/// Implementation of `Stream<Item = (impl AsyncRead + AsyncWrite, Multiaddr)` for the
/// `ConnectionReuse` struct.
pub struct ConnectionReuseListener<S, M>
where S: Stream<Item = (Result<M, IoError>, Multiaddr), Error = IoError>,
M: StreamMuxer
where
S: Stream<Item = (Result<M, IoError>, Multiaddr), Error = IoError>,
M: StreamMuxer
{
listener: StreamFuse<S>,
connections: Vec<(M, <M as StreamMuxer>::InboundSubstream, Multiaddr)>,

10
libp2p-swarm/src/transport.rs
View File

@ -305,12 +305,12 @@ pub enum EitherListenStream<A, B> {
Second(B),
}
impl<A, B, Sa, Sb> Stream for EitherListenStream<A, B>
impl<AStream, BStream, AInner, BInner> Stream for EitherListenStream<AStream, BStream>
where
A: Stream<Item = (Result<Sa, IoError>, Multiaddr), Error = IoError>,
B: Stream<Item = (Result<Sb, IoError>, Multiaddr), Error = IoError>,
AStream: Stream<Item = (Result<AInner, IoError>, Multiaddr), Error = IoError>,
BStream: Stream<Item = (Result<BInner, IoError>, Multiaddr), Error = IoError>,
{
type Item = (Result<EitherSocket<Sa, Sb>, IoError>, Multiaddr);
type Item = (Result<EitherSocket<AInner, BInner>, IoError>, Multiaddr);
type Error = IoError;
#[inline]
@ -869,7 +869,7 @@ where
// Try to negotiate the protocol.
// Note that failing to negotiate a protocol will never produce a future with an error.
// Instead the `stream` will produce an `Ok(Err(...))`.
// Instead the `stream` will produce `Ok(Err(...))`.
// `stream` can only produce an `Err` if `listening_stream` produces an `Err`.
let stream = listening_stream
// Try to negotiate the protocol