Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat(relay): hide internals of Connection #3829

Merged
merged 15 commits into from
May 2, 2023
Merged

feat(relay): hide internals of Connection #3829

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
15 commits
Select commit Hold shift + click to select a range
c7f099b
new type client::Connection with AsyncRead and AsyncWrite
tcoratger Apr 25, 2023
9d132c6
ConnectionState pub enum to private
tcoratger Apr 25, 2023
89f12ad
modify RelayedConnection deprecation message
tcoratger Apr 25, 2023
6bdc9ac
remove doc private ConnectionState
tcoratger Apr 25, 2023
e7c79a9
Update protocols/relay/src/priv_client.rs
tcoratger Apr 25, 2023
8ad71b1
Merge branch 'master' into relay-connection
tcoratger Apr 25, 2023
0e8f81e
remove dead code allowance
tcoratger Apr 25, 2023
ef550ba
Merge branch 'master' into relay-connection
tcoratger Apr 26, 2023
308db96
Document drop notifier pattern
thomaseizinger Apr 26, 2023
5716d71
Merge branch 'master' into relay-connection
tcoratger Apr 26, 2023
86abf25
Merge branch 'master' into relay-connection
tcoratger Apr 28, 2023
add9d33
Merge branch 'master' into relay-connection
tcoratger Apr 28, 2023
bc6333a
Merge branch 'master' into relay-connection
thomaseizinger May 2, 2023
3dbb583
add changelog entry
tcoratger May 2, 2023
0d57d6f
Merge branch 'master' into relay-connection
mergify[bot] May 2, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
72 changes: 43 additions & 29 deletions protocols/relay/src/priv_client.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,7 +45,6 @@ use libp2p_swarm::{
};
use std::collections::{hash_map, HashMap, VecDeque};
use std::io::{Error, ErrorKind, IoSlice};
use std::ops::DerefMut;
use std::pin::Pin;
use std::task::{Context, Poll};
use transport::Transport;
Expand Down Expand Up @@ -387,10 +386,15 @@ impl NetworkBehaviour for Behaviour {
}
}

