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

feat(swarm): deprecate NegotiatedSubstream in favor of Stream

Browse Source 
This patch tackles two things at once that are fairly intertwined:

1. There is no such thing as a "substream" in libp2p, the spec and other implementations only talk about "streams". We fix this by deprecating `NegotiatedSubstream`.
2. Previously, `NegotiatedSubstream` was a type alias that pointed to a type from `multistream-select`, effectively leaking the version of `multistream-select` to all dependencies of `libp2p-swarm`. We fix this by introducing a `Stream` newtype.

Resolves: #3759.
Related: #3748.

Pull-Request: #3912.
This commit is contained in:
Thomas Eizinger
2023-05-12 08:19:23 +02:00
committed by GitHub
parent 234a0d24db
commit 9e625881d5
24 changed files with 234 additions and 169 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
Cargo.lock generated
View File

@ -2989,6 +2989,7 @@ dependencies = [
"libp2p-swarm-test",
"libp2p-yamux",
"log",
"multistream-select",
"once_cell",
"quickcheck-ext",
"rand 0.8.5",

2
libp2p/src/lib.rs
View File

@ -177,7 +177,7 @@ pub use self::swarm::Swarm;
pub use self::transport_ext::TransportExt;
pub use libp2p_identity as identity;
pub use libp2p_identity::PeerId;
pub use libp2p_swarm::StreamProtocol;
pub use libp2p_swarm::{Stream, StreamProtocol};
/// Builds a `Transport` based on TCP/IP that supports the most commonly-used features of libp2p:
///

8
protocols/dcutr/src/protocol/inbound.rs
View File

@ -22,7 +22,7 @@ use crate::proto;
use asynchronous_codec::Framed;
use futures::{future::BoxFuture, prelude::*};
use libp2p_core::{multiaddr::Protocol, upgrade, Multiaddr};
use libp2p_swarm::{NegotiatedSubstream, StreamProtocol};
use libp2p_swarm::{Stream, StreamProtocol};
use std::convert::TryFrom;
use std::iter;
use thiserror::Error;
@ -38,12 +38,12 @@ impl upgrade::UpgradeInfo for Upgrade {
}
}
impl upgrade::InboundUpgrade<NegotiatedSubstream> for Upgrade {
impl upgrade::InboundUpgrade<Stream> for Upgrade {
type Output = PendingConnect;
type Error = UpgradeError;
type Future = BoxFuture<'static, Result<Self::Output, Self::Error>>;
fn upgrade_inbound(self, substream: NegotiatedSubstream, _: Self::Info) -> Self::Future {
fn upgrade_inbound(self, substream: Stream, _: Self::Info) -> Self::Future {
let mut substream = Framed::new(
substream,
quick_protobuf_codec::Codec::new(super::MAX_MESSAGE_SIZE_BYTES),
@ -92,7 +92,7 @@ impl upgrade::InboundUpgrade<NegotiatedSubstream> for Upgrade {
}
pub struct PendingConnect {
substream: Framed<NegotiatedSubstream, quick_protobuf_codec::Codec<proto::HolePunch>>,
substream: Framed<Stream, quick_protobuf_codec::Codec<proto::HolePunch>>,
remote_obs_addrs: Vec<Multiaddr>,
}

6
protocols/dcutr/src/protocol/outbound.rs
View File

@ -24,7 +24,7 @@ use futures::{future::BoxFuture, prelude::*};
use futures_timer::Delay;
use instant::Instant;
use libp2p_core::{multiaddr::Protocol, upgrade, Multiaddr};
use libp2p_swarm::{NegotiatedSubstream, StreamProtocol};
use libp2p_swarm::{Stream, StreamProtocol};
use std::convert::TryFrom;
use std::iter;
use thiserror::Error;
@ -48,12 +48,12 @@ impl Upgrade {
}
}
impl upgrade::OutboundUpgrade<NegotiatedSubstream> for Upgrade {
impl upgrade::OutboundUpgrade<Stream> for Upgrade {
type Output = Connect;
type Error = UpgradeError;
type Future = BoxFuture<'static, Result<Self::Output, Self::Error>>;
fn upgrade_outbound(self, substream: NegotiatedSubstream, _: Self::Info) -> Self::Future {
fn upgrade_outbound(self, substream: Stream, _: Self::Info) -> Self::Future {
let mut substream = Framed::new(
substream,
quick_protobuf_codec::Codec::new(super::MAX_MESSAGE_SIZE_BYTES),

14
protocols/gossipsub/src/handler.rs
View File

@ -33,7 +33,7 @@ use libp2p_swarm::handler::{
FullyNegotiatedInbound, FullyNegotiatedOutbound, KeepAlive, StreamUpgradeError,
SubstreamProtocol,
};
use libp2p_swarm::NegotiatedSubstream;
use libp2p_swarm::Stream;
use smallvec::SmallVec;
use std::{
pin::Pin,
@ -143,9 +143,9 @@ pub enum DisabledHandler {
/// State of the inbound substream, opened either by us or by the remote.
enum InboundSubstreamState {
/// Waiting for a message from the remote. The idle state for an inbound substream.
WaitingInput(Framed<NegotiatedSubstream, GossipsubCodec>),
WaitingInput(Framed<Stream, GossipsubCodec>),
/// The substream is being closed.
Closing(Framed<NegotiatedSubstream, GossipsubCodec>),
Closing(Framed<Stream, GossipsubCodec>),
/// An error occurred during processing.
Poisoned,
}
@ -153,11 +153,11 @@ enum InboundSubstreamState {
/// State of the outbound substream, opened either by us or by the remote.
enum OutboundSubstreamState {
/// Waiting for the user to send a message. The idle state for an outbound substream.
WaitingOutput(Framed<NegotiatedSubstream, GossipsubCodec>),
WaitingOutput(Framed<Stream, GossipsubCodec>),
/// Waiting to send a message to the remote.
PendingSend(Framed<NegotiatedSubstream, GossipsubCodec>, proto::RPC),
PendingSend(Framed<Stream, GossipsubCodec>, proto::RPC),
/// Waiting to flush the substream so that the data arrives to the remote.
PendingFlush(Framed<NegotiatedSubstream, GossipsubCodec>),
PendingFlush(Framed<Stream, GossipsubCodec>),
/// An error occurred during processing.
Poisoned,
}
@ -185,7 +185,7 @@ impl Handler {
impl EnabledHandler {
fn on_fully_negotiated_inbound(
&mut self,
(substream, peer_kind): (Framed<NegotiatedSubstream, GossipsubCodec>, PeerKind),
(substream, peer_kind): (Framed<Stream, GossipsubCodec>, PeerKind),
) {
// update the known kind of peer
if self.peer_kind.is_none() {

36
protocols/kad/src/handler.rs
View File

@ -33,7 +33,7 @@ use libp2p_swarm::handler::{
ConnectionEvent, DialUpgradeError, FullyNegotiatedInbound, FullyNegotiatedOutbound,
};
use libp2p_swarm::{
ConnectionHandler, ConnectionHandlerEvent, KeepAlive, NegotiatedSubstream, StreamUpgradeError,
ConnectionHandler, ConnectionHandlerEvent, KeepAlive, Stream, StreamUpgradeError,
SubstreamProtocol,
};
use log::trace;
@ -116,20 +116,16 @@ pub struct KademliaHandlerConfig {
/// State of an active outbound substream.
enum OutboundSubstreamState<TUserData> {
/// Waiting to send a message to the remote.
PendingSend(
KadOutStreamSink<NegotiatedSubstream>,
KadRequestMsg,
Option<TUserData>,
),
PendingSend(KadOutStreamSink<Stream>, KadRequestMsg, Option<TUserData>),
/// Waiting to flush the substream so that the data arrives to the remote.
PendingFlush(KadOutStreamSink<NegotiatedSubstream>, Option<TUserData>),
PendingFlush(KadOutStreamSink<Stream>, Option<TUserData>),
/// Waiting for an answer back from the remote.
// TODO: add timeout
WaitingAnswer(KadOutStreamSink<NegotiatedSubstream>, TUserData),
WaitingAnswer(KadOutStreamSink<Stream>, TUserData),
/// An error happened on the substream and we should report the error to the user.
ReportError(KademliaHandlerQueryErr, TUserData),
/// The substream is being closed.
Closing(KadOutStreamSink<NegotiatedSubstream>),
Closing(KadOutStreamSink<Stream>),
/// The substream is complete and will not perform any more work.
Done,
Poisoned,
@ -142,24 +138,16 @@ enum InboundSubstreamState<TUserData> {
/// Whether it is the first message to be awaited on this stream.
first: bool,
connection_id: UniqueConnecId,
substream: KadInStreamSink<NegotiatedSubstream>,
substream: KadInStreamSink<Stream>,
},
/// Waiting for the behaviour to send a [`KademliaHandlerIn`] event containing the response.
WaitingBehaviour(
UniqueConnecId,
KadInStreamSink<NegotiatedSubstream>,
Option<Waker>,
),
WaitingBehaviour(UniqueConnecId, KadInStreamSink<Stream>, Option<Waker>),
/// Waiting to send an answer back to the remote.
PendingSend(
UniqueConnecId,
KadInStreamSink<NegotiatedSubstream>,
KadResponseMsg,
),
PendingSend(UniqueConnecId, KadInStreamSink<Stream>, KadResponseMsg),
/// Waiting to flush an answer back to the remote.
PendingFlush(UniqueConnecId, KadInStreamSink<NegotiatedSubstream>),
PendingFlush(UniqueConnecId, KadInStreamSink<Stream>),
/// The substream is being closed.
Closing(KadInStreamSink<NegotiatedSubstream>),
Closing(KadInStreamSink<Stream>),
/// The substream was cancelled in favor of a new one.
Cancelled,
@ -813,7 +801,7 @@ impl Default for KademliaHandlerConfig {
}
}
impl<TUserData> Stream for OutboundSubstreamState<TUserData>
impl<TUserData> futures::Stream for OutboundSubstreamState<TUserData>
where
TUserData: Unpin,
{
@ -949,7 +937,7 @@ where
}
}
impl<TUserData> Stream for InboundSubstreamState<TUserData>
impl<TUserData> futures::Stream for InboundSubstreamState<TUserData>
where
TUserData: Unpin,
{

8
protocols/ping/src/handler.rs
View File

@ -27,7 +27,7 @@ use libp2p_swarm::handler::{
ConnectionEvent, DialUpgradeError, FullyNegotiatedInbound, FullyNegotiatedOutbound,
};
use libp2p_swarm::{
ConnectionHandler, ConnectionHandlerEvent, KeepAlive, NegotiatedSubstream, StreamProtocol,
ConnectionHandler, ConnectionHandlerEvent, KeepAlive, Stream, StreamProtocol,
StreamUpgradeError, SubstreamProtocol,
};
use std::collections::VecDeque;
@ -390,15 +390,15 @@ impl ConnectionHandler for Handler {
}
}
type PingFuture = BoxFuture<'static, Result<(NegotiatedSubstream, Duration), io::Error>>;
type PongFuture = BoxFuture<'static, Result<NegotiatedSubstream, io::Error>>;
type PingFuture = BoxFuture<'static, Result<(Stream, Duration), io::Error>>;
type PongFuture = BoxFuture<'static, Result<Stream, io::Error>>;
/// The current state w.r.t. outbound pings.
enum OutboundState {
/// A new substream is being negotiated for the ping protocol.
OpenStream,
/// The substream is idle, waiting to send the next ping.
Idle(NegotiatedSubstream),
Idle(Stream),
/// A ping is being sent and the response awaited.
Ping(PingFuture),
}

12
protocols/relay/src/behaviour/handler.rs
View File

@ -37,8 +37,8 @@ use libp2p_swarm::handler::{
ListenUpgradeError,
};
use libp2p_swarm::{
ConnectionHandler, ConnectionHandlerEvent, ConnectionId, KeepAlive, NegotiatedSubstream,
StreamUpgradeError, SubstreamProtocol,
ConnectionHandler, ConnectionHandlerEvent, ConnectionId, KeepAlive, Stream, StreamUpgradeError,
SubstreamProtocol,
};
use std::collections::VecDeque;
use std::fmt;
@ -77,7 +77,7 @@ pub enum In {
dst_peer_id: PeerId,
inbound_circuit_req: inbound_hop::CircuitReq,
dst_handler_notifier: oneshot::Sender<()>,
dst_stream: NegotiatedSubstream,
dst_stream: Stream,
dst_pending_data: Bytes,
},
}
@ -193,7 +193,7 @@ pub enum Event {
src_connection_id: ConnectionId,
inbound_circuit_req: inbound_hop::CircuitReq,
dst_handler_notifier: oneshot::Sender<()>,
dst_stream: NegotiatedSubstream,
dst_stream: Stream,
dst_pending_data: Bytes,
},
/// Negotiating an outbound substream for an inbound circuit request failed.
@ -914,10 +914,10 @@ pub struct OutboundOpenInfo {
pub(crate) struct CircuitParts {
circuit_id: CircuitId,
src_stream: NegotiatedSubstream,
src_stream: Stream,
src_pending_data: Bytes,
dst_peer_id: PeerId,
dst_handler_notifier: oneshot::Sender<()>,
dst_stream: NegotiatedSubstream,
dst_stream: Stream,
dst_pending_data: Bytes,
}

8
protocols/relay/src/priv_client.rs
View File

@ -39,8 +39,8 @@ use libp2p_identity::PeerId;
use libp2p_swarm::behaviour::{ConnectionClosed, ConnectionEstablished, FromSwarm};
use libp2p_swarm::dial_opts::DialOpts;
use libp2p_swarm::{
dummy, ConnectionDenied, ConnectionHandler, ConnectionId, DialFailure, NegotiatedSubstream,
NetworkBehaviour, NotifyHandler, PollParameters, StreamUpgradeError, THandler, THandlerInEvent,
dummy, ConnectionDenied, ConnectionHandler, ConnectionId, DialFailure, NetworkBehaviour,
NotifyHandler, PollParameters, Stream, StreamUpgradeError, THandler, THandlerInEvent,
THandlerOutEvent, ToSwarm,
};
use std::collections::{hash_map, HashMap, VecDeque};
@ -391,7 +391,7 @@ enum ConnectionState {
},
Operational {
read_buffer: Bytes,
substream: NegotiatedSubstream,
substream: Stream,
/// "Drop notifier" pattern to signal to the transport that the connection has been dropped.
///
/// This is flagged as "dead-code" by the compiler because we never read from it here.
@ -425,7 +425,7 @@ impl ConnectionState {
}
pub(crate) fn new_outbound(
substream: NegotiatedSubstream,
substream: Stream,
read_buffer: Bytes,
drop_notifier: oneshot::Sender<void::Void>,
) -> Self {

12
protocols/relay/src/protocol/inbound_hop.rs
View File

@ -26,7 +26,7 @@ use futures::{future::BoxFuture, prelude::*};
use instant::{Duration, SystemTime};
use libp2p_core::{upgrade, Multiaddr};
use libp2p_identity::PeerId;
use libp2p_swarm::{NegotiatedSubstream, StreamProtocol};
use libp2p_swarm::{Stream, StreamProtocol};
use std::convert::TryInto;
use std::iter;
use thiserror::Error;
@ -46,12 +46,12 @@ impl upgrade::UpgradeInfo for Upgrade {
}
}
impl upgrade::InboundUpgrade<NegotiatedSubstream> for Upgrade {
impl upgrade::InboundUpgrade<Stream> for Upgrade {
type Output = Req;
type Error = UpgradeError;
type Future = BoxFuture<'static, Result<Self::Output, Self::Error>>;
fn upgrade_inbound(self, substream: NegotiatedSubstream, _: Self::Info) -> Self::Future {
fn upgrade_inbound(self, substream: Stream, _: Self::Info) -> Self::Future {
let mut substream = Framed::new(
substream,
quick_protobuf_codec::Codec::new(MAX_MESSAGE_SIZE),
@ -126,7 +126,7 @@ pub enum Req {
}
pub struct ReservationReq {
substream: Framed<NegotiatedSubstream, quick_protobuf_codec::Codec<proto::HopMessage>>,
substream: Framed<Stream, quick_protobuf_codec::Codec<proto::HopMessage>>,
reservation_duration: Duration,
max_circuit_duration: Duration,
max_circuit_bytes: u64,
@ -183,7 +183,7 @@ impl ReservationReq {
pub struct CircuitReq {
dst: PeerId,
substream: Framed<NegotiatedSubstream, quick_protobuf_codec::Codec<proto::HopMessage>>,
substream: Framed<Stream, quick_protobuf_codec::Codec<proto::HopMessage>>,
}
impl CircuitReq {
@ -191,7 +191,7 @@ impl CircuitReq {
self.dst
}
pub async fn accept(mut self) -> Result<(NegotiatedSubstream, Bytes), UpgradeError> {
pub async fn accept(mut self) -> Result<(Stream, Bytes), UpgradeError> {
let msg = proto::HopMessage {
type_pb: proto::HopMessageType::STATUS,
peer: None,

10
protocols/relay/src/protocol/inbound_stop.rs
View File

@ -25,7 +25,7 @@ use bytes::Bytes;
use futures::{future::BoxFuture, prelude::*};
use libp2p_core::upgrade;
use libp2p_identity::PeerId;
use libp2p_swarm::{NegotiatedSubstream, StreamProtocol};
use libp2p_swarm::{Stream, StreamProtocol};
use std::iter;
use thiserror::Error;
@ -40,12 +40,12 @@ impl upgrade::UpgradeInfo for Upgrade {
}
}
impl upgrade::InboundUpgrade<NegotiatedSubstream> for Upgrade {
impl upgrade::InboundUpgrade<Stream> for Upgrade {
type Output = Circuit;
type Error = UpgradeError;
type Future = BoxFuture<'static, Result<Self::Output, Self::Error>>;
fn upgrade_inbound(self, substream: NegotiatedSubstream, _: Self::Info) -> Self::Future {
fn upgrade_inbound(self, substream: Stream, _: Self::Info) -> Self::Future {
let mut substream = Framed::new(
substream,
quick_protobuf_codec::Codec::new(MAX_MESSAGE_SIZE),
@ -111,7 +111,7 @@ pub enum FatalUpgradeError {
}
pub struct Circuit {
substream: Framed<NegotiatedSubstream, quick_protobuf_codec::Codec<proto::StopMessage>>,
substream: Framed<Stream, quick_protobuf_codec::Codec<proto::StopMessage>>,
src_peer_id: PeerId,
limit: Option<protocol::Limit>,
}
@ -125,7 +125,7 @@ impl Circuit {
self.limit
}
pub async fn accept(mut self) -> Result<(NegotiatedSubstream, Bytes), UpgradeError> {
pub async fn accept(mut self) -> Result<(Stream, Bytes), UpgradeError> {
let msg = proto::StopMessage {
type_pb: proto::StopMessageType::STATUS,
peer: None,

8
protocols/relay/src/protocol/outbound_hop.rs
View File

@ -27,7 +27,7 @@ use futures_timer::Delay;
use instant::{Duration, SystemTime};
use libp2p_core::{upgrade, Multiaddr};
use libp2p_identity::PeerId;
use libp2p_swarm::{NegotiatedSubstream, StreamProtocol};
use libp2p_swarm::{Stream, StreamProtocol};
use std::convert::TryFrom;
use std::iter;
use thiserror::Error;
@ -46,12 +46,12 @@ impl upgrade::UpgradeInfo for Upgrade {
}
}
impl upgrade::OutboundUpgrade<NegotiatedSubstream> for Upgrade {
impl upgrade::OutboundUpgrade<Stream> for Upgrade {
type Output = Output;
type Error = UpgradeError;
type Future = BoxFuture<'static, Result<Self::Output, Self::Error>>;
fn upgrade_outbound(self, substream: NegotiatedSubstream, _: Self::Info) -> Self::Future {
fn upgrade_outbound(self, substream: Stream, _: Self::Info) -> Self::Future {
let msg = match self {
Upgrade::Reserve => proto::HopMessage {
type_pb: proto::HopMessageType::RESERVE,
@ -269,7 +269,7 @@ pub enum Output {
limit: Option<Limit>,
},
Circuit {
substream: NegotiatedSubstream,
substream: Stream,
read_buffer: Bytes,
limit: Option<Limit>,
},

8
protocols/relay/src/protocol/outbound_stop.rs
View File

@ -25,7 +25,7 @@ use bytes::Bytes;
use futures::{future::BoxFuture, prelude::*};
use libp2p_core::upgrade;
use libp2p_identity::PeerId;
use libp2p_swarm::{NegotiatedSubstream, StreamProtocol};
use libp2p_swarm::{Stream, StreamProtocol};
use std::convert::TryInto;
use std::iter;
use std::time::Duration;
@ -46,12 +46,12 @@ impl upgrade::UpgradeInfo for Upgrade {
}
}
impl upgrade::OutboundUpgrade<NegotiatedSubstream> for Upgrade {
type Output = (NegotiatedSubstream, Bytes);
impl upgrade::OutboundUpgrade<Stream> for Upgrade {
type Output = (Stream, Bytes);
type Error = UpgradeError;
type Future = BoxFuture<'static, Result<Self::Output, Self::Error>>;
fn upgrade_outbound(self, substream: NegotiatedSubstream, _: Self::Info) -> Self::Future {
fn upgrade_outbound(self, substream: Stream, _: Self::Info) -> Self::Future {
let msg = proto::StopMessage {
type_pb: proto::StopMessageType::CONNECT,
peer: Some(proto::Peer {

12
protocols/rendezvous/src/handler/inbound.rs
View File

@ -26,7 +26,7 @@ use crate::handler::PROTOCOL_IDENT;
use crate::substream_handler::{Next, PassthroughProtocol, SubstreamHandler};
use asynchronous_codec::Framed;
use futures::{SinkExt, StreamExt};
use libp2p_swarm::{NegotiatedSubstream, SubstreamProtocol};
use libp2p_swarm::SubstreamProtocol;
use std::fmt;
use std::task::{Context, Poll};
@ -35,13 +35,13 @@ use std::task::{Context, Poll};
#[allow(clippy::enum_variant_names)]
pub enum Stream {
/// We are in the process of reading a message from the substream.
PendingRead(Framed<NegotiatedSubstream, RendezvousCodec>),
PendingRead(Framed<libp2p_swarm::Stream, RendezvousCodec>),
/// We read a message, dispatched it to the behaviour and are waiting for the response.
PendingBehaviour(Framed<NegotiatedSubstream, RendezvousCodec>),
PendingBehaviour(Framed<libp2p_swarm::Stream, RendezvousCodec>),
/// We are in the process of sending a response.
PendingSend(Framed<NegotiatedSubstream, RendezvousCodec>, Message),
PendingSend(Framed<libp2p_swarm::Stream, RendezvousCodec>, Message),
/// We've sent the message and are now closing down the substream.
PendingClose(Framed<NegotiatedSubstream, RendezvousCodec>),
PendingClose(Framed<libp2p_swarm::Stream, RendezvousCodec>),
}
impl fmt::Debug for Stream {
@ -93,7 +93,7 @@ impl SubstreamHandler for Stream {
SubstreamProtocol::new(PassthroughProtocol::new(PROTOCOL_IDENT), open_info)
}
fn new(substream: NegotiatedSubstream, _: Self::OpenInfo) -> Self {
fn new(substream: libp2p_swarm::Stream, _: Self::OpenInfo) -> Self {
Stream::PendingRead(Framed::new(substream, RendezvousCodec::default()))
}

4
protocols/rendezvous/src/handler/outbound.rs
View File

@ -25,7 +25,7 @@ use crate::substream_handler::{FutureSubstream, Next, PassthroughProtocol, Subst
use crate::{ErrorCode, Namespace, Registration, Ttl};
use asynchronous_codec::Framed;
use futures::{SinkExt, TryFutureExt, TryStreamExt};
use libp2p_swarm::{NegotiatedSubstream, SubstreamProtocol};
use libp2p_swarm::SubstreamProtocol;
use std::task::Context;
use void::Void;
@ -43,7 +43,7 @@ impl SubstreamHandler for Stream {
SubstreamProtocol::new(PassthroughProtocol::new(PROTOCOL_IDENT), open_info)
}
fn new(substream: NegotiatedSubstream, info: Self::OpenInfo) -> Self {
fn new(substream: libp2p_swarm::Stream, info: Self::OpenInfo) -> Self {
let mut stream = Framed::new(substream, RendezvousCodec::default());
let sent_message = match info {
OpenInfo::RegisterRequest(new_registration) => Message::Register(new_registration),

7
protocols/rendezvous/src/substream_handler.rs
View File

@ -31,8 +31,7 @@ use instant::Instant;
use libp2p_core::{InboundUpgrade, OutboundUpgrade, UpgradeInfo};
use libp2p_swarm::handler::{ConnectionEvent, FullyNegotiatedInbound, FullyNegotiatedOutbound};
use libp2p_swarm::{
ConnectionHandler, ConnectionHandlerEvent, KeepAlive, NegotiatedSubstream, StreamProtocol,
SubstreamProtocol,
ConnectionHandler, ConnectionHandlerEvent, KeepAlive, Stream, StreamProtocol, SubstreamProtocol,
};
use std::collections::{HashMap, VecDeque};
use std::fmt;
@ -51,7 +50,7 @@ pub trait SubstreamHandler: Sized {
fn upgrade(open_info: Self::OpenInfo)
-> SubstreamProtocol<PassthroughProtocol, Self::OpenInfo>;
fn new(substream: NegotiatedSubstream, info: Self::OpenInfo) -> Self;
fn new(substream: Stream, info: Self::OpenInfo) -> Self;
fn on_event(self, event: Self::InEvent) -> Self;
fn advance(self, cx: &mut Context<'_>) -> Result<Next<Self, Self::OutEvent>, Self::Error>;
}
@ -541,7 +540,7 @@ impl SubstreamHandler for void::Void {
type Error = void::Void;
type OpenInfo = ();
fn new(_: NegotiatedSubstream, _: Self::OpenInfo) -> Self {
fn new(_: Stream, _: Self::OpenInfo) -> Self {
unreachable!("we should never yield a substream")
}

18
protocols/request-response/src/handler/protocol.rs
View File

@ -28,7 +28,7 @@ use crate::RequestId;
use futures::{channel::oneshot, future::BoxFuture, prelude::*};
use libp2p_core::upgrade::{InboundUpgrade, OutboundUpgrade, UpgradeInfo};
use libp2p_swarm::NegotiatedSubstream;
use libp2p_swarm::Stream;
use smallvec::SmallVec;
use std::{fmt, io};
@ -88,7 +88,7 @@ where
}
}
impl<TCodec> InboundUpgrade<NegotiatedSubstream> for ResponseProtocol<TCodec>
impl<TCodec> InboundUpgrade<Stream> for ResponseProtocol<TCodec>
where
TCodec: Codec + Send + 'static,
{
@ -96,11 +96,7 @@ where
type Error = io::Error;
type Future = BoxFuture<'static, Result<Self::Output, Self::Error>>;
fn upgrade_inbound(
mut self,
mut io: NegotiatedSubstream,
protocol: Self::Info,
) -> Self::Future {
fn upgrade_inbound(mut self, mut io: Stream, protocol: Self::Info) -> Self::Future {
async move {
let read = self.codec.read_request(&protocol, &mut io);
let request = read.await?;
@ -163,7 +159,7 @@ where
}
}
impl<TCodec> OutboundUpgrade<NegotiatedSubstream> for RequestProtocol<TCodec>
impl<TCodec> OutboundUpgrade<Stream> for RequestProtocol<TCodec>
where
TCodec: Codec + Send + 'static,
{
@ -171,11 +167,7 @@ where
type Error = io::Error;
type Future = BoxFuture<'static, Result<Self::Output, Self::Error>>;
fn upgrade_outbound(
mut self,
mut io: NegotiatedSubstream,
protocol: Self::Info,
) -> Self::Future {
fn upgrade_outbound(mut self, mut io: Stream, protocol: Self::Info) -> Self::Future {
async move {
let write = self.codec.write_request(&protocol, &mut io, self.request);
write.await?;

4
swarm/CHANGELOG.md
View File

@ -42,6 +42,9 @@
See [PR 3651].
- Deprecate the `NegotiatedSubstream` type and replace it with `Stream`.
See [PR 3912].
[PR 3605]: https://github.com/libp2p/rust-libp2p/pull/3605
[PR 3651]: https://github.com/libp2p/rust-libp2p/pull/3651
[PR 3715]: https://github.com/libp2p/rust-libp2p/pull/3715
@ -51,6 +54,7 @@
[PR 3884]: https://github.com/libp2p/rust-libp2p/pull/3884
[PR 3885]: https://github.com/libp2p/rust-libp2p/pull/3885
[PR 3886]: https://github.com/libp2p/rust-libp2p/pull/3886
[PR 3912]: https://github.com/libp2p/rust-libp2p/pull/3912
## 0.42.2

1
swarm/Cargo.toml
View File

@ -26,6 +26,7 @@ void = "1"
wasm-bindgen-futures = { version = "0.4.34", optional = true }
getrandom = { version = "0.2.9", features = ["js"], optional = true } # Explicit dependency to be used in `wasm-bindgen` feature
once_cell = "1.17.1"
multistream-select = { workspace = true }
[target.'cfg(not(any(target_os = "emscripten", target_os = "wasi", target_os = "unknown")))'.dependencies]
async-std = { version = "1.6.2", optional = true }

125
swarm/src/connection.rs
View File

@ -34,10 +34,12 @@ use crate::handler::{
FullyNegotiatedOutbound, ListenUpgradeError, ProtocolSupport, ProtocolsAdded, ProtocolsChange,
UpgradeInfoSend,
};
use crate::upgrade::{InboundUpgradeSend, OutboundUpgradeSend, SendWrapper};
use crate::upgrade::{InboundUpgradeSend, OutboundUpgradeSend};
use crate::{
ConnectionHandlerEvent, KeepAlive, StreamProtocol, StreamUpgradeError, SubstreamProtocol,
ConnectionHandlerEvent, KeepAlive, Stream, StreamProtocol, StreamUpgradeError,
SubstreamProtocol,
};
use futures::future::BoxFuture;
use futures::stream::FuturesUnordered;
use futures::FutureExt;
use futures::StreamExt;
@ -47,9 +49,7 @@ use libp2p_core::connection::ConnectedPoint;
use libp2p_core::multiaddr::Multiaddr;
use libp2p_core::muxing::{StreamMuxerBox, StreamMuxerEvent, StreamMuxerExt, SubstreamBox};
use libp2p_core::upgrade;
use libp2p_core::upgrade::{
InboundUpgradeApply, NegotiationError, OutboundUpgradeApply, ProtocolError,
};
use libp2p_core::upgrade::{NegotiationError, ProtocolError};
use libp2p_core::Endpoint;
use libp2p_identity::PeerId;
use std::collections::HashSet;
@ -120,16 +120,18 @@ where
handler: THandler,
/// Futures that upgrade incoming substreams.
negotiating_in: FuturesUnordered<
SubstreamUpgrade<
StreamUpgrade<
THandler::InboundOpenInfo,
InboundUpgradeApply<SubstreamBox, SendWrapper<THandler::InboundProtocol>>,
<THandler::InboundProtocol as InboundUpgradeSend>::Output,
<THandler::InboundProtocol as InboundUpgradeSend>::Error,
>,
>,
/// Futures that upgrade outgoing substreams.
negotiating_out: FuturesUnordered<
SubstreamUpgrade<
StreamUpgrade<
THandler::OutboundOpenInfo,
OutboundUpgradeApply<SubstreamBox, SendWrapper<THandler::OutboundProtocol>>,
<THandler::OutboundProtocol as OutboundUpgradeSend>::Output,
<THandler::OutboundProtocol as OutboundUpgradeSend>::Error,
>,
>,
/// The currently planned connection & handler shutdown.
@ -396,7 +398,7 @@ where
Poll::Ready(substream) => {
let (user_data, timeout, upgrade) = requested_substream.extract();
negotiating_out.push(SubstreamUpgrade::new_outbound(
negotiating_out.push(StreamUpgrade::new_outbound(
substream,
user_data,
timeout,
@ -415,7 +417,7 @@ where
Poll::Ready(substream) => {
let protocol = handler.listen_protocol();
negotiating_in.push(SubstreamUpgrade::new_inbound(substream, protocol));
negotiating_in.push(StreamUpgrade::new_inbound(substream, protocol));
continue; // Go back to the top, handler can potentially make progress again.
}
@ -470,24 +472,23 @@ impl<'a> IncomingInfo<'a> {
}
}
struct SubstreamUpgrade<UserData, Upgrade> {
struct StreamUpgrade<UserData, TOk, TErr> {
user_data: Option<UserData>,
timeout: Delay,
upgrade: Upgrade,
upgrade: BoxFuture<'static, Result<TOk, StreamUpgradeError<TErr>>>,
}
impl<UserData, Upgrade>
SubstreamUpgrade<UserData, OutboundUpgradeApply<SubstreamBox, SendWrapper<Upgrade>>>
where
Upgrade: Send + OutboundUpgradeSend,
{
fn new_outbound(
impl<UserData, TOk, TErr> StreamUpgrade<UserData, TOk, TErr> {
fn new_outbound<Upgrade>(
substream: SubstreamBox,
user_data: UserData,
timeout: Delay,
upgrade: Upgrade,
version_override: Option<upgrade::Version>,
) -> Self {
) -> Self
where
Upgrade: OutboundUpgradeSend<Output = TOk, Error = TErr>,
{
let effective_version = match version_override {
Some(version_override) if version_override != upgrade::Version::default() => {
log::debug!(
@ -500,45 +501,77 @@ where
}
_ => upgrade::Version::default(),
};
let protocols = upgrade.protocol_info();
Self {
user_data: Some(user_data),
timeout,
upgrade: upgrade::apply_outbound(substream, SendWrapper(upgrade), effective_version),
upgrade: Box::pin(async move {
let (info, stream) = multistream_select::dialer_select_proto(
substream,
protocols,
effective_version,
)
.await
.map_err(to_stream_upgrade_error)?;
let output = upgrade
.upgrade_outbound(Stream::new(stream), info)
.await
.map_err(StreamUpgradeError::Apply)?;
Ok(output)
}),
}
}
}
impl<UserData, Upgrade>
SubstreamUpgrade<UserData, InboundUpgradeApply<SubstreamBox, SendWrapper<Upgrade>>>
where
Upgrade: Send + InboundUpgradeSend,
{
fn new_inbound(
impl<UserData, TOk, TErr> StreamUpgrade<UserData, TOk, TErr> {
fn new_inbound<Upgrade>(
substream: SubstreamBox,
protocol: SubstreamProtocol<Upgrade, UserData>,
) -> Self {
) -> Self
where
Upgrade: InboundUpgradeSend<Output = TOk, Error = TErr>,
{
let timeout = *protocol.timeout();
let (upgrade, open_info) = protocol.into_upgrade();
let protocols = upgrade.protocol_info();
Self {
user_data: Some(open_info),
timeout: Delay::new(timeout),
upgrade: upgrade::apply_inbound(substream, SendWrapper(upgrade)),
upgrade: Box::pin(async move {
let (info, stream) =
multistream_select::listener_select_proto(substream, protocols)
.await
.map_err(to_stream_upgrade_error)?;
let output = upgrade
.upgrade_inbound(Stream::new(stream), info)
.await
.map_err(StreamUpgradeError::Apply)?;
Ok(output)
}),
}
}
}
impl<UserData, Upgrade> Unpin for SubstreamUpgrade<UserData, Upgrade> {}
fn to_stream_upgrade_error<T>(e: NegotiationError) -> StreamUpgradeError<T> {
match e {
NegotiationError::Failed => StreamUpgradeError::NegotiationFailed,
NegotiationError::ProtocolError(ProtocolError::IoError(e)) => StreamUpgradeError::Io(e),
NegotiationError::ProtocolError(other) => {
StreamUpgradeError::Io(io::Error::new(io::ErrorKind::Other, other))
}
}
}
impl<UserData, Upgrade, UpgradeOutput, TUpgradeError> Future for SubstreamUpgrade<UserData, Upgrade>
where
Upgrade: Future<Output = Result<UpgradeOutput, upgrade::UpgradeError<TUpgradeError>>> + Unpin,
{
type Output = (
UserData,
Result<UpgradeOutput, StreamUpgradeError<TUpgradeError>>,
);
impl<UserData, TOk, TErr> Unpin for StreamUpgrade<UserData, TOk, TErr> {}
impl<UserData, TOk, TErr> Future for StreamUpgrade<UserData, TOk, TErr> {
type Output = (UserData, Result<TOk, StreamUpgradeError<TErr>>);
fn poll(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
match self.timeout.poll_unpin(cx) {
@ -560,21 +593,7 @@ where
.take()
.expect("Future not to be polled again once ready.");
Poll::Ready((
user_data,
result.map_err(|e| match e {
upgrade::UpgradeError::Select(NegotiationError::Failed) => {
StreamUpgradeError::NegotiationFailed
}
upgrade::UpgradeError::Select(NegotiationError::ProtocolError(
ProtocolError::IoError(e),
)) => StreamUpgradeError::Io(e),
upgrade::UpgradeError::Select(NegotiationError::ProtocolError(other)) => {
StreamUpgradeError::Io(io::Error::new(io::ErrorKind::Other, other))
}
upgrade::UpgradeError::Apply(e) => StreamUpgradeError::Apply(e),
}),
))
Poll::Ready((user_data, result))
}
}

6
swarm/src/handler/multi.rs
View File

@ -27,7 +27,7 @@ use crate::handler::{
SubstreamProtocol,
};
use crate::upgrade::{InboundUpgradeSend, OutboundUpgradeSend, UpgradeInfoSend};
use crate::NegotiatedSubstream;
use crate::Stream;
use futures::{future::BoxFuture, prelude::*};
use rand::Rng;
use std::{
@ -373,7 +373,7 @@ where
type Error = (K, <H as InboundUpgradeSend>::Error);
type Future = BoxFuture<'static, Result<Self::Output, Self::Error>>;
fn upgrade_inbound(mut self, resource: NegotiatedSubstream, info: Self::Info) -> Self::Future {
fn upgrade_inbound(mut self, resource: Stream, info: Self::Info) -> Self::Future {
let IndexedProtoName(index, info) = info;
let (key, upgrade) = self.upgrades.remove(index);
upgrade
@ -395,7 +395,7 @@ where
type Error = (K, <H as OutboundUpgradeSend>::Error);
type Future = BoxFuture<'static, Result<Self::Output, Self::Error>>;
fn upgrade_outbound(mut self, resource: NegotiatedSubstream, info: Self::Info) -> Self::Future {
fn upgrade_outbound(mut self, resource: Stream, info: Self::Info) -> Self::Future {
let IndexedProtoName(index, info) = info;
let (key, upgrade) = self.upgrades.remove(index);
upgrade

12
swarm/src/lib.rs
View File

@ -58,6 +58,7 @@
mod connection;
mod executor;
mod registry;
mod stream;
mod stream_protocol;
#[cfg(test)]
mod test;
@ -125,6 +126,7 @@ pub use handler::{
#[cfg(feature = "macros")]
pub use libp2p_swarm_derive::NetworkBehaviour;
pub use registry::{AddAddressResult, AddressRecord, AddressScore};
pub use stream::Stream;
pub use stream_protocol::{InvalidProtocol, StreamProtocol};
use crate::handler::UpgradeInfoSend;
@ -135,14 +137,13 @@ use connection::{
};
use dial_opts::{DialOpts, PeerCondition};
use futures::{executor::ThreadPoolBuilder, prelude::*, stream::FusedStream};
use libp2p_core::muxing::SubstreamBox;
use libp2p_core::{
connection::ConnectedPoint,
multiaddr,
multihash::Multihash,
muxing::StreamMuxerBox,
transport::{self, ListenerId, TransportError, TransportEvent},
Endpoint, Multiaddr, Negotiated, Transport,
Endpoint, Multiaddr, Transport,
};
use libp2p_identity::PeerId;
use registry::{AddressIntoIter, Addresses};
@ -160,7 +161,8 @@ use std::{
///
/// Implements the [`AsyncRead`](futures::io::AsyncRead) and
/// [`AsyncWrite`](futures::io::AsyncWrite) traits.
pub type NegotiatedSubstream = Negotiated<SubstreamBox>;
#[deprecated(note = "The 'substream' terminology is deprecated. Use 'Stream' instead")]
pub type NegotiatedSubstream = Stream;
/// Event generated by the [`NetworkBehaviour`] that the swarm will report back.
type TBehaviourOutEvent<TBehaviour> = <TBehaviour as NetworkBehaviour>::OutEvent;
@ -1378,8 +1380,8 @@ where
/// connection and listener status. See [`SwarmEvent`] for details.
///
/// Note: This stream is infinite and it is guaranteed that
/// [`Stream::poll_next`] will never return `Poll::Ready(None)`.
impl<TBehaviour> Stream for Swarm<TBehaviour>
/// [`futures::Stream::poll_next`] will never return `Poll::Ready(None)`.
impl<TBehaviour> futures::Stream for Swarm<TBehaviour>
where
TBehaviour: NetworkBehaviour,
{

59
swarm/src/stream.rs Normal file
View File

@ -0,0 +1,59 @@
use futures::{AsyncRead, AsyncWrite};
use libp2p_core::muxing::SubstreamBox;
use libp2p_core::Negotiated;
use std::io::{IoSlice, IoSliceMut};
use std::pin::Pin;
use std::task::{Context, Poll};
#[derive(Debug)]
pub struct Stream(Negotiated<SubstreamBox>);
impl Stream {
pub(crate) fn new(stream: Negotiated<SubstreamBox>) -> Self {
Self(stream)
}
}
impl AsyncRead for Stream {
fn poll_read(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &mut [u8],
) -> Poll<std::io::Result<usize>> {
Pin::new(&mut self.get_mut().0).poll_read(cx, buf)
}
fn poll_read_vectored(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
bufs: &mut [IoSliceMut<'_>],
) -> Poll<std::io::Result<usize>> {
Pin::new(&mut self.get_mut().0).poll_read_vectored(cx, bufs)
}
}
impl AsyncWrite for Stream {
fn poll_write(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &[u8],
) -> Poll<std::io::Result<usize>> {
Pin::new(&mut self.get_mut().0).poll_write(cx, buf)
}
fn poll_write_vectored(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
bufs: &[IoSlice<'_>],
) -> Poll<std::io::Result<usize>> {
Pin::new(&mut self.get_mut().0).poll_write_vectored(cx, bufs)
}
fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
Pin::new(&mut self.get_mut().0).poll_flush(cx)
}
fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
Pin::new(&mut self.get_mut().0).poll_close(cx)
}
}

22
swarm/src/upgrade.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::NegotiatedSubstream;
use crate::Stream;
use futures::prelude::*;
use libp2p_core::upgrade;
@ -66,12 +66,12 @@ pub trait OutboundUpgradeSend: UpgradeInfoSend {
type Future: Future<Output = Result<Self::Output, Self::Error>> + Send + 'static;
/// Equivalent to [`OutboundUpgrade::upgrade_outbound`](upgrade::OutboundUpgrade::upgrade_outbound).
fn upgrade_outbound(self, socket: NegotiatedSubstream, info: Self::Info) -> Self::Future;
fn upgrade_outbound(self, socket: Stream, info: Self::Info) -> Self::Future;
}
impl<T, TInfo> OutboundUpgradeSend for T
where
T: upgrade::OutboundUpgrade<NegotiatedSubstream, Info = TInfo> + UpgradeInfoSend<Info = TInfo>,
T: upgrade::OutboundUpgrade<Stream, Info = TInfo> + UpgradeInfoSend<Info = TInfo>,
TInfo: AsRef<str> + Clone + Send + 'static,
T::Output: Send + 'static,
T::Error: Send + 'static,
@ -81,7 +81,7 @@ where
type Error = T::Error;
type Future = T::Future;
fn upgrade_outbound(self, socket: NegotiatedSubstream, info: TInfo) -> Self::Future {
fn upgrade_outbound(self, socket: Stream, info: TInfo) -> Self::Future {
upgrade::OutboundUpgrade::upgrade_outbound(self, socket, info)
}
}
@ -100,12 +100,12 @@ pub trait InboundUpgradeSend: UpgradeInfoSend {
type Future: Future<Output = Result<Self::Output, Self::Error>> + Send + 'static;
/// Equivalent to [`InboundUpgrade::upgrade_inbound`](upgrade::InboundUpgrade::upgrade_inbound).
fn upgrade_inbound(self, socket: NegotiatedSubstream, info: Self::Info) -> Self::Future;
fn upgrade_inbound(self, socket: Stream, info: Self::Info) -> Self::Future;
}
impl<T, TInfo> InboundUpgradeSend for T
where
T: upgrade::InboundUpgrade<NegotiatedSubstream, Info = TInfo> + UpgradeInfoSend<Info = TInfo>,
T: upgrade::InboundUpgrade<Stream, Info = TInfo> + UpgradeInfoSend<Info = TInfo>,
TInfo: AsRef<str> + Clone + Send + 'static,
T::Output: Send + 'static,
T::Error: Send + 'static,
@ -115,7 +115,7 @@ where
type Error = T::Error;
type Future = T::Future;
fn upgrade_inbound(self, socket: NegotiatedSubstream, info: TInfo) -> Self::Future {
fn upgrade_inbound(self, socket: Stream, info: TInfo) -> Self::Future {
upgrade::InboundUpgrade::upgrade_inbound(self, socket, info)
}
}
@ -137,22 +137,22 @@ impl<T: UpgradeInfoSend> upgrade::UpgradeInfo for SendWrapper<T> {
}
}
impl<T: OutboundUpgradeSend> upgrade::OutboundUpgrade<NegotiatedSubstream> for SendWrapper<T> {
impl<T: OutboundUpgradeSend> upgrade::OutboundUpgrade<Stream> for SendWrapper<T> {
type Output = T::Output;
type Error = T::Error;
type Future = T::Future;
fn upgrade_outbound(self, socket: NegotiatedSubstream, info: T::Info) -> Self::Future {
fn upgrade_outbound(self, socket: Stream, info: T::Info) -> Self::Future {
OutboundUpgradeSend::upgrade_outbound(self.0, socket, info)
}
}
impl<T: InboundUpgradeSend> upgrade::InboundUpgrade<NegotiatedSubstream> for SendWrapper<T> {
impl<T: InboundUpgradeSend> upgrade::InboundUpgrade<Stream> for SendWrapper<T> {
type Output = T::Output;
type Error = T::Error;
type Future = T::Future;
fn upgrade_inbound(self, socket: NegotiatedSubstream, info: T::Info) -> Self::Future {
fn upgrade_inbound(self, socket: Stream, info: T::Info) -> Self::Future {
InboundUpgradeSend::upgrade_inbound(self.0, socket, info)
}
}