caffeinatedgaze
/
lavina
forked from lavina/lavina
1
0
Fork 0
Code Pull Requests Activity

use borrows in http endpoint handlers

This commit is contained in:
Nikita Vilunov 2024-05-04 01:07:23 +02:00
parent 8ac64ba8f5
commit bb0fe3bf0b
2 changed files with 15 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
crates/lavina-core/src/repo/mod.rs
View File

@ -143,7 +143,7 @@ impl Storage {
} }
#[tracing::instrument(skip(self), name = "Storage::create_user")] #[tracing::instrument(skip(self), name = "Storage::create_user")]
pub async fn create_user(&mut self, name: &str) -> Result<()> { pub async fn create_user(&self, name: &str) -> Result<()> {
let query = sqlx::query( let query = sqlx::query(
"insert into users(name) "insert into users(name)
values (?);", values (?);",

30
src/http.rs
View File

@ -61,10 +61,8 @@ async fn main_loop(
let core = core.clone(); let core = core.clone();
let storage = storage.clone(); let storage = storage.clone();
tokio::task::spawn(async move { tokio::task::spawn(async move {
let registry = metrics.clone(); let svc_fn = service_fn(|r| route(&metrics, &core, &storage, r));
let core = core.clone(); let server = http1::Builder::new().serve_connection(stream, svc_fn);
let storage = storage.clone();
let server = http1::Builder::new().serve_connection(stream, service_fn(move |r| route(registry.clone(), core.clone(), storage.clone(), r)));
if let Err(err) = server.await { if let Err(err) = server.await {
tracing::error!("Error serving connection: {:?}", err); tracing::error!("Error serving connection: {:?}", err);
} }
@ -77,16 +75,16 @@ async fn main_loop(
} }
async fn route( async fn route(
registry: MetricsRegistry, registry: &MetricsRegistry,
core: LavinaCore, core: &LavinaCore,
storage: Storage, storage: &Storage,
request: Request<hyper::body::Incoming>, request: Request<hyper::body::Incoming>,
) -> HttpResult<Response<Full<Bytes>>> { ) -> HttpResult<Response<Full<Bytes>>> {
let res = match (request.method(), request.uri().path()) { let res = match (request.method(), request.uri().path()) {
(&Method::GET, "/metrics") => endpoint_metrics(registry), (&Method::GET, "/metrics") => endpoint_metrics(registry),
(&Method::GET, "/rooms") => endpoint_rooms(core.rooms).await, (&Method::GET, "/rooms") => endpoint_rooms(&core.rooms).await,
(&Method::POST, paths::CREATE_PLAYER) => endpoint_create_player(request, storage).await.or5xx(), (&Method::POST, paths::CREATE_PLAYER) => endpoint_create_player(request, storage).await.or5xx(),
(&Method::POST, paths::STOP_PLAYER) => endpoint_stop_player(request, core.players).await.or5xx(), (&Method::POST, paths::STOP_PLAYER) => endpoint_stop_player(request, &core.players).await.or5xx(),
(&Method::POST, paths::SET_PASSWORD) => endpoint_set_password(request, core).await.or5xx(), (&Method::POST, paths::SET_PASSWORD) => endpoint_set_password(request, core).await.or5xx(),
(&Method::POST, rooms::paths::SEND_MESSAGE) => endpoint_send_room_message(request, core).await.or5xx(), (&Method::POST, rooms::paths::SEND_MESSAGE) => endpoint_send_room_message(request, core).await.or5xx(),
(&Method::POST, rooms::paths::SET_TOPIC) => endpoint_set_room_topic(request, core).await.or5xx(), (&Method::POST, rooms::paths::SET_TOPIC) => endpoint_set_room_topic(request, core).await.or5xx(),
@ -95,7 +93,7 @@ async fn route(
Ok(res) Ok(res)
} }
fn endpoint_metrics(registry: MetricsRegistry) -> Response<Full<Bytes>> { fn endpoint_metrics(registry: &MetricsRegistry) -> Response<Full<Bytes>> {
let mf = registry.gather(); let mf = registry.gather();
let mut buffer = vec![]; let mut buffer = vec![];
TextEncoder.encode(&mf, &mut buffer).expect("write to vec cannot fail"); TextEncoder.encode(&mf, &mut buffer).expect("write to vec cannot fail");
@ -103,7 +101,7 @@ fn endpoint_metrics(registry: MetricsRegistry) -> Response<Full<Bytes>> {
} }
#[tracing::instrument(skip_all)] #[tracing::instrument(skip_all)]
async fn endpoint_rooms(rooms: RoomRegistry) -> Response<Full<Bytes>> { async fn endpoint_rooms(rooms: &RoomRegistry) -> Response<Full<Bytes>> {
// TODO introduce management API types independent from core-domain types // TODO introduce management API types independent from core-domain types
// TODO remove `Serialize` implementations from all core-domain types // TODO remove `Serialize` implementations from all core-domain types
let room_list = rooms.get_all_rooms().await.to_body(); let room_list = rooms.get_all_rooms().await.to_body();
@ -113,7 +111,7 @@ async fn endpoint_rooms(rooms: RoomRegistry) -> Response<Full<Bytes>> {
#[tracing::instrument(skip_all)] #[tracing::instrument(skip_all)]
async fn endpoint_create_player( async fn endpoint_create_player(
request: Request<hyper::body::Incoming>, request: Request<hyper::body::Incoming>,
mut storage: Storage, storage: &Storage,
) -> Result<Response<Full<Bytes>>> { ) -> Result<Response<Full<Bytes>>> {
let str = request.collect().await?.to_bytes(); let str = request.collect().await?.to_bytes();
let Ok(res) = serde_json::from_slice::<CreatePlayerRequest>(&str[..]) else { let Ok(res) = serde_json::from_slice::<CreatePlayerRequest>(&str[..]) else {
@ -129,7 +127,7 @@ async fn endpoint_create_player(
#[tracing::instrument(skip_all)] #[tracing::instrument(skip_all)]
async fn endpoint_stop_player( async fn endpoint_stop_player(
request: Request<hyper::body::Incoming>, request: Request<hyper::body::Incoming>,
players: PlayerRegistry, players: &PlayerRegistry,
) -> Result<Response<Full<Bytes>>> { ) -> Result<Response<Full<Bytes>>> {
let str = request.collect().await?.to_bytes(); let str = request.collect().await?.to_bytes();
let Ok(res) = serde_json::from_slice::<StopPlayerRequest>(&str[..]) else { let Ok(res) = serde_json::from_slice::<StopPlayerRequest>(&str[..]) else {
@ -147,7 +145,7 @@ async fn endpoint_stop_player(
#[tracing::instrument(skip_all)] #[tracing::instrument(skip_all)]
async fn endpoint_set_password( async fn endpoint_set_password(
request: Request<hyper::body::Incoming>, request: Request<hyper::body::Incoming>,
core: LavinaCore, core: &LavinaCore,
) -> Result<Response<Full<Bytes>>> { ) -> Result<Response<Full<Bytes>>> {
let str = request.collect().await?.to_bytes(); let str = request.collect().await?.to_bytes();
let Ok(res) = serde_json::from_slice::<ChangePasswordRequest>(&str[..]) else { let Ok(res) = serde_json::from_slice::<ChangePasswordRequest>(&str[..]) else {
@ -165,7 +163,7 @@ async fn endpoint_set_password(
async fn endpoint_send_room_message( async fn endpoint_send_room_message(
request: Request<hyper::body::Incoming>, request: Request<hyper::body::Incoming>,
core: LavinaCore, core: &LavinaCore,
) -> Result<Response<Full<Bytes>>> { ) -> Result<Response<Full<Bytes>>> {
let str = request.collect().await?.to_bytes(); let str = request.collect().await?.to_bytes();
let Ok(req) = serde_json::from_slice::<rooms::SendMessageReq>(&str[..]) else { let Ok(req) = serde_json::from_slice::<rooms::SendMessageReq>(&str[..]) else {
@ -187,7 +185,7 @@ async fn endpoint_send_room_message(
async fn endpoint_set_room_topic( async fn endpoint_set_room_topic(
request: Request<hyper::body::Incoming>, request: Request<hyper::body::Incoming>,
core: LavinaCore, core: &LavinaCore,
) -> Result<Response<Full<Bytes>>> { ) -> Result<Response<Full<Bytes>>> {
let str = request.collect().await?.to_bytes(); let str = request.collect().await?.to_bytes();
let Ok(req) = serde_json::from_slice::<rooms::SetTopicReq>(&str[..]) else { let Ok(req) = serde_json::from_slice::<rooms::SetTopicReq>(&str[..]) else {