Convert to actix

Cargo.toml

@@ -9,13 +9,17 @@ repository = "https://github.com/vijfhoek/empede"
 [dependencies]
 anyhow = "1.0.70"
 askama = { version = "0.12.0", default-features = false, features = ["serde-json"] }
-askama_tide = "0.15.0"
-async-std = { version = "1.12.0", features = ["attributes"] }
 infer = { version = "0.15.0", default-features = false }
 percent-encoding = "2.2.0"
 serde = { version = "1.0.160", features = ["derive"] }
 serde_qs = "0.12.0"
-tide = "0.16.0"
-tide-tracing = "0.0.12"
-tracing = { version = "0.1.37", default-features = false, features = ["std"] }
-tracing-subscriber = { version = "0.3.17", default-features = false, features = ["std", "fmt"] }
+askama_actix = "0.14.0"
+tokio = { version = "1.35.1", features = ["full"] }
+actix-web = "4.4.0"
+thiserror = "1.0.51"
+actix-files = "0.6.2"
+actix-web-lab = "0.20.1"
+tokio-stream = "0.1.14"
+futures = "0.3.29"
+async-stream = "0.3.5"
+env_logger = "0.10.1"

src/main.rs

@@ -1,43 +1,43 @@
+use actix_web::{middleware::Logger, web, App, HttpServer};
+
 mod crate_version;
 mod mpd;
 mod routes;
 
