Merge pull request 'Update to Actix 4' (#19) from feature/update-to-actix-4 into master

Reviewed-on: #19

Cargo.toml

@@ -10,11 +10,11 @@ edition = "2018"
 lto = true
 
 [dependencies]
-actix = "0.10"
-actix-web = "3"
-actix-rt = "1"
-actix-files = "0.5"
-actix-multipart = "0.3.0"
+actix = "0.12"
+actix-web = "4"
+actix-rt = "2.6"
+actix-files = "0.6"
+actix-multipart = "0.4.0"
 futures = "0.3.5"
 jsonwebtoken = "7.2.0"
 serde = "1"
@@ -32,7 +32,7 @@ notify = "4.0"
 path-absolutize = "3.0.6"
 log="0.4"
 env_logger="0.8"
-actix-web-prom = "0.5.1"
-prometheus = "0.11"
+actix-web-prom = "0.6"
+prometheus = "0.13"
 lazy_static = "1.1"
 anyhow = "1.0"

Jenkinsfile

@@ -1,7 +1,7 @@
 pipeline {
   agent {
     docker {
-      image 'rust:1.55'
+      image 'rust:1.59'
       args '-v "$PWD":/usr/src/image-api'
     }
   }

src/auth.rs

@@ -1,5 +1,8 @@
-use actix_web::web::{self, HttpResponse, Json};
 use actix_web::{post, Responder};
+use actix_web::{
+    web::{self, Json},
+    HttpResponse,
+};
 use chrono::{Duration, Utc};
 use jsonwebtoken::{encode, EncodingKey, Header};
 use log::{debug, error};
@@ -57,6 +60,7 @@ pub async fn login(
 
 #[cfg(test)]
 mod tests {
+
     use super::*;
     use crate::testhelpers::{BodyReader, TestUserDao};
 
@@ -88,7 +92,9 @@ mod tests {
         let response = login(j, web::Data::new(Box::new(dao))).await;
 
         assert_eq!(response.status(), 200);
-        assert!(response.body().read_to_str().contains("\"token\""));
+        let response_text: String = response.read_to_str();
+
+        assert!(response_text.contains("\"token\""));
     }
 
     #[actix_rt::test]

src/data/mod.rs

@@ -53,7 +53,6 @@ impl FromStr for Claims {
 impl FromRequest for Claims {
     type Error = Error;
     type Future = Ready<Result<Self, Self::Error>>;
-    type Config = ();
 
     fn from_request(req: &HttpRequest, _payload: &mut dev::Payload) -> Self::Future {
         req.headers()

src/files.rs

@@ -5,7 +5,7 @@ use std::path::{Path, PathBuf};
 use ::anyhow;
 use anyhow::{anyhow, Context};
 
-use actix_web::web::{HttpResponse, Query};
+use actix_web::{web::Query, HttpResponse};
 
 use log::{debug, error};
 
@@ -140,7 +140,7 @@ mod tests {
         use super::list_photos;
         use crate::{
             data::{Claims, PhotosResponse, ThumbnailRequest},
-            testhelpers::TypedBodyReader,
+            testhelpers::BodyReader,
         };
 
         use std::fs;
@@ -172,10 +172,11 @@ mod tests {
             fs::File::create(temp_photo).unwrap();
 
             let response: HttpResponse = list_photos(claims, request).await;
+            let status = response.status();
 
-            let body: PhotosResponse = response.body().read_body();
+            let body: PhotosResponse = serde_json::from_str(&response.read_to_str()).unwrap();
 
-            assert_eq!(response.status(), 200);
+            assert_eq!(status, 200);
             assert!(body.photos.contains(&String::from("photo.jpg")));
             assert!(body.dirs.contains(&String::from("test-dir")));
             assert!(body

src/main.rs

@@ -2,7 +2,8 @@
 extern crate diesel;
 extern crate rayon;
 
-use actix_web_prom::PrometheusMetrics;
+use actix_web::web::Data;
+use actix_web_prom::PrometheusMetricsBuilder;
 use futures::stream::StreamExt;
 use lazy_static::lazy_static;
 use prometheus::{self, IntGauge};
@@ -19,11 +20,9 @@ use actix::prelude::*;
 use actix_files::NamedFile;
 use actix_multipart as mp;
 use actix_web::{
-    delete,
-    error::BlockingError,
-    get, middleware, post, put,
-    web::{self, BufMut, BytesMut, HttpRequest, HttpResponse},
-    App, HttpServer, Responder,
+    delete, get, middleware, post, put,
+    web::{self, BufMut, BytesMut},
+    App, HttpRequest, HttpResponse, HttpServer, Responder,
 };
 use notify::{watcher, DebouncedEvent, RecursiveMode, Watcher};
 use rayon::prelude::*;
@@ -74,12 +73,12 @@ async fn get_image(
 
             debug!("{:?}", thumb_path);
             if let Ok(file) = NamedFile::open(&thumb_path) {
-                file.into_response(&request).unwrap()
+                file.into_response(&request)
             } else {
                 HttpResponse::NotFound().finish()
             }
         } else if let Ok(file) = NamedFile::open(path) {
-            file.into_response(&request).unwrap()
+            file.into_response(&request)
         } else {
             HttpResponse::NotFound().finish()
         }
@@ -114,7 +113,7 @@ async fn upload_image(_: Claims, mut payload: mp::Multipart) -> impl Responder {
     let mut file_path: Option<String> = None;
 
     while let Some(Ok(mut part)) = payload.next().await {
-        if let Some(content_type) = part.content_disposition() {
+        let content_type = part.content_disposition();
         debug!("{:?}", content_type);
         if let Some(filename) = content_type.get_filename() {
             debug!("Name: {:?}", filename);
@@ -131,7 +130,6 @@ async fn upload_image(_: Claims, mut payload: mp::Multipart) -> impl Responder {
             }
         }
     }
-    }
 
     let path = file_path.unwrap_or_else(|| dotenv::var("BASE_PATH").unwrap());
     if !file_content.is_empty() {
@@ -194,7 +192,7 @@ async fn stream_video(
     if !playlist.starts_with("tmp") && is_valid_path(playlist) != None {
         HttpResponse::BadRequest().finish()
     } else if let Ok(file) = NamedFile::open(playlist) {
-        file.into_response(&request).unwrap()
+        file.into_response(&request)
     } else {
         HttpResponse::NotFound().finish()
     }
@@ -210,7 +208,7 @@ async fn get_video_part(
     debug!("Video part: {}", part);
 
     if let Ok(file) = NamedFile::open(String::from("tmp/") + part) {
-        file.into_response(&request).unwrap()
+        file.into_response(&request)
     } else {
         error!("Video part not found: tmp/{}", part);
         HttpResponse::NotFound().finish()
@@ -222,10 +220,10 @@ async fn favorites(
     claims: Claims,
     favorites_dao: web::Data<Box<dyn FavoriteDao>>,
 ) -> impl Responder {
-    let favorites =
-        web::block(move || favorites_dao.get_favorites(claims.sub.parse::<i32>().unwrap()))
-            .await
-            .unwrap()
+    match web::block(move || favorites_dao.get_favorites(claims.sub.parse::<i32>().unwrap())).await
+    {
+        Ok(Ok(favorites)) => {
+            let favorites = favorites
                 .into_iter()
                 .map(|favorite| favorite.path)
                 .collect::<Vec<String>>();
@@ -234,6 +232,13 @@ async fn favorites(
                 photos: favorites,
                 dirs: Vec::new(),
             })
+        }
+        Ok(Err(e)) => {
+            error!("Error getting favorites: {:?}", e);
+            HttpResponse::InternalServerError().finish()
+        }
+        Err(_) => HttpResponse::InternalServerError().finish(),
+    }
 }
 
 #[put("image/favorites")]
@@ -244,21 +249,27 @@ async fn put_add_favorite(
 ) -> impl Responder {
     if let Ok(user_id) = claims.sub.parse::<i32>() {
         let path = body.path.clone();
-        match web::block::<_, usize, DbError>(move || favorites_dao.add_favorite(user_id, &path))
+        match web::block::<_, Result<usize, DbError>>(move || {
+            favorites_dao.add_favorite(user_id, &path)
+        })
         .await
         {
-            Err(BlockingError::Error(e)) if e.kind == DbErrorKind::AlreadyExists => {
+            Ok(Err(e)) if e.kind == DbErrorKind::AlreadyExists => {
                 debug!("Favorite: {} exists for user: {}", &body.path, user_id);
                 HttpResponse::Ok()
             }
-            Err(e) => {
+            Ok(Err(e)) => {
                 info!("{:?} {}. for user: {}", e, body.path, user_id);
                 HttpResponse::BadRequest()
             }
-            Ok(_) => {
+            Ok(Ok(_)) => {
                 debug!("Adding favorite \"{}\" for userid: {}", body.path, user_id);
                 HttpResponse::Created()
             }
+            Err(e) => {
+                error!("Blocking error while inserting favorite: {:?}", e);
+                HttpResponse::InternalServerError()
+            }
         }
     } else {
         error!("Unable to parse sub as i32: {}", claims.sub);
@@ -274,9 +285,8 @@ async fn delete_favorite(
 ) -> impl Responder {
     if let Ok(user_id) = claims.sub.parse::<i32>() {
         let path = body.path.clone();
-        web::block::<_, _, String>(move || {
+        web::block(move || {
             favorites_dao.remove_favorite(user_id, path);
-            Ok(())
         })
         .await
         .unwrap();
@@ -440,7 +450,8 @@ fn main() -> std::io::Result<()> {
         }
     });
 
-    let system = actix::System::new("image-api");
+    let system = actix::System::new();
+    system.block_on(async {
         let act = StreamActor {}.start();
 
         let app_data = web::Data::new(AppState {
@@ -448,7 +459,11 @@ fn main() -> std::io::Result<()> {
         });
 
         let labels = HashMap::new();
-    let prometheus = PrometheusMetrics::new("", Some("/metrics"), Some(labels));
+        let prometheus = PrometheusMetricsBuilder::new("api")
+            .const_labels(labels)
+            .build()
+            .expect("Unable to build prometheus metrics middleware");
+
         prometheus
             .registry
             .register(Box::new(IMAGE_GAUGE.clone()))
@@ -475,15 +490,15 @@ fn main() -> std::io::Result<()> {
                 .service(delete_favorite)
                 .service(get_file_metadata)
                 .app_data(app_data.clone())
-            .data::<Box<dyn UserDao>>(Box::new(user_dao))
-            .data::<Box<dyn FavoriteDao>>(Box::new(favorites_dao))
+                .app_data::<Data<Box<dyn UserDao>>>(Data::new(Box::new(user_dao)))
+                .app_data::<Data<Box<dyn FavoriteDao>>>(Data::new(Box::new(favorites_dao)))
                 .wrap(prometheus.clone())
         })
         .bind(dotenv::var("BIND_URL").unwrap())?
         .bind("localhost:8088")?
-    .run();
-
-    system.run()
+        .run()
+        .await
+    })
 }
 
 struct AppState {

src/testhelpers.rs

@@ -1,5 +1,6 @@
-use actix_web::dev::{Body, ResponseBody};
-use serde::Deserialize;
+use actix_web::body::MessageBody;
+use actix_web::{body::BoxBody, HttpResponse};
+use serde::de::DeserializeOwned;
 
 use crate::database::{models::User, UserDao};
 use std::cell::RefCell;
@@ -55,32 +56,26 @@ impl UserDao for TestUserDao {
 }
 
 pub trait BodyReader {
-    fn read_to_str(&self) -> &str;
+    fn read_to_str(self) -> String;
 }
 
-impl BodyReader for ResponseBody<Body> {
-    fn read_to_str(&self) -> &str {
-        match self {
-            ResponseBody::Body(Body::Bytes(ref b)) => std::str::from_utf8(b).unwrap(),
-            _ => panic!("Unknown response body"),
-        }
+impl BodyReader for HttpResponse<BoxBody> {
+    fn read_to_str(self) -> String {
+        let body = self.into_body().try_into_bytes().unwrap();
+        std::str::from_utf8(&body).unwrap().to_string()
     }
 }
 
-pub trait TypedBodyReader<'a, T>
+pub trait TypedBodyReader<T>
 where
-    T: Deserialize<'a>,
+    T: DeserializeOwned,
 {
-    fn read_body(&'a self) -> T;
+    fn read_body(self) -> T;
 }
 
-impl<'a, T: Deserialize<'a>> TypedBodyReader<'a, T> for ResponseBody<Body> {
-    fn read_body(&'a self) -> T {
-        match self {
-            ResponseBody::Body(Body::Bytes(ref b)) => {
-                serde_json::from_str(std::str::from_utf8(b).unwrap()).unwrap()
-            }
-            _ => panic!("Unknown response body"),
-        }
+impl<T: DeserializeOwned> TypedBodyReader<T> for HttpResponse<BoxBody> {
+    fn read_body(self) -> T {
+        let body = self.read_to_str();
+        serde_json::from_value(serde_json::Value::String(body.clone())).unwrap()
     }
 }

src/video.rs

@@ -67,7 +67,7 @@ pub async fn create_playlist(video_path: &str, playlist_file: &str) -> Result<Ch
 
     let start_time = std::time::Instant::now();
     loop {
-        actix::clock::delay_for(std::time::Duration::from_secs(1)).await;
+        actix::clock::sleep(std::time::Duration::from_secs(1)).await;
 
         if Path::new(playlist_file).exists()
             || std::time::Instant::now() - start_time > std::time::Duration::from_secs(5)