http.rs/ hyper -> axum

crates/server/src/http.rs

@@ -1,73 +1,120 @@
-use std::{convert::Infallible, error::Error, net::SocketAddr, sync::Arc};
+use std::{
+    hash::{DefaultHasher, Hash, Hasher},
+    net::SocketAddr,
+    sync::Arc,
+};
 
-use bytes::Bytes;
+use axum::{
-use http_body_util::Full;
+    Json, Router,
-use hyper::{Request, Response, server::conn::http2, service::service_fn};
+    body::Bytes,
-use hyper_util::rt::{TokioExecutor, TokioIo};
+    extract::{Form, Query, State},
-use serde::Serialize;
+    http::StatusCode,
+    response::{IntoResponse, Response},
+    routing::{get, post},
+};
+use serde::{Deserialize, Serialize};
 use tokio::{
     net::TcpListener,
-    sync::{
+    sync::{Mutex, mpsc::Sender},
-        Mutex,
-        mpsc::{Receiver, Sender},
-    },
 };
 
 use crate::AppState;
 
 pub struct HttpServer {
-    pub offer_tx: Sender<(String, String)>,
+    pub offer_tx: Sender<(i32, String, String)>,
-    pub accept_rx: Receiver<(String, String)>,
+    pub accept_rx: async_broadcast::Receiver<(i32, String)>,
     pub appstate: Arc<Mutex<AppState>>,
+    pub request_count: Mutex<i32>,
 }
 
 impl HttpServer {
-    pub fn start(&mut self) -> Result<(), Box<dyn Error>> {
+    pub fn start(self) -> Result<(), Box<dyn std::error::Error>> {
-        let state = self.appstate.clone();
+        let state = Arc::new(self);
+
         tokio::spawn(async move {
-            let socket = SocketAddr::from(([127, 0, 0, 1], 3000));
+            let app = Router::new()
-            let listener = TcpListener::bind(socket).await.unwrap();
+                .route("/api/catalog", get(catalog_handler))
-            loop {
+                .route("/api/user", post(create_user_handler))
-                let (stream, _) = listener.accept().await.unwrap();
+                .route("/api/stream", get(stream_handler))
-                let tokio_io = TokioIo::new(stream);
+                .route("/api/meow", get(meow_handler))
-                let state = state.clone();
+                .with_state(state);
-                tokio::spawn(async move {
+
-                    let service = service_fn(move |req| {
+            let addr = SocketAddr::from(([127, 0, 0, 1], 3000));
-                        let state = state.clone();
+            let listener = TcpListener::bind(addr).await.unwrap();
-                        async move { handle_request(req, &state).await }
+            axum::serve(listener, app).await.unwrap();
-                    });
-                    let hyper = http2::Builder::new(TokioExecutor::new());
-                    hyper.serve_connection(tokio_io, service).await.unwrap();
-                });
-            }
         });
         Ok(())
     }
 }
 
-async fn handle_request(
+fn calculate_hash<T: Hash>(t: &T) -> u64 {
-    request: Request<impl hyper::body::Body>,
+    let mut s = DefaultHasher::new();
-    appstate: &Arc<Mutex<AppState>>,
+    t.hash(&mut s);
-) -> Result<Response<Full<Bytes>>, Infallible> {
+    s.finish()
-    match request.uri().path() {
+}
-        "/api/catalog" => {
+
-            let catalog = catalog_from_state(appstate).await;
+async fn catalog_handler(State(state): State<Arc<HttpServer>>) -> impl IntoResponse {
-            let body = serde_json::to_vec(&catalog).unwrap();
+    let catalog = catalog_from_state(&state.appstate).await;
-            Ok(Response::builder()
+    Json(catalog)
-                .status(200)
+}
-                .header("content-type", "application/json")
+
-                .body(Full::new(Bytes::from(body)))
+#[derive(Deserialize)]
-                .unwrap())
+struct CreateUserForm {
+    username: String,
+    password: String,
+    ref_token: String,
+}
+
+async fn create_user_handler(Form(form): Form<CreateUserForm>) -> StatusCode {
+    // TODO: ref_token validation, db integration
+    let _ = (
+        form.username,
+        calculate_hash(&form.password),
+        form.ref_token,
+    );
+    todo!("user creation not yet implemented")
+}
+
+#[derive(Deserialize)]
+struct StreamQuery {
+    stream_label: String,
+}
+
+async fn stream_handler(
+    State(state): State<Arc<HttpServer>>,
+    Query(query): Query<StreamQuery>,
+    body: Bytes,
+) -> impl IntoResponse {
+    let body_str = String::from_utf8(body.to_vec()).unwrap_or_default();
+
+    let mut request_id = state.request_count.lock().await;
+    *request_id += 1;
+    let request_id_clone = *request_id;
+    drop(request_id);
+
+    let mut accept_rx = state.accept_rx.new_receiver();
+    let _ = state
+        .offer_tx
+        .send((request_id_clone, query.stream_label, body_str))
+        .await;
+
+    let mut reply_body = String::new();
+    while let Ok(answer) = accept_rx.recv().await {
+        if answer.0 == request_id_clone {
+            reply_body = answer.1;
+            break;
         }
-        "/api/meow" => Ok(Response::builder()
-            .status(200)
-            .body(Full::new(Bytes::from("meow")))
-            .unwrap()),
-        _ => Ok(Response::builder()
-            .status(404)
-            .body(Full::new(Bytes::from("")))
-            .unwrap()),
     }
+
+    Response::builder()
+        .status(StatusCode::CREATED)
+        .header("content-type", "application/sdp")
+        .body(reply_body)
+        .unwrap()
+}
+
+async fn meow_handler() -> &'static str {
+    "meow"
 }
 
 #[derive(Serialize)]