-#[async_std::main]
-async fn main() -> tide::Result<()> {
-    tracing_subscriber::fmt()
-        .with_max_level(tracing::Level::WARN)
-        .init();
+#[actix_web::main]
+async fn main() -> std::io::Result<()> {
+    let bind = std::env::var("EMPEDE_BIND").unwrap_or("0.0.0.0:8080".into());
+    let (host, port) = bind.split_once(':').unwrap();
 
-    let mut app = tide::new();
-    app.with(tide_tracing::TraceMiddleware::new());
+    env_logger::init_from_env(env_logger::Env::new().default_filter_or("info"));
 
-    app.at("/").get(routes::index::get_index);
-    app.at("/player").get(routes::player::get_player);
-    app.at("/browser").get(routes::browser::get_browser);
-    app.at("/art").get(routes::art::get_art);
-
-    app.at("/sse").get(tide::sse::endpoint(routes::sse::sse));
-
-    app.at("/queue").get(routes::queue::get_queue);
-    app.at("/queue").post(routes::queue::post_queue);
-    app.at("/queue").delete(routes::queue::delete_queue);
-    app.at("/queue/move").post(routes::queue::post_queue_move);
-
-    app.at("/play").post(routes::controls::post_play);
-    app.at("/pause").post(routes::controls::post_pause);
-    app.at("/previous").post(routes::controls::post_previous);
-    app.at("/next").post(routes::controls::post_next);
-
-    app.at("/consume").post(routes::controls::post_consume);
-    app.at("/random").post(routes::controls::post_random);
-    app.at("/repeat").post(routes::controls::post_repeat);
-    app.at("/single").post(routes::controls::post_single);
-    app.at("/shuffle").post(routes::controls::post_shuffle);
-
-    app.at("/static").serve_dir("static/")?;
-
-    let bind = std::env::var("EMPEDE_BIND").unwrap_or("0.0.0.0:8080".to_string());
-    app.listen(bind).await?;
+    HttpServer::new(|| {
+        App::new().wrap(Logger::default()).service(
+            web::scope("")
+                .service(routes::index::get_index)
+                .service(routes::player::get_player)
+                .service(routes::browser::get_browser)
+                .service(routes::art::get_art)
+                .service(routes::sse::idle)
+                .service(routes::queue::get_queue)
+                .service(routes::queue::post_queue)
+                .service(routes::queue::delete_queue)
+                .service(routes::queue::post_queue_move)
+                .service(routes::controls::post_play)
+                .service(routes::controls::post_pause)
+                .service(routes::controls::post_previous)
+                .service(routes::controls::post_next)
+                .service(routes::controls::post_consume)
+                .service(routes::controls::post_random)
+                .service(routes::controls::post_repeat)
+                .service(routes::controls::post_single)
+                .service(routes::controls::post_shuffle)
+                .service(actix_files::Files::new("/static", "./static")),
+        )
+    })
+    .bind((host, port.parse().unwrap()))?
+    .run()
+    .await?;
 
     Ok(())
 }

src/mpd.rs

@@ -1,11 +1,10 @@
-use std::{collections::HashMap, sync::OnceLock};
+use std::collections::HashMap;
 
 use anyhow::anyhow;
-use async_std::{
-    io::{prelude::BufReadExt, BufReader, ReadExt, WriteExt},
+use tokio::{
+    io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufStream},
     net::TcpStream,
-    sync::{Mutex, MutexGuard},
-    task::block_on,
+    sync::{Mutex, MutexGuard, OnceCell},
 };
 
 pub fn host() -> String {
@@ -42,19 +41,21 @@ pub enum Entry {
 
 #[derive(Debug)]
 pub struct Mpd {
-    stream: Option<TcpStream>,
-    reader: Option<BufReader<TcpStream>>,
+    bufstream: Option<BufStream<TcpStream>>,
 }
 
-pub static INSTANCE: OnceLock<Mutex<Mpd>> = OnceLock::new();
+pub static INSTANCE: OnceCell<Mutex<Mpd>> = OnceCell::const_new();
 
 pub async fn get_instance() -> MutexGuard<'static, Mpd> {
-    let instance = INSTANCE.get_or_init(|| {
-        let mut mpd = Mpd::new();
-        block_on(mpd.connect()).unwrap();
-        Mutex::from(mpd)
-    });
-    instance.lock().await
+    INSTANCE
+        .get_or_init(|| async {
+            let mut mpd = Mpd::new();
+            mpd.connect().await.unwrap();
+            Mutex::from(mpd)
+        })
+        .await
+        .lock()
+        .await
 }
 
 pub async fn command(command: &str) -> anyhow::Result<CommandResult> {
@@ -116,45 +117,42 @@ impl Mpd {
     }
 
     pub fn new() -> Self {
-        Self {
-            stream: None,
-            reader: None,
-        }
+        Self { bufstream: None }
     }
 
     pub async fn connect(&mut self) -> anyhow::Result<()> {
-        self.stream = Some(TcpStream::connect(host()).await?);
-        self.reader = Some(BufReader::new(self.stream.as_mut().unwrap().clone()));
+        let stream = TcpStream::connect(host()).await?;
+        let mut bufstream = BufStream::new(stream);
 
         // skip OK MPD line
         // TODO check if it is indeed OK
         let mut buffer = String::new();
-        self.reader.as_mut().unwrap().read_line(&mut buffer).await?;
+        bufstream.read_line(&mut buffer).await?;
 
         let password = std::env::var("MPD_PASSWORD").unwrap_or_default();
         if !password.is_empty() {
             let password = Self::escape_str(&password);
-            self.stream
-                .as_mut()
-                .unwrap()
-                .write_all(format!(r#"password "{password}"\n"#).as_bytes())
+            bufstream
+                .write_all(format!("password \"{password}\"\n").as_bytes())
                 .await?;
-            self.reader.as_mut().unwrap().read_line(&mut buffer).await?;
+            bufstream.flush().await?;
+            bufstream.read_line(&mut buffer).await?;
         }
 
-        self.stream
-            .as_mut()
-            .unwrap()
+        bufstream
             .write_all("binarylimit 1048576\n".as_bytes())
             .await?;
-        self.reader.as_mut().unwrap().read_line(&mut buffer).await?;
+        bufstream.flush().await?;
+        bufstream.read_line(&mut buffer).await?;
+
+        self.bufstream = Some(bufstream);
 
         Ok(())
     }
 
     async fn read_binary_data(&mut self, size: usize) -> anyhow::Result<Vec<u8>> {
         let mut binary = vec![0u8; size];
-        self.reader
+        self.bufstream
             .as_mut()
             .unwrap()
             .read_exact(&mut binary)
@@ -163,11 +161,19 @@ impl Mpd {
         let mut buffer = String::new();
 
         // Skip the newline after the binary data
-        self.reader.as_mut().unwrap().read_line(&mut buffer).await?;
+        self.bufstream
+            .as_mut()
+            .unwrap()
+            .read_line(&mut buffer)
+            .await?;
 
         // Skip the "OK" after the binary data
         // TODO Check if actually OK
-        self.reader.as_mut().unwrap().read_line(&mut buffer).await?;
+        self.bufstream
+            .as_mut()
+            .unwrap()
+            .read_line(&mut buffer)
+            .await?;
 
         Ok(binary)
     }
@@ -176,16 +182,21 @@ impl Mpd {
         let mut properties = Vec::new();
 
         'retry: loop {
-            self.stream
+            self.bufstream
                 .as_mut()
                 .unwrap()
                 .write_all(format!("{command}\n").as_bytes())
                 .await?;
+            self.bufstream.as_mut().unwrap().flush().await?;
 
             let mut buffer = String::new();
             break 'retry (loop {
                 buffer.clear();
-                self.reader.as_mut().unwrap().read_line(&mut buffer).await?;
+                self.bufstream
+                    .as_mut()
+                    .unwrap()
+                    .read_line(&mut buffer)
+                    .await?;
 
                 if let Some((key, value)) = buffer.split_once(": ") {
                     let value = value.trim_end();

src/routes/art.rs

@@ -1,4 +1,9 @@
 use crate::mpd;
+use actix_web::{
+    get,
+    http::header::{self, CacheDirective},
+    web, HttpResponse, Responder,
+};
 use percent_encoding::percent_decode_str;
 use serde::Deserialize;
 
@@ -8,33 +13,30 @@ struct ArtQuery {
     path: String,
 }
 
-pub async fn get_art(req: tide::Request<()>) -> tide::Result {
-    let query: ArtQuery = req.query()?;
+#[get("/art")]
+pub async fn get_art(query: web::Query<ArtQuery>) -> impl Responder {
     let path = percent_decode_str(&query.path).decode_utf8_lossy();
-
     let mut mpd = mpd::get_instance().await;
 
-    let resp = if let Ok(art) = mpd.albumart(&path).await {
+    if let Ok(art) = mpd.albumart(&path).await {
         let mime = infer::get(&art)
             .map(|k| k.mime_type())
             .unwrap_or("application/octet-stream");
 
-        tide::Response::builder(tide::StatusCode::Ok)
-            .body(art)
+        HttpResponse::Ok()
             .content_type(mime)
-            .header("cache-control", "max-age=3600")
+            .append_header(header::CacheControl(vec![CacheDirective::MaxAge(3600)]))
+            .body(art)
     } else if let Ok(art) = mpd.readpicture(&path).await {
         let mime = infer::get(&art)
             .map(|k| k.mime_type())
             .unwrap_or("application/octet-stream");
 
-        tide::Response::builder(tide::StatusCode::Ok)
-            .body(art)
+        HttpResponse::Ok()
             .content_type(mime)
-            .header("cache-control", "max-age=3600")
+            .append_header(header::CacheControl(vec![CacheDirective::MaxAge(3600)]))
+            .body(art)
     } else {
-        tide::Response::builder(tide::StatusCode::NotFound)
-    };
-
-    Ok(resp.into())
+        HttpResponse::NotFound().finish()
+    }
 }

src/routes/browser.rs

@@ -1,4 +1,5 @@
 use crate::mpd;
+use actix_web::{get, web, Responder};
 use askama::Template;
 use percent_encoding::percent_decode_str;
 use serde::Deserialize;
@@ -17,19 +18,17 @@ struct BrowserQuery {
     path: String,
 }
 
-pub async fn get_browser(req: tide::Request<()>) -> tide::Result {
-    let query: BrowserQuery = req.query()?;
+#[get("/browser")]
+pub async fn get_browser(query: web::Query<BrowserQuery>) -> impl Responder {
     let path = percent_decode_str(&query.path).decode_utf8_lossy();
     let mut mpd = mpd::get_instance().await;
-    let entries = mpd.ls(&path).await?;
+    let entries = mpd.ls(&path).await.unwrap();
 
-    let template = BrowserTemplate {
+    BrowserTemplate {
         path: Path::new(&*path)
             .iter()
             .map(|s| s.to_string_lossy().to_string())
             .collect(),
         entries,
-    };
-
-    Ok(template.into())
+    }
 }

src/routes/controls.rs

@@ -1,3 +1,5 @@
+use actix_web::{post, HttpResponse, Responder};
+
 use crate::mpd;
 
 async fn toggle_setting(setting: &str) -> anyhow::Result<()> {
@@ -11,47 +13,56 @@ async fn toggle_setting(setting: &str) -> anyhow::Result<()> {
     Ok(())
 }
 
-pub async fn post_play(_req: tide::Request<()>) -> tide::Result {
-    mpd::command("play").await?;
-    Ok("".into())
+#[post("/play")]
+pub async fn post_play() -> impl Responder {
+    mpd::command("play").await.unwrap();
+    HttpResponse::NoContent()
 }
 
-pub async fn post_pause(_req: tide::Request<()>) -> tide::Result {
-    mpd::command("pause 1").await?;
-    Ok("".into())
+#[post("/pause")]
+pub async fn post_pause() -> impl Responder {
+    mpd::command("pause 1").await.unwrap();
+    HttpResponse::NoContent()
 }
 
-pub async fn post_previous(_req: tide::Request<()>) -> tide::Result {
-    mpd::command("previous").await?;
-    Ok("".into())
+#[post("/previous")]
+pub async fn post_previous() -> impl Responder {
+    mpd::command("previous").await.unwrap();
+    HttpResponse::NoContent()
 }
 
-pub async fn post_next(_req: tide::Request<()>) -> tide::Result {
-    mpd::command("next").await?;
-    Ok("".into())
+#[post("/next")]
+pub async fn post_next() -> impl Responder {
+    mpd::command("next").await.unwrap();
+    HttpResponse::NoContent()
 }
 
-pub async fn post_consume(_req: tide::Request<()>) -> tide::Result {
-    toggle_setting("consume").await?;
-    Ok("".into())
+#[post("/consume")]
+pub async fn post_consume() -> impl Responder {
+    toggle_setting("consume").await.unwrap();
+    HttpResponse::NoContent()
 }
 
-pub async fn post_random(_req: tide::Request<()>) -> tide::Result {
-    toggle_setting("random").await?;
-    Ok("".into())
+#[post("/random")]
+pub async fn post_random() -> impl Responder {
+    toggle_setting("random").await.unwrap();
+    HttpResponse::NoContent()
 }
 
-pub async fn post_repeat(_req: tide::Request<()>) -> tide::Result {
-    toggle_setting("repeat").await?;
-    Ok("".into())
+#[post("/repeat")]
+pub async fn post_repeat() -> impl Responder {
+    toggle_setting("repeat").await.unwrap();
+    HttpResponse::NoContent()
 }
 
-pub async fn post_shuffle(_req: tide::Request<()>) -> tide::Result {
-    mpd::command("shuffle").await?;
-    Ok("".into())
+#[post("/shuffle")]
+pub async fn post_shuffle() -> impl Responder {
+    mpd::command("shuffle").await.unwrap();
+    HttpResponse::NoContent()
 }
 
-pub async fn post_single(_req: tide::Request<()>) -> tide::Result {
-    toggle_setting("single").await?;
-    Ok("".into())
-}
+#[post("/single")]
+pub async fn post_single() -> impl Responder {
+    toggle_setting("single").await.unwrap();
+    HttpResponse::NoContent()
+}

src/routes/index.rs

@@ -1,4 +1,5 @@
 use crate::crate_version;
+use actix_web::{get, Responder};
 use askama::Template;
 use serde::Deserialize;
 
@@ -12,6 +13,7 @@ struct IndexQuery {
     path: String,
 }
 
-pub async fn get_index(_req: tide::Request<()>) -> tide::Result {
-    Ok(askama_tide::into_response(&IndexTemplate))
+#[get("/")]
+pub async fn get_index() -> impl Responder {
+    IndexTemplate
 }

src/routes/mod.rs

@@ -1,7 +1,7 @@
 pub mod art;
 pub mod browser;
+pub mod controls;
 pub mod index;
 pub mod player;
 pub mod queue;
-pub mod controls;
 pub mod sse;

src/routes/player.rs

@@ -1,13 +1,14 @@
 use crate::mpd;
+use actix_web::{get, Responder};
 use askama::Template;
 use std::collections::HashMap;
 
 #[derive(Template)]
 #[template(path = "player.html")]
-struct PlayerTemplate<'a> {
-    song: Option<&'a HashMap<String, String>>,
+struct PlayerTemplate {
+    song: Option<HashMap<String, String>>,
     name: Option<String>,
-    state: &'a str,
+    state: String,
     consume: bool,
     random: bool,
     repeat: bool,
@@ -16,10 +17,11 @@ struct PlayerTemplate<'a> {
     duration: f32,
 }
 
-pub async fn get_player(_req: tide::Request<()>) -> tide::Result {
+#[get("/player")]
+pub async fn get_player() -> impl Responder {
     let mut mpd = mpd::get_instance().await;
-    let song = mpd.command("currentsong").await?.into_hashmap();
-    let status = mpd.command("status").await?.into_hashmap();
+    let song = mpd.command("currentsong").await.unwrap().into_hashmap();
+    let status = mpd.command("status").await.unwrap().into_hashmap();
 
     let elapsed = status
         .get("elapsed")
@@ -31,9 +33,13 @@ pub async fn get_player(_req: tide::Request<()>) -> tide::Result {
         .unwrap_or(1.0);
 
     let mut template = PlayerTemplate {
-        song: if song.is_empty() { None } else { Some(&song) },
+        song: if song.is_empty() {
+            None
+        } else {
+            Some(song.clone())
+        },
         name: None,
-        state: &status["state"],
+        state: status["state"].clone(),
         consume: status["consume"] == "1",
         random: status["random"] == "1",
         repeat: status["repeat"] == "1",
@@ -47,5 +53,5 @@ pub async fn get_player(_req: tide::Request<()>) -> tide::Result {
         template.name = Some(name);
     }
 
-    Ok(template.into())
+    template
 }

src/routes/queue.rs

@@ -1,4 +1,5 @@
 use crate::mpd;
+use actix_web::{delete, get, post, web, HttpResponse, Responder};
 use askama::Template;
 use percent_encoding::percent_decode_str;
 use serde::Deserialize;
@@ -9,11 +10,11 @@ struct QueueTemplate {
     queue: Vec<mpd::QueueItem>,
 }
 
-pub async fn get_queue(_req: tide::Request<()>) -> tide::Result {
+#[get("/queue")]
+pub async fn get_queue() -> impl Responder {
     let mut mpd = mpd::get_instance().await;
-    let queue = mpd.playlist().await?;
-    let template = QueueTemplate { queue };
-    Ok(template.into())
+    let queue = mpd.playlist().await.unwrap();
+    QueueTemplate { queue }
 }
 
 #[derive(Deserialize)]
@@ -27,26 +28,26 @@ struct PostQueueQuery {
     play: bool,
 }
 
-pub async fn post_queue(req: tide::Request<()>) -> tide::Result {
-    let query: PostQueueQuery = req.query()?;
+#[post("/queue")]
+pub async fn post_queue(query: web::Query<PostQueueQuery>) -> impl Responder {
     let path = percent_decode_str(&query.path).decode_utf8_lossy();
     let mut mpd = mpd::get_instance().await;
 
     if query.replace {
-        mpd.clear().await?;
+        mpd.clear().await.unwrap();
     }
 
     if query.next {
-        mpd.add_pos(&path, "+0").await?;
+        mpd.add_pos(&path, "+0").await.unwrap();
     } else {
-        mpd.add(&path).await?;
+        mpd.add(&path).await.unwrap();
     }
 
     if query.play {
-        mpd.play().await?;
+        mpd.play().await.unwrap();
     }
 
-    Ok("".into())
+    HttpResponse::NoContent()
 }
 
 #[derive(Deserialize)]
@@ -55,17 +56,16 @@ struct DeleteQueueQuery {
     id: Option<u32>,
 }
 
-pub async fn delete_queue(req: tide::Request<()>) -> tide::Result {
-    let query: DeleteQueueQuery = req.query()?;
-
+#[delete("/queue")]
+pub async fn delete_queue(query: web::Query<DeleteQueueQuery>) -> impl Responder {
     let mut mpd = mpd::get_instance().await;
     if let Some(id) = query.id {
-        mpd.command(&format!("deleteid {id}")).await?;
+        mpd.command(&format!("deleteid {id}")).await.unwrap();
     } else {
-        mpd.command("clear").await?;
+        mpd.command("clear").await.unwrap();
     }
 
-    Ok("".into())
+    HttpResponse::NoContent()
 }
 
 #[derive(Deserialize, Debug)]
@@ -74,10 +74,11 @@ struct UpdateQueueBody {
     to: u32,
 }
 
-pub async fn post_queue_move(mut req: tide::Request<()>) -> tide::Result {
-    let body: UpdateQueueBody = req.body_json().await?;
+#[post("/queue/move")]
+pub async fn post_queue_move(body: web::Json<UpdateQueueBody>) -> impl Responder {
     let mut mpd = mpd::get_instance().await;
     mpd.command(&format!("move {} {}", body.from, body.to))
-        .await?;
-    Ok("".into())
+        .await
+        .unwrap();
+    HttpResponse::NoContent()
 }

src/routes/sse.rs

@@ -1,19 +1,33 @@
-use crate::mpd;
+use std::time::Duration;
 
-pub async fn sse(_req: tide::Request<()>, sender: tide::sse::Sender) -> tide::Result<()> {
-    // Update everything on connect
-    sender.send("playlist", "", None).await?;
-    sender.send("player", "", None).await?;
+use actix_web::{get, Responder};
+use actix_web_lab::sse;
 
-    let mut mpd = mpd::Mpd::new();
+use crate::mpd::Mpd;
+
+#[get("/idle")]
+pub async fn idle() -> impl Responder {
+    let mut mpd = Mpd::new();
     mpd.connect().await.unwrap();
 
-    loop {
-        let systems = mpd
-            .idle(&["playlist", "player", "database", "options"])
-            .await?;
-        for system in systems {
-            sender.send(&system, "", None).await?;
-        }
+    const SYSTEMS: &[&str] = &["playlist", "player", "database", "options"];
+
+    let (tx, rx) = tokio::sync::mpsc::channel(10);
+    for system in SYSTEMS {
+        _ = tx
+            .send(sse::Data::new("").event(system.to_owned()).into())
+            .await;
     }
-}
+
+    actix_web::rt::spawn(async move {
+        loop {
+            let systems = mpd.idle(SYSTEMS).await.unwrap();
+
+            for system in systems {
+                _ = tx.send(sse::Data::new("").event(system).into()).await;
+            }
+        }
+    });
+
+    sse::Sse::from_infallible_receiver(rx).with_retry_duration(Duration::from_secs(10))
+}

templates/index.html

@@ -24,7 +24,7 @@
     </script>
   </head>
 
-  <body hx-ext="sse" sse-connect="/sse">
+  <body hx-ext="sse" sse-connect="/idle">
     <div 
       class="browser" 
       hx-trigger="load,sse:database"