pass Acceptor as mut ref + remove futures depend

change version of quic dependencies to '*' to follow quinn depend versions

Cargo.toml

@@ -22,7 +22,7 @@ tls = ["tokio-native-tls"]
 # Noise support
 noise = ["snowstorm", "base64"]
 #QUIC support
-quic = ["quinn", "rustls", "rustls-pemfile", "openssl", "futures-util", "futures"]
+quic = ["quinn", "rustls", "rustls-pemfile", "openssl", "futures-util"]
 
 # Configuration hot-reload support
 hot-reload = ["notify"]
@@ -74,11 +74,10 @@ console-subscriber = { version = "0.1", optional = true, features = ["parking_lo
 const_format = "0.2"
 atty = "0.2"
 quinn = { version = "0.8.0", optional = true}
-rustls = { version = "0.20", default-features = false, features = ["quic"], optional = true }
-rustls-pemfile = { version = "0.2.1", optional = true }
+rustls = { version = "*", default-features = false, features = ["quic"], optional = true }
+rustls-pemfile = { version = "*", optional = true }
 openssl = { version = "0.10.38", optional = true }
-futures-util = { version = "0.3.11", optional = true}
-futures = { version = "0.3.11", optional = true}
+futures-util = { version = "*", optional = true}
 
 [build-dependencies]
 vergen = { version = "6.0", default-features = false, features = ["build", "git", "cargo"] }

src/server.rs

@@ -131,7 +131,7 @@ impl<'a, T: 'static + Transport> Server<'a, T> {
         mut service_rx: mpsc::Receiver<ServiceChange>,
     ) -> Result<()> {
         // Listen at `server.bind_addr`
-        let l = self
+        let mut l = self
             .transport
             .bind(&self.config.bind_addr)
             .await
@@ -149,7 +149,7 @@ impl<'a, T: 'static + Transport> Server<'a, T> {
         loop {
             tokio::select! {
                 // Wait for incoming control and data channels
-                ret = self.transport.accept(&l) => {
+                ret = self.transport.accept(&mut l) => {
                     match ret {
                         Err(err) => {
                             // Detects whether it's an IO error

src/transport/mod.rs

@@ -17,7 +17,7 @@ pub trait Transport: Debug + Send + Sync {
     where
         Self: Sized;
     async fn bind<T: ToSocketAddrs + Send + Sync>(&self, addr: T) -> Result<Self::Acceptor>;
-    async fn accept(&self, a: &Self::Acceptor) -> Result<(Self::RawStream, SocketAddr)>;
+    async fn accept(&self, a: &mut Self::Acceptor) -> Result<(Self::RawStream, SocketAddr)>;
     async fn handshake(&self, conn: Self::RawStream) -> Result<Self::Stream>;
     async fn connect(&self, addr: &str) -> Result<Self::Stream>;
     async fn close(&self, a: Self::Acceptor);

src/transport/noise.rs

@@ -72,7 +72,7 @@ impl Transport for NoiseTransport {
         Ok(TcpListener::bind(addr).await?)
     }
 
-    async fn accept(&self, a: &Self::Acceptor) -> Result<(Self::RawStream, SocketAddr)> {
+    async fn accept(&self, a: &mut Self::Acceptor) -> Result<(Self::RawStream, SocketAddr)> {
         let (conn, addr) = a
             .accept()
             .await

src/transport/quic.rs

@@ -1,4 +1,3 @@
-use futures::lock::Mutex;
 use std::borrow::{Borrow, BorrowMut};
 use std::fmt::{Debug, Formatter};
 use std::io;
@@ -15,7 +14,7 @@ use super::Transport;
 use crate::config::{TlsConfig, TransportConfig};
 use anyhow::{anyhow, Context, Result};
 use async_trait::async_trait;
-use futures_util::{ready, StreamExt};
+use futures_util::{StreamExt};
 use openssl::pkcs12::Pkcs12;
 use quinn::{Connecting, Connection, Endpoint, EndpointConfig, Incoming, NewConnection};
 use rustls::internal::msgs::codec::Codec;
@@ -103,11 +102,11 @@ impl AsyncWrite for QuicBiStream {
     }
 }
 
-pub struct QuicAcceptor(Arc<Mutex<(Endpoint, Incoming)>>);
+pub struct QuicAcceptor(Endpoint, Incoming);
 
 impl Into<Endpoint> for QuicAcceptor {
     fn into(self) -> Endpoint {
-        Arc::try_unwrap(self.0).unwrap().into_inner().0
+        self.0
     }
 }
 
@@ -212,12 +211,12 @@ impl Transport for QuicTransport {
         let socket = UdpSocket::bind(addr).await?.into_std()?;
         quinn::Endpoint::new(EndpointConfig::default(), Some(server_config), socket)
             .with_context(|| "Failed to start server")
-            .map(|e_i| QuicAcceptor(Arc::new(Mutex::new(e_i))))
+            .map(|(e, i)| QuicAcceptor(e, i))
     }
 
 
-    async fn accept(&self, a: &Self::Acceptor) -> Result<(Self::RawStream, SocketAddr)> {
-        while let Some(connecting) = a.0.lock().await.1.next().await {
+    async fn accept(&self, a: &mut Self::Acceptor) -> Result<(Self::RawStream, SocketAddr)> {
+        while let Some(connecting) = a.1.next().await {
             let addr = connecting.remote_address();
             return Ok((connecting, addr));
         }
@@ -258,7 +257,7 @@ impl Transport for QuicTransport {
     }
 
     async fn close(&self, a: Self::Acceptor) {
-        let e: Endpoint = a.into();
+        let e: Endpoint = a.into();  // drops Incoming
         e.close(0u8.into(), &[]);
         // wait for all connections to signal close as per spec
         // See https://github.com/quinn-rs/quinn/issues/1102

src/transport/tcp.rs

@@ -24,7 +24,7 @@ impl Transport for TcpTransport {
         Ok(TcpListener::bind(addr).await?)
     }
 
-    async fn accept(&self, a: &Self::Acceptor) -> Result<(Self::RawStream, SocketAddr)> {
+    async fn accept(&self, a: &mut Self::Acceptor) -> Result<(Self::RawStream, SocketAddr)> {
         let (s, addr) = a.accept().await?;
         set_tcp_keepalive(&s);
         Ok((s, addr))

src/transport/tls.rs

@@ -74,7 +74,7 @@ impl Transport for TlsTransport {
         Ok(l)
     }
 
-    async fn accept(&self, a: &Self::Acceptor) -> Result<(Self::RawStream, SocketAddr)> {
+    async fn accept(&self, a: &mut Self::Acceptor) -> Result<(Self::RawStream, SocketAddr)> {
         let (conn, addr) = a.accept().await?;
         set_tcp_keepalive(&conn);