fluencelabs/rust-libp2p
1
0
Fork 0
mirror of https://github.com/fluencelabs/rust-libp2p synced 2025-06-24 15:21:33 +00:00
Code Issues Projects Releases Wiki Activity

Allow StreamMuxer to notify changes in the address (#1621)

Browse Source 
* Allow StreamMuxer to notify changes in the address

* Fix doc link

* Revert accidental rename

* Other accidental rename

Co-authored-by: Roman Borschel <romanb@users.noreply.github.com>
This commit is contained in:
Pierre Krieger
2020-06-30 17:10:53 +02:00
committed by GitHub
parent d5b26359bb
commit 826f5130cd
23 changed files with 330 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
core/src/connection.rs
View File

@ -132,6 +132,16 @@ impl ConnectedPoint {
ConnectedPoint::Listener { .. } => true
}
}
/// Modifies the address of the remote stored in this struct.
///
/// For `Dialer`, this modifies `address`. For `Listener`, this modifies `send_back_addr`.
pub fn set_remote_address(&mut self, new_address: Multiaddr) {
match self {
ConnectedPoint::Dialer { address } => *address = new_address,
ConnectedPoint::Listener { send_back_addr, .. } => *send_back_addr = new_address,
}
}
}
/// Information about a successfully established connection.
@ -169,6 +179,15 @@ impl ConnectionInfo for PeerId {
}
}
/// Event generated by a [`Connection`].
#[derive(Debug, Clone)]
pub enum Event<T> {
/// Event generated by the [`ConnectionHandler`].
Handler(T),
/// Address of the remote has changed.
AddressChange(Multiaddr),
}
/// A multiplexed connection to a peer with an associated `ConnectionHandler`.
pub struct Connection<TMuxer, THandler>
where
@ -239,7 +258,7 @@ where
/// Polls the connection for events produced by the associated handler
/// as a result of I/O activity on the substream multiplexer.
pub fn poll(mut self: Pin<&mut Self>, cx: &mut Context)
-> Poll<Result<THandler::OutEvent, ConnectionError<THandler::Error>>>
-> Poll<Result<Event<THandler::OutEvent>, ConnectionError<THandler::Error>>>
{
loop {
let mut io_pending = false;
@ -255,6 +274,10 @@ where
let endpoint = SubstreamEndpoint::Dialer(user_data);
self.handler.inject_substream(substream, endpoint)
}
Poll::Ready(Ok(SubstreamEvent::AddressChange(address))) => {
self.handler.inject_address_change(&address);
return Poll::Ready(Ok(Event::AddressChange(address)));
}
Poll::Ready(Err(err)) => return Poll::Ready(Err(ConnectionError::IO(err))),
}
@ -269,7 +292,7 @@ where
self.muxing.open_substream(user_data);
}
Poll::Ready(Ok(ConnectionHandlerEvent::Custom(event))) => {
return Poll::Ready(Ok(event));
return Poll::Ready(Ok(Event::Handler(event)));
}
Poll::Ready(Err(err)) => return Poll::Ready(Err(ConnectionError::Handler(err))),
}

5
core/src/connection/handler.rs
View File

