libp2p_core/

transport.rs

// Copyright 2017-2018 Parity Technologies (UK) Ltd.
//
// Permission is hereby granted, free of charge, to any person obtaining a
// copy of this software and associated documentation files (the "Software"),
// to deal in the Software without restriction, including without limitation
// the rights to use, copy, modify, merge, publish, distribute, sublicense,
// and/or sell copies of the Software, and to permit persons to whom the
// Software is furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
// DEALINGS IN THE SOFTWARE.

//! Connection-oriented communication channels.
//!
//! The main entity of this module is the [`Transport`] trait, which provides an
//! interface for establishing connections with other nodes, thereby negotiating
//! any desired protocols. The rest of the module defines combinators for
//! modifying a transport through composition with other transports or protocol upgrades.

use std::{
    error::Error,
    fmt,
    pin::Pin,
    sync::atomic::{AtomicUsize, Ordering},
    task::{Context, Poll},
};

use futures::prelude::*;
use multiaddr::Multiaddr;

pub mod and_then;
pub mod choice;
pub mod dummy;
pub mod global_only;
pub mod map;
pub mod map_err;
pub mod memory;
pub mod timeout;
pub mod upgrade;

mod boxed;
mod optional;

pub use self::{
    boxed::Boxed, choice::OrTransport, memory::MemoryTransport, optional::OptionalTransport,
    upgrade::Upgrade,
};
use crate::{ConnectedPoint, Endpoint};

static NEXT_LISTENER_ID: AtomicUsize = AtomicUsize::new(1);

/// The port use policy for a new connection.
#[derive(Debug, Copy, Clone, Default, PartialEq, Eq, Hash)]
pub enum PortUse {
    /// Always allocate a new port for the dial.
    New,
    /// Best effor reusing of an existing port.
    ///
    /// If there is no listener present that can be used to dial, a new port is allocated.
    #[default]
    Reuse,
}

/// Options to customize the behaviour during dialing.
#[derive(Debug, Copy, Clone)]
pub struct DialOpts {
    /// The endpoint establishing a new connection.
    ///
    /// When attempting a hole-punch, both parties simultaneously "dial" each other but one party
    /// has to be the "listener" on the final connection. This option specifies the role of
    /// this node in the final connection.
    pub role: Endpoint,
    /// The port use policy for a new connection.
    pub port_use: PortUse,
}

/// A transport provides connection-oriented communication between two peers
/// through ordered streams of data (i.e. connections).
///
/// Connections are established either by [listening](Transport::listen_on)
/// or [dialing](Transport::dial) on a [`Transport`]. A peer that
/// obtains a connection by listening is often referred to as the *listener* and the
/// peer that initiated the connection through dialing as the *dialer*, in
/// contrast to the traditional roles of *server* and *client*.
///
/// Most transports also provide a form of reliable delivery on the established
/// connections but the precise semantics of these guarantees depend on the
/// specific transport.
///
/// This trait is implemented for concrete connection-oriented transport protocols
/// like TCP or Unix Domain Sockets, but also on wrappers that add additional
/// functionality to the dialing or listening process (e.g. name resolution via
/// the DNS).
///
/// Additional protocols can be layered on top of the connections established
/// by a [`Transport`] through an upgrade mechanism that is initiated via
/// [`upgrade`](Transport::upgrade).
///
/// Note for implementers: Futures returned by [`Transport::dial`] should only
/// do work once polled for the first time. E.g. in the case of TCP, connecting
/// to the remote should not happen immediately on [`Transport::dial`] but only
/// once the returned [`Future`] is polled. The caller of [`Transport::dial`]
/// may call the method multiple times with a set of addresses, racing a subset
/// of the returned dials to success concurrently.
pub trait Transport {
    /// The result of a connection setup process, including protocol upgrades.
    ///
    /// Typically the output contains at least a handle to a data stream (i.e. a
    /// connection or a substream multiplexer on top of a connection) that
    /// provides APIs for sending and receiving data through the connection.
    type Output;

    /// An error that occurred during connection setup.
    type Error: Error;

    /// A pending [`Output`](Transport::Output) for an inbound connection,
    /// obtained from the [`Transport`] stream.
    ///
    /// After a connection has been accepted by the transport, it may need to go through
    /// asynchronous post-processing (i.e. protocol upgrade negotiations). Such
    /// post-processing should not block the `Listener` from producing the next
    /// connection, hence further connection setup proceeds asynchronously.
    /// Once a `ListenerUpgrade` future resolves it yields the [`Output`](Transport::Output)
    /// of the connection setup process.
    type ListenerUpgrade: Future<Output = Result<Self::Output, Self::Error>>;