/// A [`NegotiatedSubstream`] acting as a [`Connection`].
pub enum Connection {
pub struct Connection {
tcoratger marked this conversation as resolved.
Show resolved Hide resolved
state: ConnectionState,
}

/// A [`NegotiatedSubstream`] acting as a [`ConnectionState`].
tcoratger marked this conversation as resolved.
Show resolved Hide resolved
#[allow(dead_code)]
tcoratger marked this conversation as resolved.
Show resolved Hide resolved
enum ConnectionState {
InboundAccepting {
accept: BoxFuture<'static, Result<Connection, Error>>,
accept: BoxFuture<'static, Result<ConnectionState, Error>>,
},
Operational {
read_buffer: Bytes,
Expand All @@ -399,20 +403,20 @@ pub enum Connection {
},
}

impl Unpin for Connection {}
impl Unpin for ConnectionState {}

impl Connection {
impl ConnectionState {
pub(crate) fn new_inbound(
circuit: inbound_stop::Circuit,
drop_notifier: oneshot::Sender<void::Void>,
) -> Self {
Connection::InboundAccepting {
ConnectionState::InboundAccepting {
accept: async {
let (substream, read_buffer) = circuit
.accept()
.await
.map_err(|e| Error::new(ErrorKind::Other, e))?;
Ok(Connection::Operational {
Ok(ConnectionState::Operational {
read_buffer,
substream,
drop_notifier,
Expand All @@ -427,7 +431,7 @@ impl Connection {
read_buffer: Bytes,
drop_notifier: oneshot::Sender<void::Void>,
) -> Self {
Connection::Operational {
ConnectionState::Operational {
substream,
read_buffer,
drop_notifier,
Expand All @@ -442,35 +446,41 @@ impl AsyncWrite for Connection {
buf: &[u8],
) -> Poll<Result<usize, Error>> {
loop {
match self.deref_mut() {
Connection::InboundAccepting { accept } => {
*self = ready!(accept.poll_unpin(cx))?;
match &mut self.state {
ConnectionState::InboundAccepting { accept } => {
*self = Connection {
state: ready!(accept.poll_unpin(cx))?,
};
}
Connection::Operational { substream, .. } => {
ConnectionState::Operational { substream, .. } => {
return Pin::new(substream).poll_write(cx, buf);
}
}
}
}
fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Result<(), Error>> {
loop {
match self.deref_mut() {
Connection::InboundAccepting { accept } => {
*self = ready!(accept.poll_unpin(cx))?;
match &mut self.state {
ConnectionState::InboundAccepting { accept } => {
*self = Connection {
state: ready!(accept.poll_unpin(cx))?,
};
}
Connection::Operational { substream, .. } => {
ConnectionState::Operational { substream, .. } => {
return Pin::new(substream).poll_flush(cx);
}
}
}
}
fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Result<(), Error>> {
loop {
match self.deref_mut() {
Connection::InboundAccepting { accept } => {
*self = ready!(accept.poll_unpin(cx))?;
match &mut self.state {
ConnectionState::InboundAccepting { accept } => {
*self = Connection {
state: ready!(accept.poll_unpin(cx))?,
};
}
Connection::Operational { substream, .. } => {
ConnectionState::Operational { substream, .. } => {
return Pin::new(substream).poll_close(cx);
}
}
Expand All @@ -483,11 +493,13 @@ impl AsyncWrite for Connection {
bufs: &[IoSlice],
) -> Poll<Result<usize, Error>> {
loop {
match self.deref_mut() {
Connection::InboundAccepting { accept } => {
*self = ready!(accept.poll_unpin(cx))?;
match &mut self.state {
ConnectionState::InboundAccepting { accept } => {
*self = Connection {
state: ready!(accept.poll_unpin(cx))?,
};
}
Connection::Operational { substream, .. } => {
ConnectionState::Operational { substream, .. } => {
return Pin::new(substream).poll_write_vectored(cx, bufs);
}
}
Expand All @@ -502,11 +514,13 @@ impl AsyncRead for Connection {
buf: &mut [u8],
) -> Poll<Result<usize, Error>> {
loop {
match self.deref_mut() {
Connection::InboundAccepting { accept } => {
*self = ready!(accept.poll_unpin(cx))?;
match &mut self.state {
ConnectionState::InboundAccepting { accept } => {
*self = Connection {
state: ready!(accept.poll_unpin(cx))?,
};
}
Connection::Operational {
ConnectionState::Operational {
read_buffer,
substream,
..
Expand Down
13 changes: 6 additions & 7 deletions protocols/relay/src/priv_client/handler.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -189,10 +189,11 @@ impl Handler {

let (tx, rx) = oneshot::channel();
self.alive_lend_out_substreams.push(rx);
let connection = super::Connection::new_inbound(inbound_circuit, tx);
let connection = super::ConnectionState::new_inbound(inbound_circuit, tx);

pending_msgs.push_back(transport::ToListenerMsg::IncomingRelayedConnection {
stream: connection,
// stream: connection,
stream: super::Connection { state: connection },
src_peer_id,
relay_peer_id: self.remote_peer_id,
relay_addr: self.remote_addr.clone(),
Expand Down Expand Up @@ -271,11 +272,9 @@ impl Handler {
OutboundOpenInfo::Connect { send_back },
) => {
let (tx, rx) = oneshot::channel();
match send_back.send(Ok(super::Connection::new_outbound(
substream,
read_buffer,
tx,
))) {
match send_back.send(Ok(super::Connection {
state: super::ConnectionState::new_outbound(substream, read_buffer, tx),
})) {
Ok(()) => {
self.alive_lend_out_substreams.push(rx);
self.queued_events.push_back(ConnectionHandlerEvent::Custom(
Expand Down
2 changes: 1 addition & 1 deletion protocols/relay/src/v2.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,7 @@ pub mod client {

#[deprecated(
since = "0.15.0",
note = "Use libp2p_relay::client::Connection instead."
note = "Use libp2p_relay::client::ConnectionState instead."
tcoratger marked this conversation as resolved.
Show resolved Hide resolved
)]
pub type RelayedConnection = crate::client::Connection;

Expand Down