@ -18,7 +18,7 @@
// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
// DEALINGS IN THE SOFTWARE.
use crate::PeerId;
use crate::{Multiaddr, PeerId};
use std::{task::Context, task::Poll};
use super::{Connected, SubstreamEndpoint};
@ -58,6 +58,9 @@ pub trait ConnectionHandler {
/// Notifies the handler of an event.
fn inject_event(&mut self, event: Self::InEvent);
/// Notifies the handler of a change in the address of the remote.
fn inject_address_change(&mut self, new_address: &Multiaddr);
/// Polls the handler for events.
///
/// Returning an error will close the connection to the remote.

31
core/src/connection/manager.rs
View File

@ -32,11 +32,13 @@ use std::{
collections::hash_map,
error,
fmt,
mem,
pin::Pin,
task::{Context, Poll},
};
use super::{
Connected,
ConnectedPoint,
Connection,
ConnectionError,
ConnectionHandler,
@ -220,7 +222,17 @@ pub enum Event<'a, I, O, H, TE, HE, C> {
entry: EstablishedEntry<'a, I, C>,
/// The produced event.
event: O
}
},
/// A connection to a node has changed its address.
AddressChange {
/// The entry associated with the connection that changed address.
entry: EstablishedEntry<'a, I, C>,
/// The former [`ConnectedPoint`].
old_endpoint: ConnectedPoint,
/// The new [`ConnectedPoint`].
new_endpoint: ConnectedPoint,
},
}
impl<I, O, H, TE, HE, C> Manager<I, O, H, TE, HE, C> {
@ -369,6 +381,23 @@ impl<I, O, H, TE, HE, C> Manager<I, O, H, TE, HE, C> {
let _ = task.remove();
Event::PendingConnectionError { id, error, handler }
}
task::Event::AddressChange { id: _, new_address } => {
let (new, old) = if let TaskState::Established(c) = &mut task.get_mut().state {
let mut new_endpoint = c.endpoint.clone();
new_endpoint.set_remote_address(new_address);
let old_endpoint = mem::replace(&mut c.endpoint, new_endpoint.clone());
(new_endpoint, old_endpoint)
} else {
unreachable!(
"`Event::AddressChange` implies (2) occurred on that task and thus (3)."
)
};
Event::AddressChange {
entry: EstablishedEntry { task },
old_endpoint: old,
new_endpoint: new,
}
},
task::Event::Error { id, error } => {
let id = ConnectionId(id);
let task = task.remove();

18
core/src/connection/manager/task.rs
View File

@ -19,8 +19,10 @@
// DEALINGS IN THE SOFTWARE.
use crate::{
Multiaddr,
muxing::StreamMuxer,
connection::{
self,
Close,
Connected,
Connection,
@ -55,8 +57,10 @@ pub enum Event<T, H, TE, HE, C> {
Error { id: TaskId, error: ConnectionError<HE> },
/// A pending connection failed.
Failed { id: TaskId, error: PendingConnectionError<TE>, handler: H },
/// A node we are connected to has changed its address.
AddressChange { id: TaskId, new_address: Multiaddr },
/// Notify the manager of an event from the connection.
Notify { id: TaskId, event: T }
Notify { id: TaskId, event: T },
}
impl<T, H, TE, HE, C> Event<T, H, TE, HE, C> {
@ -64,8 +68,9 @@ impl<T, H, TE, HE, C> Event<T, H, TE, HE, C> {
match self {
Event::Established { id, .. } => id,
Event::Error { id, .. } => id,
Event::Notify { id, .. } => id,
Event::Failed { id, .. } => id,
Event::AddressChange { id, .. } => id,
Event::Notify { id, .. } => id,
}
}
}
@ -245,13 +250,20 @@ where
this.state = State::EstablishedPending(connection);
return Poll::Pending
}
Poll::Ready(Ok(event)) => {
Poll::Ready(Ok(connection::Event::Handler(event))) => {
this.state = State::EstablishedReady {
connection: Some(connection),
event: Event::Notify { id, event }
};
continue 'poll
}
Poll::Ready(Ok(connection::Event::AddressChange(new_address))) => {
this.state = State::EstablishedReady {
connection: Some(connection),
event: Event::AddressChange { id, new_address }
};
continue 'poll
}
Poll::Ready(Err(error)) => {
// Notify the manager of the error via an event,
// dropping the connection.

39
core/src/connection/pool.rs
View File

@ -125,6 +125,16 @@ pub enum PoolEvent<'a, TInEvent, TOutEvent, THandler, TTransErr, THandlerErr, TC
/// The produced event.
event: TOutEvent,
},
/// The connection to a node has changed its address.
AddressChange {
/// The connection that has changed address.
connection: EstablishedConnection<'a, TInEvent, TConnInfo, TPeerId>,
/// The new endpoint.
new_endpoint: ConnectedPoint,
/// The old endpoint.
old_endpoint: ConnectedPoint,
},
}
impl<'a, TInEvent, TOutEvent, THandler, TTransErr, THandlerErr, TConnInfo, TPeerId> fmt::Debug
@ -162,6 +172,13 @@ where
.field("event", event)
.finish()
},
PoolEvent::AddressChange { ref connection, ref new_endpoint, ref old_endpoint } => {
f.debug_struct("PoolEvent::AddressChange")
.field("conn_info", connection.info())
.field("new_endpoint", new_endpoint)
.field("old_endpoint", old_endpoint)
.finish()
},
}
}
}
@ -639,7 +656,27 @@ where
}),
_ => unreachable!("since `entry` is an `EstablishedEntry`.")
}
}
},
manager::Event::AddressChange { entry, new_endpoint, old_endpoint } => {
let id = entry.id();
match self.established.get_mut(entry.connected().peer_id()) {
Some(list) => *list.get_mut(&id)
.expect("state inconsistency: entry is `EstablishedEntry` but absent \
from `established`") = new_endpoint.clone(),
None => unreachable!("since `entry` is an `EstablishedEntry`.")
};
match self.get(id) {
Some(PoolConnection::Established(connection)) =>
return Poll::Ready(PoolEvent::AddressChange {
connection,
new_endpoint,
old_endpoint,
}),
_ => unreachable!("since `entry` is an `EstablishedEntry`.")
}
},
}
}
}

