update to axum 0.7 (stable hyper)

Cargo.toml

@@ -14,7 +14,7 @@ edition = "2021"
 build = "build.rs"
 
 [dependencies]
-axum = { version = "0.6", features = ["macros", "multipart"] }
+axum = { version = "0.7", features = ["macros", "multipart"] }
 bytes = "1"
 clap = { version = "4", features = ["derive", "env", "deprecated"] }
 csv = "1"
@@ -41,7 +41,7 @@ time = "0.3"
 tokio = { version = "1", features = ["full"] }
 tokio-stream = { version = "0.1", features = ["io-util"] }
 tokio-util = { version = "0.7", features = ["io"] }
-tower-http = { version = "0.4", features = ["set-header"] }
+tower-http = { version = "0.5", features = ["set-header"] }
 
 [dev-dependencies]
 quickcheck = "1"

src/api/nd_json.rs

@@ -5,8 +5,7 @@ use std::{
 };
 
 use axum::{
-    body::{self, HttpBody},
-    http::HeaderMap,
+    body::Body,
     response::{IntoResponse, Response},
 };
 use bytes::Bytes;
@@ -33,37 +32,33 @@ where
         Response::builder()
             .header("X-Accel-Buffering", "no")
             .header(axum::http::header::CONTENT_TYPE, "application/x-ndjson")
-            .body(body::boxed(NdJsonBody {
-                stream: SyncWrapper::new(self.0),
+            .body(Body::from_stream(NdJsonStream {
+                item_stream: SyncWrapper::new(self.0),
                 keep_alive,
             }))
             .unwrap()
     }
 }
 
 pin_project! {
-    pub struct NdJsonBody<S> {
+    pub struct NdJsonStream<S> {
         #[pin]
-        stream: SyncWrapper<S>,
+        item_stream: SyncWrapper<S>,
         keep_alive: Interval,
     }
 }
 
-impl<S, T> HttpBody for NdJsonBody<S>
+impl<S, T> Stream for NdJsonStream<S>
 where
     S: Stream<Item = T>,
     T: Serialize,
 {
-    type Data = Bytes;
-    type Error = serde_json::Error;
+    type Item = Result<Bytes, serde_json::Error>;
 
-    fn poll_data(
-        self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
-    ) -> Poll<Option<Result<Self::Data, Self::Error>>> {
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         let this = self.project();
 
-        let without_keepalive = this.stream.get_pin_mut().poll_next(cx).map(|item| {
+        let without_keepalive = this.item_stream.get_pin_mut().poll_next(cx).map(|item| {
             item.map(|item| {
                 serde_json::to_vec(&item).map(|mut buf| {
                     buf.push(b'\n');
@@ -84,11 +79,4 @@ where
             Poll::Ready(end_or_err) => Poll::Ready(end_or_err),
         }
     }
-
-    fn poll_trailers(
-        self: Pin<&mut Self>,
-        _cx: &mut Context<'_>,
-    ) -> Poll<Result<Option<HeaderMap>, Self::Error>> {
-        Poll::Ready(Ok(None))
-    }
 }

src/indexer/mod.rs

@@ -4,10 +4,10 @@ use std::{
     time::{Duration, Instant},
 };
 
-use axum::http::StatusCode;
 use clap::Parser;
 use futures_util::StreamExt;
 use nohash_hasher::IntMap;
+use reqwest::StatusCode;
 use shakmaty::{
     uci::Uci,
     variant::VariantPosition,

src/main.rs

@@ -34,7 +34,7 @@ use shakmaty::{
     Color, EnPassantMode,
 };
 use tikv_jemallocator::Jemalloc;
-use tokio::sync::Semaphore;
+use tokio::{net::TcpListener, sync::Semaphore};
 
 use crate::{
     api::{
@@ -160,10 +160,8 @@ async fn serve() {
         app
     };
 
-    axum::Server::bind(&opt.bind)
-        .serve(app.into_make_service())
-        .await
-        .expect("bind");
+    let listener = TcpListener::bind(&opt.bind).await.expect("bind");
+    axum::serve(listener, app).await.expect("serve");
 }
 
 #[derive(Deserialize)]
@@ -368,7 +366,7 @@ fn finalize_lichess_games(
         .games(games.iter().map(|(_, id)| *id))
         .expect("get games")
         .into_iter()
-        .zip(games.into_iter())
+        .zip(games)
         .filter_map(|(info, (uci, id))| {
             info.map(|info| ExplorerGameWithUci {
                 uci,