Skip to content

Commit

Permalink
transports/quic: apply suggestions from review
Browse files Browse the repository at this point in the history
Address leftovers from review in elenaf9#6.
  • Loading branch information
elenaf9 committed Sep 22, 2022
1 parent 7f902b4 commit 9df25f1
Show file tree
Hide file tree
Showing 4 changed files with 19 additions and 23 deletions.
2 changes: 1 addition & 1 deletion swarm-derive/src/lib.rs
Expand Up @@ -125,7 +125,7 @@ fn build_struct(ast: &DeriveInput, data_struct: &DataStruct) -> TokenStream {
)
.unwrap();
let ty = &field.ty;
quote! {#variant(<#ty as ::libp2p::swarm::NetworkBehaviour>::OutEvent)}
quote! {#variant(<#ty as NetworkBehaviour>::OutEvent)}
})
.collect::<Vec<_>>();
let visibility = &ast.vis;
Expand Down
16 changes: 4 additions & 12 deletions transports/quic/src/endpoint.rs
Expand Up @@ -40,7 +40,6 @@ use futures::{
use quinn_proto::{ClientConfig as QuinnClientConfig, ServerConfig as QuinnServerConfig};
use std::{
collections::HashMap,
fmt,
net::{Ipv4Addr, Ipv6Addr, SocketAddr, UdpSocket},
sync::Arc,
task::{Context, Poll},
Expand Down Expand Up @@ -88,7 +87,7 @@ impl Config {
}

/// Object containing all the QUIC resources shared between all connections.
#[derive(Clone)]
#[derive(Debug, Clone)]
pub struct Endpoint {
/// Channel to the background of the endpoint.
to_endpoint: mpsc::Sender<ToEndpoint>,
Expand Down Expand Up @@ -456,8 +455,9 @@ async fn background_task(
return;
}
}
_ => tracing::warn!(
"Dropping new incoming connection because the channel to the listener is full."
Err(_) => tracing::warn!(
"Dropping new incoming connection {:?} because the channel to the listener is full",
connec_id
)
}
},
Expand All @@ -466,11 +466,3 @@ async fn background_task(
}
}
}

impl fmt::Debug for Endpoint {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
f.debug_struct("Endpoint")
.field("socket_addr", &self.socket_addr)
.finish()
}
}
22 changes: 13 additions & 9 deletions transports/quic/src/muxer.rs
Expand Up @@ -195,25 +195,29 @@ impl StreamMuxer for QuicMuxer {
}

fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
let mut inner = self.inner.lock();
if inner.connection.connection.is_drained() {
let Inner {
substreams,
connection,
..
} = &mut *self.inner.lock();
if connection.connection.is_drained() {
return Poll::Ready(Ok(()));
}

if inner.connection.connection.streams().send_streams() != 0 {
for substream in inner.substreams.keys().cloned().collect::<Vec<_>>() {
if let Err(e) = inner.connection.finish_substream(substream) {
if connection.connection.streams().send_streams() != 0 {
for substream in substreams.keys() {
if let Err(e) = connection.finish_substream(*substream) {
tracing::warn!("substream finish error on muxer close: {}", e);
}
}
}
loop {
if inner.connection.connection.streams().send_streams() == 0
&& !inner.connection.connection.is_closed()
if connection.connection.streams().send_streams() == 0
&& !connection.connection.is_closed()
{
inner.connection.close()
connection.close()
}
match inner.connection.poll_event(cx) {
match connection.poll_event(cx) {
Poll::Ready(ConnectionEvent::ConnectionLost(_)) => return Poll::Ready(Ok(())),
Poll::Ready(_) => {}
Poll::Pending => break,
Expand Down
2 changes: 1 addition & 1 deletion transports/quic/src/transport.rs
Expand Up @@ -411,7 +411,7 @@ impl Stream for Listener {
///
/// Returns `None` if the address is not the same socket family as the
/// address that the endpoint is bound to.
pub fn ip_to_listenaddr(endpoint: &Endpoint, ip: IpAddr) -> Option<Multiaddr> {
fn ip_to_listenaddr(endpoint: &Endpoint, ip: IpAddr) -> Option<Multiaddr> {
// True if either both addresses are Ipv4 or both Ipv6.
let is_same_ip_family = endpoint.socket_addr().is_ipv4() == ip.is_ipv4();
if !is_same_ip_family {
Expand Down

0 comments on commit 9df25f1

Please sign in to comment.