20
core/src/connection/substream.rs
View File

@ -18,8 +18,9 @@
// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
// DEALINGS IN THE SOFTWARE.
use crate::muxing::{StreamMuxer, SubstreamRef, substream_from_ref};
use crate::muxing::{StreamMuxer, StreamMuxerEvent, SubstreamRef, substream_from_ref};
use futures::prelude::*;
use multiaddr::Multiaddr;
use smallvec::SmallVec;
use std::sync::Arc;
use std::{fmt, io::Error as IoError, pin::Pin, task::Context, task::Poll};
@ -95,6 +96,12 @@ where
/// destroyed or `close_graceful` is called.
substream: Substream<TMuxer>,
},
/// Address to the remote has changed. The previous one is now obsolete.
///
/// > **Note**: This can for example happen when using the QUIC protocol, where the two nodes
/// > can change their IP address while retaining the same QUIC connection.
AddressChange(Multiaddr),
}
/// Identifier for a substream being opened.
@ -145,13 +152,15 @@ where
/// Provides an API similar to `Future`.
pub fn poll(&mut self, cx: &mut Context) -> Poll<Result<SubstreamEvent<TMuxer, TUserData>, IoError>> {
// Polling inbound substream.
match self.inner.poll_inbound(cx) {
Poll::Ready(Ok(substream)) => {
match self.inner.poll_event(cx) {
Poll::Ready(Ok(StreamMuxerEvent::InboundSubstream(substream))) => {
let substream = substream_from_ref(self.inner.clone(), substream);
return Poll::Ready(Ok(SubstreamEvent::InboundSubstream {
substream,
}));
}
Poll::Ready(Ok(StreamMuxerEvent::AddressChange(addr))) =>
return Poll::Ready(Ok(SubstreamEvent::AddressChange(addr))),
Poll::Ready(Err(err)) => return Poll::Ready(Err(err.into())),
Poll::Pending => {}
}
@ -253,6 +262,11 @@ where
.field("substream", substream)
.finish()
},
SubstreamEvent::AddressChange(address) => {
f.debug_struct("SubstreamEvent::AddressChange")
.field("address", address)
.finish()
},
}
}
}

24
core/src/either.rs
View File

