Merge branch 'main' into no_publish

Author: SirCipher

Cargo.toml

@@ -120,7 +120,7 @@ duration-str = "0.11.2"
 quick-xml = "0.34.0"
 csv = "1.2"
 serde-xml-rs = "0.6"
-axum = "0.6.20"
+axum = "0.7.5"
 hyper-staticfile = "0.9"
 httparse = "1.8"
 sha-1 = "0.10.1"

example_apps/transit/Cargo.toml

@@ -7,6 +7,7 @@ publish = false
 [dependencies]
 swimos = { path = "../../swimos", features = ["server", "agent"] }
 swimos_form = { path = "../../api/swimos_form" }
+swimos_utilities = { path = "../../swimos_utilities", features = ["trigger"] }
 tokio = { workspace = true, features = ["rt-multi-thread", "macros", "sync", "io-util", "fs"] }
 tokio-stream = { workspace = true, features = ["io-util"] }
 tokio-util = { workspace = true, features = ["io"] }

example_apps/transit/src/bin/ui.rs

@@ -12,11 +12,14 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::{error::Error, pin::pin, sync::Arc, time::Duration};
+use std::future::IntoFuture;
+use std::{error::Error, pin::pin, time::Duration};
 
 use clap::Parser;
 use futures::future::{select, Either};
-use tokio::sync::Notify;
+use tokio::net::TcpListener;
+
+use swimos_utilities::trigger::trigger;
 use transit::ui::ui_server_router;
 
 const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(1);
@@ -32,19 +35,24 @@ async fn ui_server(
     shutdown_timeout: Duration,
 ) -> Result<(), Box<dyn Error + Send + Sync>> {
     let app = ui_server_router(port);
-    let server = axum::Server::try_bind(&"0.0.0.0:0".parse()?)?.serve(app.into_make_service());
-    let ui_addr = server.local_addr();
+    let (stop_tx, stop_rx) = trigger();
+
+    let listener = TcpListener::bind("0.0.0.0:0").await?;
+    let ui_addr = listener.local_addr()?;
     println!("UI bound to: {}", ui_addr);
-    let stop_tx = Arc::new(Notify::new());
-    let stop_rx = stop_tx.clone();
-    let server_task = pin!(server.with_graceful_shutdown(stop_rx.notified()));
+
+    let server =
+        axum::serve(listener, app.into_make_service()).with_graceful_shutdown(async move {
+            let _r = stop_rx.await;
+        });
+    let server_task = pin!(server.into_future());
     let shutdown_notified = pin!(async move {
         let _ = tokio::signal::ctrl_c().await;
     });
     match select(server_task, shutdown_notified).await {
         Either::Left((result, _)) => result?,
         Either::Right((_, server)) => {
-            stop_tx.notify_one();
+            assert!(stop_tx.trigger());
             tokio::time::timeout(shutdown_timeout, server).await??;
         }
     }

example_apps/transit/src/main.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::future::IntoFuture;
 use std::{error::Error, net::SocketAddr, pin::pin, sync::Arc, time::Duration};
 
 use clap::Parser;
@@ -22,6 +23,8 @@ use swimos::{
     route::RouteUri,
     server::{Server, ServerBuilder},
 };
+use swimos_utilities::trigger::trigger;
+use tokio::net::TcpListener;
 use tokio::sync::{oneshot, Notify};
 use tracing_subscriber::filter::LevelFilter;
 use transit::start_agencies_and_wait;
@@ -74,20 +77,24 @@ async fn ui_server(
 ) -> Result<(), Box<dyn Error + Send + Sync>> {
     if let Ok(addr) = swim_addr_rx.await {
         let app = ui_server_router(addr.port());
+        let bind_to: SocketAddr = format!("0.0.0.0:{}", port.unwrap_or_default()).parse()?;
+        let (stop_tx, stop_rx) = trigger();
 
-        let bind_to = format!("0.0.0.0:{}", port.unwrap_or_default()).parse()?;
-
-        let server = axum::Server::try_bind(&bind_to)?.serve(app.into_make_service());
-        let ui_addr = server.local_addr();
+        let listener = TcpListener::bind(bind_to).await?;
+        let ui_addr = listener.local_addr()?;
         println!("UI bound to: {}", ui_addr);
-        let stop_tx = Arc::new(Notify::new());
-        let stop_rx = stop_tx.clone();
-        let server_task = pin!(server.with_graceful_shutdown(stop_rx.notified()));
+
+        let server =
+            axum::serve(listener, app.into_make_service()).with_graceful_shutdown(async move {
+                let _ = stop_rx.await;
+            });
+
+        let server_task = pin!(server.into_future());
         let shutdown_notified = pin!(shutdown_signal.notified());
         match select(server_task, shutdown_notified).await {
             Either::Left((result, _)) => result?,
             Either::Right((_, server)) => {
-                stop_tx.notify_one();
+                assert!(stop_tx.trigger());
                 tokio::time::timeout(shutdown_timeout, server).await??;
             }
         }

example_apps/transit/src/ui.rs

@@ -12,11 +12,9 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use axum::body::Body;
 use axum::http::{header, HeaderValue};
-use axum::{
-    body::StreamBody, extract::State, http::StatusCode, response::IntoResponse, routing::get,
-    Router,
-};
+use axum::{extract::State, http::StatusCode, response::IntoResponse, routing::get, Router};
 use futures::{TryFutureExt, TryStreamExt};
 use tokio::fs::File;
 use tokio::io::{AsyncBufReadExt, BufReader};
@@ -102,7 +100,7 @@ async fn load_file(path: &str) -> impl IntoResponse {
         .await
     {
         let headers = [(header::CONTENT_LENGTH, HeaderValue::from(len))];
-        Ok((headers, StreamBody::new(ReaderStream::new(file))))
+        Ok((headers, Body::from_stream(ReaderStream::new(file))))
     } else {
         Err((StatusCode::NOT_FOUND, format!("File not found: {}", target)))
     }