    /// A pending [`Output`](Transport::Output) for an outbound connection,
    /// obtained from [dialing](Transport::dial).
    type Dial: Future<Output = Result<Self::Output, Self::Error>>;

    /// Listens on the given [`Multiaddr`] for inbound connections with a provided [`ListenerId`].
    fn listen_on(
        &mut self,
        id: ListenerId,
        addr: Multiaddr,
    ) -> Result<(), TransportError<Self::Error>>;

    /// Remove a listener.
    ///
    /// Return `true` if there was a listener with this Id, `false`
    /// otherwise.
    fn remove_listener(&mut self, id: ListenerId) -> bool;

    /// Dials the given [`Multiaddr`], returning a future for a pending outbound connection.
    ///
    /// If [`TransportError::MultiaddrNotSupported`] is returned, it may be desirable to
    /// try an alternative [`Transport`], if available.
    fn dial(
        &mut self,
        addr: Multiaddr,
        opts: DialOpts,
    ) -> Result<Self::Dial, TransportError<Self::Error>>;

    /// Poll for [`TransportEvent`]s.
    ///
    /// A [`TransportEvent::Incoming`] should be produced whenever a connection is received at the
    /// lowest level of the transport stack. The item must be a
    /// [`ListenerUpgrade`](Transport::ListenerUpgrade) future that resolves to an
    /// [`Output`](Transport::Output) value once all protocol upgrades have been applied.
    ///
    /// Transports are expected to produce [`TransportEvent::Incoming`] events only for
    /// listen addresses which have previously been announced via
    /// a [`TransportEvent::NewAddress`] event and which have not been invalidated by
    /// an [`TransportEvent::AddressExpired`] event yet.
    fn poll(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<TransportEvent<Self::ListenerUpgrade, Self::Error>>;

    /// Boxes the transport, including custom transport errors.
    fn boxed(self) -> boxed::Boxed<Self::Output>
    where
        Self: Sized + Send + Unpin + 'static,
        Self::Dial: Send + 'static,
        Self::ListenerUpgrade: Send + 'static,
        Self::Error: Send + Sync,
    {
        boxed::boxed(self)
    }

    /// Applies a function on the connections created by the transport.
    fn map<F, O>(self, f: F) -> map::Map<Self, F>
    where
        Self: Sized,
        F: FnOnce(Self::Output, ConnectedPoint) -> O,
    {
        map::Map::new(self, f)
    }

    /// Applies a function on the errors generated by the futures of the transport.
    fn map_err<F, E>(self, f: F) -> map_err::MapErr<Self, F>
    where
        Self: Sized,
        F: FnOnce(Self::Error) -> E,
    {
        map_err::MapErr::new(self, f)
    }

    /// Adds a fallback transport that is used when encountering errors
    /// while establishing inbound or outbound connections.
    ///
    /// The returned transport will act like `self`, except that if `listen_on` or `dial`
    /// return an error then `other` will be tried.
    fn or_transport<U>(self, other: U) -> OrTransport<Self, U>
    where
        Self: Sized,
        U: Transport,
        <U as Transport>::Error: 'static,
    {
        OrTransport::new(self, other)
    }

    /// Applies a function producing an asynchronous result to every connection
    /// created by this transport.
    ///
    /// This function can be used for ad-hoc protocol upgrades or
    /// for processing or adapting the output for following configurations.
    ///
    /// For the high-level transport upgrade procedure, see [`Transport::upgrade`].
    fn and_then<C, F, O>(self, f: C) -> and_then::AndThen<Self, C>
    where
        Self: Sized,
        C: FnOnce(Self::Output, ConnectedPoint) -> F,
        F: TryFuture<Ok = O>,
        <F as TryFuture>::Error: Error + 'static,
    {
        and_then::AndThen::new(self, f)
    }

    /// Begins a series of protocol upgrades via an [`upgrade::Builder`].
    fn upgrade(self, version: upgrade::Version) -> upgrade::Builder<Self>
    where
        Self: Sized,
        Self::Error: 'static,
    {
        upgrade::Builder::new(self, version)
    }
}

/// The ID of a single listener.
#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash)]
pub struct ListenerId(usize);

impl ListenerId {
    /// Creates a new `ListenerId`.
    pub fn next() -> Self {
        ListenerId(NEXT_LISTENER_ID.fetch_add(1, Ordering::SeqCst))
    }
}

impl fmt::Display for ListenerId {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "{}", self.0)
    }
}