@ -19,7 +19,7 @@
// DEALINGS IN THE SOFTWARE.
use crate::{
muxing::StreamMuxer,
muxing::{StreamMuxer, StreamMuxerEvent},
ProtocolName,
transport::{Transport, ListenerEvent, TransportError},
Multiaddr
@ -189,10 +189,26 @@ where
type OutboundSubstream = EitherOutbound<A, B>;
type Error = IoError;
fn poll_inbound(&self, cx: &mut Context) -> Poll<Result<Self::Substream, Self::Error>> {
fn poll_event(&self, cx: &mut Context) -> Poll<Result<StreamMuxerEvent<Self::Substream>, Self::Error>> {
match self {
EitherOutput::First(inner) => inner.poll_inbound(cx).map(|p| p.map(EitherOutput::First)).map_err(|e| e.into()),
EitherOutput::Second(inner) => inner.poll_inbound(cx).map(|p| p.map(EitherOutput::Second)).map_err(|e| e.into()),
EitherOutput::First(inner) => inner.poll_event(cx).map(|result| {
result.map_err(|e| e.into()).map(|event| {
match event {
StreamMuxerEvent::AddressChange(addr) => StreamMuxerEvent::AddressChange(addr),
StreamMuxerEvent::InboundSubstream(substream) =>
StreamMuxerEvent::InboundSubstream(EitherOutput::First(substream))
}
})
}),
EitherOutput::Second(inner) => inner.poll_event(cx).map(|result| {
result.map_err(|e| e.into()).map(|event| {
match event {
StreamMuxerEvent::AddressChange(addr) => StreamMuxerEvent::AddressChange(addr),
StreamMuxerEvent::InboundSubstream(substream) =>
StreamMuxerEvent::InboundSubstream(EitherOutput::Second(substream))
}
})
}),
}
}

71
core/src/muxing.rs
View File

@ -53,6 +53,7 @@
use fnv::FnvHashMap;
use futures::{future, prelude::*, task::Context, task::Poll};
use multiaddr::Multiaddr;
use parking_lot::Mutex;
use std::{io, ops::Deref, fmt, pin::Pin, sync::atomic::{AtomicUsize, Ordering}};
@ -64,10 +65,11 @@ mod singleton;
///
/// The state of a muxer, as exposed by this API, is the following:
///
/// - A connection to the remote. The `flush_all` and `close` methods operate on this.
/// - A list of substreams that are open. The `poll_inbound`, `poll_outbound`, `read_substream`,
/// `write_substream`, `flush_substream`, `shutdown_substream` and `destroy_substream` methods
/// allow controlling these entries.
/// - A connection to the remote. The `poll_event`, `flush_all` and `close` methods operate
/// on this.
/// - A list of substreams that are open. The `poll_outbound`, `read_substream`, `write_substream`,
/// `flush_substream`, `shutdown_substream` and `destroy_substream` methods allow controlling
/// these entries.
/// - A list of outbound substreams being opened. The `open_outbound`, `poll_outbound` and
/// `destroy_outbound` methods allow controlling these entries.
///
@ -81,7 +83,7 @@ pub trait StreamMuxer {
/// Error type of the muxer
type Error: Into<io::Error>;
/// Polls for an inbound substream.
/// Polls for a connection-wide event.
///
/// This function behaves the same as a `Stream`.
///
@ -90,7 +92,7 @@ pub trait StreamMuxer {
/// Only the latest task that was used to call this method may be notified.
///
/// An error can be generated if the connection has been closed.
fn poll_inbound(&self, cx: &mut Context) -> Poll<Result<Self::Substream, Self::Error>>;
fn poll_event(&self, cx: &mut Context) -> Poll<Result<StreamMuxerEvent<Self::Substream>, Self::Error>>;
/// Opens a new outgoing substream, and produces the equivalent to a future that will be
/// resolved when it becomes available.
@ -206,18 +208,49 @@ pub trait StreamMuxer {
fn flush_all(&self, cx: &mut Context) -> Poll<Result<(), Self::Error>>;
}
/// Polls for an inbound from the muxer but wraps the output in an object that
/// implements `Read`/`Write`/`AsyncRead`/`AsyncWrite`.
pub fn inbound_from_ref_and_wrap<P>(
/// Event about a connection, reported by an implementation of [`StreamMuxer`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum StreamMuxerEvent<T> {
/// Remote has opened a new substream. Contains the substream in question.
InboundSubstream(T),
/// Address to the remote has changed. The previous one is now obsolete.
///
/// > **Note**: This can for example happen when using the QUIC protocol, where the two nodes
/// > can change their IP address while retaining the same QUIC connection.
AddressChange(Multiaddr),
}
impl<T> StreamMuxerEvent<T> {
/// If `self` is a [`StreamMuxerEvent::InboundSubstream`], returns the content. Otherwise
/// returns `None`.
pub fn into_inbound_substream(self) -> Option<T> {
if let StreamMuxerEvent::InboundSubstream(s) = self {
Some(s)
} else {
None
}
}
}
/// Polls for an event from the muxer and, if an inbound substream, wraps this substream in an
/// object that implements `Read`/`Write`/`AsyncRead`/`AsyncWrite`.
pub fn event_from_ref_and_wrap<P>(
muxer: P,
) -> impl Future<Output = Result<SubstreamRef<P>, <P::Target as StreamMuxer>::Error>>
) -> impl Future<Output = Result<StreamMuxerEvent<SubstreamRef<P>>, <P::Target as StreamMuxer>::Error>>
where
P: Deref + Clone,
P::Target: StreamMuxer,
{
let muxer2 = muxer.clone();
future::poll_fn(move |cx| muxer.poll_inbound(cx))
.map_ok(|substream| substream_from_ref(muxer2, substream))
future::poll_fn(move |cx| muxer.poll_event(cx))
.map_ok(|event| {
match event {
StreamMuxerEvent::InboundSubstream(substream) =>
StreamMuxerEvent::InboundSubstream(substream_from_ref(muxer2, substream)),
StreamMuxerEvent::AddressChange(addr) => StreamMuxerEvent::AddressChange(addr),
}
})
}
/// Same as `outbound_from_ref`, but wraps the output in an object that
@ -478,8 +511,8 @@ impl StreamMuxer for StreamMuxerBox {
type Error = io::Error;
#[inline]
fn poll_inbound(&self, cx: &mut Context) -> Poll<Result<Self::Substream, Self::Error>> {
self.inner.poll_inbound(cx)
fn poll_event(&self, cx: &mut Context) -> Poll<Result<StreamMuxerEvent<Self::Substream>, Self::Error>> {
self.inner.poll_event(cx)
}
#[inline]
@ -550,16 +583,18 @@ where
type Error = io::Error;
#[inline]
fn poll_inbound(&self, cx: &mut Context) -> Poll<Result<Self::Substream, Self::Error>> {
let substream = match self.inner.poll_inbound(cx) {
fn poll_event(&self, cx: &mut Context) -> Poll<Result<StreamMuxerEvent<Self::Substream>, Self::Error>> {
let substream = match self.inner.poll_event(cx) {
Poll::Pending => return Poll::Pending,
Poll::Ready(Ok(s)) => s,
Poll::Ready(Ok(StreamMuxerEvent::AddressChange(a))) =>
return Poll::Ready(Ok(StreamMuxerEvent::AddressChange(a))),
Poll::Ready(Ok(StreamMuxerEvent::InboundSubstream(s))) => s,
Poll::Ready(Err(err)) => return Poll::Ready(Err(err.into())),
};
let id = self.next_substream.fetch_add(1, Ordering::Relaxed);
self.substreams.lock().insert(id, substream);
Poll::Ready(Ok(id))
Poll::Ready(Ok(StreamMuxerEvent::InboundSubstream(id)))
}
#[inline]

7
core/src/muxing/singleton.rs
View File

@ -18,7 +18,8 @@
// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
// DEALINGS IN THE SOFTWARE.
use crate::{connection::Endpoint, muxing::StreamMuxer};
use crate::{connection::Endpoint, muxing::{StreamMuxer, StreamMuxerEvent}};
use futures::prelude::*;
use parking_lot::Mutex;
use std::{io, pin::Pin, sync::atomic::{AtomicBool, Ordering}, task::Context, task::Poll};
@ -64,14 +65,14 @@ where
type OutboundSubstream = OutboundSubstream;
type Error = io::Error;
fn poll_inbound(&self, _: &mut Context) -> Poll<Result<Self::Substream, io::Error>> {
fn poll_event(&self, _: &mut Context) -> Poll<Result<StreamMuxerEvent<Self::Substream>, io::Error>> {
match self.endpoint {
Endpoint::Dialer => return Poll::Pending,
Endpoint::Listener => {}
}
if !self.substream_extracted.swap(true, Ordering::Relaxed) {
Poll::Ready(Ok(Substream {}))
Poll::Ready(Ok(StreamMuxerEvent::InboundSubstream(Substream {})))
} else {
Poll::Pending
}

9
core/src/network.rs
View File

@ -416,7 +416,14 @@ where
Poll::Ready(PoolEvent::ConnectionEvent { connection, event }) => {
NetworkEvent::ConnectionEvent {
connection,
event
event,
}
}
Poll::Ready(PoolEvent::AddressChange { connection, new_endpoint, old_endpoint }) => {
NetworkEvent::AddressChange {
connection,
new_endpoint,
old_endpoint,
}
}
};

17
core/src/network/event.rs
View File

@ -155,6 +155,16 @@ where
/// Event that was produced by the node.
event: TOutEvent,
},
/// An established connection has changed its address.
AddressChange {
/// The connection whose address has changed.
connection: EstablishedConnection<'a, TInEvent, TConnInfo, TPeerId>,
/// New endpoint of this connection.
new_endpoint: ConnectedPoint,
/// Old endpoint of this connection.
old_endpoint: ConnectedPoint,
},
}
impl<TTrans, TInEvent, TOutEvent, THandler, TConnInfo, TPeerId> fmt::Debug for
@ -240,6 +250,13 @@ where
.field("event", event)
.finish()
}
NetworkEvent::AddressChange { connection, new_endpoint, old_endpoint } => {
f.debug_struct("AddressChange")
.field("connection", connection)
.field("new_endpoint", new_endpoint)
.field("old_endpoint", old_endpoint)
.finish()
}
}
}
}