Upgrade quinn to 0.9

hotham-asset-client/Cargo.toml

@@ -8,7 +8,7 @@ version = "0.1.0"
 [dependencies]
 anyhow = "1.0"
 futures-util = {version = "0.3.11", default-features = false}
-quinn = {version = "0.8.5", features = ["tls-rustls"]}
+quinn = {version = "0.9", features = ["tls-rustls"]}
 rustls = {version = "0.20.3", features = ["dangerous_configuration", "quic"]}
 tokio = {version = "1.0.1", default-features = false, features = ["sync"]}
 

hotham-asset-client/src/client.rs

@@ -1,7 +1,7 @@
 use crate::message::Message;
 use crate::AssetUpdatedMessage;
 use anyhow::{anyhow, bail, Context, Result};
-use futures_util::{StreamExt, TryStreamExt};
+use futures_util::TryStreamExt;
 use quinn::{ClientConfig, Endpoint};
 use std::sync::Arc;
 use tokio::sync::mpsc::Sender;
@@ -23,11 +23,7 @@ async fn run_client(
     endpoint.set_default_client_config(client_cfg);
 
     // connect to server
-    let quinn::NewConnection {
-        connection,
-        bi_streams,
-        ..
-    } = endpoint
+    let connection = endpoint
         .connect(server_addr, "hotham_asset_server")?
         .await?;
     println!("[CLIENT] connected: addr={}", connection.remote_address());
@@ -42,7 +38,7 @@ async fn run_client(
         .try_collect::<Vec<_>>()
         .await?;
 
-    wait_for_updates(bi_streams, connection.clone(), sender)
+    wait_for_updates(connection.clone(), sender)
         .await
         .context("Watching file")?;
 
@@ -55,12 +51,11 @@ async fn run_client(
 }
 
 async fn wait_for_updates(
-    mut bi_streams: quinn::IncomingBiStreams,
     connection: quinn::Connection,
     sender: Sender<AssetUpdatedMessage>,
 ) -> Result<()> {
-    while let Some(stream) = bi_streams.next().await {
-        let stream = match stream {
+    loop {
+        let stream = match connection.accept_bi().await {
             Err(quinn::ConnectionError::ApplicationClosed { .. }) => {
                 println!("[CLIENT] Connection closed");
                 return Ok(());
@@ -72,8 +67,6 @@ async fn wait_for_updates(
         };
         tokio::spawn(handle_incoming(stream, connection.clone(), sender.clone()));
     }
-
-    Ok(())
 }
 
 async fn ask_for_watch(connection: quinn::Connection, asset_name: String) -> Result<()> {

hotham-asset-server/Cargo.toml

@@ -10,7 +10,7 @@ anyhow = "1.0"
 futures-util = {version = "0.3.11", default-features = false, features = ["io"]}
 hotham-asset-client = {path = "../hotham-asset-client"}
 notify-debouncer-mini = {version = "*", default-features = false}
-quinn = {version = "0.8.5", features = ["tls-rustls"]}
+quinn = {version = "0.9", features = ["tls-rustls"]}
 rcgen = "0.10.0"
 rustls = {version = "0.20.3", features = ["dangerous_configuration", "quic"]}
 tokio = {version = "1.21.2", features = ["full"]}

hotham-asset-server/src/main.rs

@@ -4,7 +4,6 @@ use std::{collections::HashMap, sync::Arc, time::SystemTime};
 use tokio::sync::Mutex;
 
 use anyhow::Result;
-use futures_util::StreamExt;
 /// A simple server that serves assets to localhost or remote targets. It's great and has no flaws.
 // TODO:
 // 1. Accept connections
@@ -19,16 +18,15 @@ pub type WatchList = Arc<Mutex<HashMap<String, anyhow::Result<SystemTime>>>>;
 #[tokio::main]
 async fn main() -> Result<()> {
     let addr = "0.0.0.0:5000".parse().unwrap();
-    let (mut incoming, _server_cert) = make_server_endpoint(addr).unwrap();
+    let (endpoint, _server_cert) = make_server_endpoint(addr).unwrap();
 
     loop {
-        let incoming_conn = incoming.next().await.unwrap();
+        let incoming_conn = endpoint.accept().await.unwrap();
         let new_conn = incoming_conn.await.unwrap();
         let watch_list = WatchList::default();
 
         let watcher_watch_list = watch_list.clone();
-        let watcher_connection = new_conn.connection.clone();
-        tokio::spawn(watch_files(watcher_connection, watcher_watch_list));
+        tokio::spawn(watch_files(new_conn.clone(), watcher_watch_list));
         tokio::spawn(handle_connection(new_conn, watch_list.clone()));
     }
 }

hotham-asset-server/src/server.rs

@@ -1,5 +1,5 @@
 use anyhow::{bail, Result};
-use futures_util::{StreamExt, TryFutureExt};
+use futures_util::TryFutureExt;
 use hotham_asset_client::message::Message;
 /// A simple server that serves assets to localhost or remote targets. It's great and has no flaws.
 // TODO:
@@ -8,7 +8,7 @@ use hotham_asset_client::message::Message;
 // 3. Watch for file updates
 // 4. Send a "file updated" message back to the client on update
 // 5. GOTO 2
-use quinn::{Endpoint, Incoming, ServerConfig};
+use quinn::{Endpoint, ServerConfig};
 use std::{
     error::Error,
     net::SocketAddr,
@@ -18,11 +18,10 @@ use std::{
 
 use crate::WatchList;
 
-pub async fn handle_connection(conn: quinn::NewConnection, watch_list: WatchList) -> Result<()> {
+pub async fn handle_connection(conn: quinn::Connection, watch_list: WatchList) -> Result<()> {
     println!("[SERVER] Connection established!");
-    let mut bi_streams = conn.bi_streams;
-    while let Some(stream) = bi_streams.next().await {
-        let (send, recv) = match stream {
+    loop {
+        let (send, recv) = match conn.accept_bi().await {
             Err(quinn::ConnectionError::ApplicationClosed { .. }) => {
                 println!("[SERVER] Connection closed");
                 return Ok(());
@@ -37,7 +36,6 @@ pub async fn handle_connection(conn: quinn::NewConnection, watch_list: WatchList
                 .map_err(|e| eprintln!("[SERVER] Error in incoming: {e:?}")),
         );
     }
-    Ok(())
 }
 
 pub async fn watch_files(connection: quinn::Connection, watch_list: WatchList) {
@@ -152,10 +150,10 @@ async fn get_last_updated(path: &str) -> anyhow::Result<SystemTime> {
     Ok(tokio::fs::metadata(path).await?.modified()?)
 }
 
-pub fn make_server_endpoint(bind_addr: SocketAddr) -> Result<(Incoming, Vec<u8>), Box<dyn Error>> {
+pub fn make_server_endpoint(bind_addr: SocketAddr) -> Result<(Endpoint, Vec<u8>), Box<dyn Error>> {
     let (server_config, server_cert) = configure_server()?;
-    let (_endpoint, incoming) = Endpoint::server(server_config, bind_addr)?;
-    Ok((incoming, server_cert))
+    let endpoint = Endpoint::server(server_config, bind_addr)?;
+    Ok((endpoint, server_cert))
 }
 
 /// Returns default server configuration along with its certificate.