/// Event produced by [`Transport`]s.
pub enum TransportEvent<TUpgr, TErr> {
    /// A new address is being listened on.
    NewAddress {
        /// The listener that is listening on the new address.
        listener_id: ListenerId,
        /// The new address that is being listened on.
        listen_addr: Multiaddr,
    },
    /// An address is no longer being listened on.
    AddressExpired {
        /// The listener that is no longer listening on the address.
        listener_id: ListenerId,
        /// The new address that is being listened on.
        listen_addr: Multiaddr,
    },
    /// A connection is incoming on one of the listeners.
    Incoming {
        /// The listener that produced the upgrade.
        listener_id: ListenerId,
        /// The produced upgrade.
        upgrade: TUpgr,
        /// Local connection address.
        local_addr: Multiaddr,
        /// Address used to send back data to the incoming client.
        send_back_addr: Multiaddr,
    },
    /// A listener closed.
    ListenerClosed {
        /// The ID of the listener that closed.
        listener_id: ListenerId,
        /// Reason for the closure. Contains `Ok(())` if the stream produced `None`, or `Err`
        /// if the stream produced an error.
        reason: Result<(), TErr>,
    },
    /// A listener errored.
    ///
    /// The listener will continue to be polled for new events and the event
    /// is for informational purposes only.
    ListenerError {
        /// The ID of the listener that errored.
        listener_id: ListenerId,
        /// The error value.
        error: TErr,
    },
}

impl<TUpgr, TErr> TransportEvent<TUpgr, TErr> {
    /// In case this [`TransportEvent`] is an upgrade, apply the given function
    /// to the upgrade and produce another transport event based the function's result.
    pub fn map_upgrade<U>(self, map: impl FnOnce(TUpgr) -> U) -> TransportEvent<U, TErr> {
        match self {
            TransportEvent::Incoming {
                listener_id,
                upgrade,
                local_addr,
                send_back_addr,
            } => TransportEvent::Incoming {
                listener_id,
                upgrade: map(upgrade),
                local_addr,
                send_back_addr,
            },
            TransportEvent::NewAddress {
                listen_addr,
                listener_id,
            } => TransportEvent::NewAddress {
                listen_addr,
                listener_id,
            },
            TransportEvent::AddressExpired {
                listen_addr,
                listener_id,
            } => TransportEvent::AddressExpired {
                listen_addr,
                listener_id,
            },
            TransportEvent::ListenerError { listener_id, error } => {
                TransportEvent::ListenerError { listener_id, error }
            }
            TransportEvent::ListenerClosed {
                listener_id,
                reason,
            } => TransportEvent::ListenerClosed {
                listener_id,
                reason,
            },
        }
    }

    /// In case this [`TransportEvent`] is an [`ListenerError`](TransportEvent::ListenerError),
    /// or [`ListenerClosed`](TransportEvent::ListenerClosed) apply the given function to the
    /// error and produce another transport event based on the function's result.
    pub fn map_err<E>(self, map_err: impl FnOnce(TErr) -> E) -> TransportEvent<TUpgr, E> {
        match self {
            TransportEvent::Incoming {
                listener_id,
                upgrade,
                local_addr,
                send_back_addr,
            } => TransportEvent::Incoming {
                listener_id,
                upgrade,
                local_addr,
                send_back_addr,
            },
            TransportEvent::NewAddress {
                listen_addr,
                listener_id,
            } => TransportEvent::NewAddress {
                listen_addr,
                listener_id,
            },
            TransportEvent::AddressExpired {
                listen_addr,
                listener_id,
            } => TransportEvent::AddressExpired {
                listen_addr,
                listener_id,
            },
            TransportEvent::ListenerError { listener_id, error } => TransportEvent::ListenerError {
                listener_id,
                error: map_err(error),
            },
            TransportEvent::ListenerClosed {
                listener_id,
                reason,
            } => TransportEvent::ListenerClosed {
                listener_id,
                reason: reason.map_err(map_err),
            },
        }
    }

    /// Returns `true` if this is an [`Incoming`](TransportEvent::Incoming) transport event.
    pub fn is_upgrade(&self) -> bool {
        matches!(self, TransportEvent::Incoming { .. })
    }

    /// Try to turn this transport event into the upgrade parts of the
    /// incoming connection.
    ///
    /// Returns `None` if the event is not actually an incoming connection,
    /// otherwise the upgrade and the remote address.
    pub fn into_incoming(self) -> Option<(TUpgr, Multiaddr)> {
        let TransportEvent::Incoming {
            upgrade,
            send_back_addr,
            ..
        } = self
        else {
            return None;
        };

        Some((upgrade, send_back_addr))
    }

