diff --git a/crates/server/src/http.rs b/crates/server/src/http.rs index 3fd49a0..88be626 100644 --- a/crates/server/src/http.rs +++ b/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 http_body_util::Full; -use hyper::{Request, Response, server::conn::http2, service::service_fn}; -use hyper_util::rt::{TokioExecutor, TokioIo}; -use serde::Serialize; +use axum::{ + Json, Router, + body::Bytes, + extract::{Form, Query, State}, + http::StatusCode, + response::{IntoResponse, Response}, + routing::{get, post}, +}; +use serde::{Deserialize, Serialize}; use tokio::{ net::TcpListener, - sync::{ - Mutex, - mpsc::{Receiver, Sender}, - }, + sync::{Mutex, mpsc::Sender}, }; use crate::AppState; pub struct HttpServer { - pub offer_tx: Sender<(String, String)>, - pub accept_rx: Receiver<(String, String)>, + pub offer_tx: Sender<(i32, String, String)>, + pub accept_rx: async_broadcast::Receiver<(i32, String)>, pub appstate: Arc>, + pub request_count: Mutex, } impl HttpServer { - pub fn start(&mut self) -> Result<(), Box> { - let state = self.appstate.clone(); + pub fn start(self) -> Result<(), Box> { + let state = Arc::new(self); + tokio::spawn(async move { - let socket = SocketAddr::from(([127, 0, 0, 1], 3000)); - let listener = TcpListener::bind(socket).await.unwrap(); - loop { - let (stream, _) = listener.accept().await.unwrap(); - let tokio_io = TokioIo::new(stream); - let state = state.clone(); - tokio::spawn(async move { - let service = service_fn(move |req| { - let state = state.clone(); - async move { handle_request(req, &state).await } - }); - let hyper = http2::Builder::new(TokioExecutor::new()); - hyper.serve_connection(tokio_io, service).await.unwrap(); - }); - } + let app = Router::new() + .route("/api/catalog", get(catalog_handler)) + .route("/api/user", post(create_user_handler)) + .route("/api/stream", get(stream_handler)) + .route("/api/meow", get(meow_handler)) + .with_state(state); + + let addr = SocketAddr::from(([127, 0, 0, 1], 3000)); + let listener = TcpListener::bind(addr).await.unwrap(); + axum::serve(listener, app).await.unwrap(); }); Ok(()) } } -async fn handle_request( - request: Request, - appstate: &Arc>, -) -> Result>, Infallible> { - match request.uri().path() { - "/api/catalog" => { - let catalog = catalog_from_state(appstate).await; - let body = serde_json::to_vec(&catalog).unwrap(); - Ok(Response::builder() - .status(200) - .header("content-type", "application/json") - .body(Full::new(Bytes::from(body))) - .unwrap()) +fn calculate_hash(t: &T) -> u64 { + let mut s = DefaultHasher::new(); + t.hash(&mut s); + s.finish() +} + +async fn catalog_handler(State(state): State>) -> impl IntoResponse { + let catalog = catalog_from_state(&state.appstate).await; + Json(catalog) +} + +#[derive(Deserialize)] +struct CreateUserForm { + username: String, + password: String, + ref_token: String, +} + +async fn create_user_handler(Form(form): Form) -> 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>, + Query(query): Query, + 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)]