    /// Returns `true` if this is a [`TransportEvent::NewAddress`].
    pub fn is_new_address(&self) -> bool {
        matches!(self, TransportEvent::NewAddress { .. })
    }

    /// Try to turn this transport event into the new `Multiaddr`.
    ///
    /// Returns `None` if the event is not actually a [`TransportEvent::NewAddress`],
    /// otherwise the address.
    pub fn into_new_address(self) -> Option<Multiaddr> {
        if let TransportEvent::NewAddress { listen_addr, .. } = self {
            Some(listen_addr)
        } else {
            None
        }
    }

    /// Returns `true` if this is an [`TransportEvent::AddressExpired`].
    pub fn is_address_expired(&self) -> bool {
        matches!(self, TransportEvent::AddressExpired { .. })
    }

    /// Try to turn this transport event into the expire `Multiaddr`.
    ///
    /// Returns `None` if the event is not actually a [`TransportEvent::AddressExpired`],
    /// otherwise the address.
    pub fn into_address_expired(self) -> Option<Multiaddr> {
        if let TransportEvent::AddressExpired { listen_addr, .. } = self {
            Some(listen_addr)
        } else {
            None
        }
    }

    /// Returns `true` if this is an [`TransportEvent::ListenerError`] transport event.
    pub fn is_listener_error(&self) -> bool {
        matches!(self, TransportEvent::ListenerError { .. })
    }

    /// Try to turn this transport event into the listener error.
    ///
    /// Returns `None` if the event is not actually a [`TransportEvent::ListenerError`]`,
    /// otherwise the error.
    pub fn into_listener_error(self) -> Option<TErr> {
        if let TransportEvent::ListenerError { error, .. } = self {
            Some(error)
        } else {
            None
        }
    }
}

impl<TUpgr, TErr: fmt::Debug> fmt::Debug for TransportEvent<TUpgr, TErr> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> Result<(), fmt::Error> {
        match self {
            TransportEvent::NewAddress {
                listener_id,
                listen_addr,
            } => f
                .debug_struct("TransportEvent::NewAddress")
                .field("listener_id", listener_id)
                .field("listen_addr", listen_addr)
                .finish(),
            TransportEvent::AddressExpired {
                listener_id,
                listen_addr,
            } => f
                .debug_struct("TransportEvent::AddressExpired")
                .field("listener_id", listener_id)
                .field("listen_addr", listen_addr)
                .finish(),
            TransportEvent::Incoming {
                listener_id,
                local_addr,
                ..
            } => f
                .debug_struct("TransportEvent::Incoming")
                .field("listener_id", listener_id)
                .field("local_addr", local_addr)
                .finish(),
            TransportEvent::ListenerClosed {
                listener_id,
                reason,
            } => f
                .debug_struct("TransportEvent::Closed")
                .field("listener_id", listener_id)
                .field("reason", reason)
                .finish(),
            TransportEvent::ListenerError { listener_id, error } => f
                .debug_struct("TransportEvent::ListenerError")
                .field("listener_id", listener_id)
                .field("error", error)
                .finish(),
        }
    }
}

/// An error during [dialing][Transport::dial] or [listening][Transport::listen_on]
/// on a [`Transport`].
#[derive(Debug, Clone)]
pub enum TransportError<TErr> {
    /// The [`Multiaddr`] passed as parameter is not supported.
    ///
    /// Contains back the same address.
    MultiaddrNotSupported(Multiaddr),

    /// Any other error that a [`Transport`] may produce.
    Other(TErr),
}

impl<TErr> TransportError<TErr> {
    /// Applies a function to the error in [`TransportError::Other`].
    pub fn map<TNewErr>(self, map: impl FnOnce(TErr) -> TNewErr) -> TransportError<TNewErr> {
        match self {
            TransportError::MultiaddrNotSupported(addr) => {
                TransportError::MultiaddrNotSupported(addr)
            }
            TransportError::Other(err) => TransportError::Other(map(err)),
        }
    }
}

impl<TErr> fmt::Display for TransportError<TErr>
where
    TErr: fmt::Display,
{
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            TransportError::MultiaddrNotSupported(addr) => {
                write!(f, "Multiaddr is not supported: {addr}")
            }
            TransportError::Other(_) => Ok(()),
        }
    }
}

impl<TErr> Error for TransportError<TErr>
where
    TErr: Error + 'static,
{
    fn source(&self) -> Option<&(dyn Error + 'static)> {
        match self {
            TransportError::MultiaddrNotSupported(_) => None,
            TransportError::Other(err) => Some(err),
        }
    }
}