Added first image processing decoupled from the upload

Georg Hopp
1 parent abe77b68
Showing 6 changed files with 125 additions and 30 deletions
server/Cargo.toml
server/src/main.rs
server/src/routes/markdown.rs
server/src/routes/upload.rs
server/src/routes/user.rs
server/src/uuid.rs
--- a/server/Cargo.toml
View file @76eebfe
+++ b/server/Cargo.toml
View file @76eebfe
@@ -20,9 +20,10 @@ diffy = "0.2"
 dotenv = "0.15.0"
 flate2 = "^1.0"
 futures = "^0.3"
+futures-util = { version = "0", features = ["std"] }
 listenfd = "0.3"
 r2d2 = "0.8.9"
 serde = { version = "^1.0", features = ["derive"] }
 serde_derive = "1.0"
 serde_json = "1.0"
-uuid = { version = "^0.8", features = ["v4"] }
+uuid = { version = "^0.8", features = ["v4", "v5"] }
--- a/server/src/main.rs
View file @76eebfe
+++ b/server/src/main.rs
View file @76eebfe
@@ -5,33 +5,94 @@ mod error;
 mod models;
 mod routes;
 mod schema;
+mod uuid;
 use crate::routes::markdown::*;
 use crate::routes::other::*;
 use crate::routes::user::*;
+use crate::uuid::Uuid;
 use actix_web::{guard, web, App, HttpResponse, HttpServer};
+use async_std::channel::Receiver;
+use async_std::channel::Sender;
+use async_std::channel::bounded;
+use async_std::fs::File;
 use diesel::r2d2::{self, ConnectionManager};
 use diesel::SqliteConnection;
+use futures::AsyncReadExt;
+use futures::AsyncSeekExt;
 use listenfd::ListenFd;
 use routes::markdown::get_markdown;
 use routes::upload::upload;
+use std::io::SeekFrom;
+use std::sync::Arc;
 pub(crate) type Pool = r2d2::Pool<ConnectionManager<SqliteConnection>>;
+#[derive(Clone)]
+pub struct AppData {
+    pub database_pool: Arc<Pool>,
+    pub tx_upload_worker: Sender<String>,
+}
+
+async fn read_at( f   :&mut File
+                , pos :SeekFrom
+                , buf :&mut [u8]) -> std::io::Result<()> {
+    f.seek(pos).await?;
+    f.read_exact(buf).await
+}
+
+async fn get_sample( f   :&mut File
+                   , buf :&mut [u8]) -> std::io::Result<()> {
+    let file_len = f.metadata().await?.len();
+    let chunk_size = buf.len() / 3;
+
+    read_at(f, SeekFrom::Start(0), &mut buf[0..chunk_size]).await?;
+    if file_len >= 2 * chunk_size as u64 {
+        read_at( f
+               , SeekFrom::End(-(chunk_size as i64))
+               , &mut buf[2*chunk_size..]).await?;
+    }
+    if file_len >= 3 * chunk_size as u64 {
+        read_at( f
+               , SeekFrom::Start((file_len-chunk_size as u64) / 2)
+               , &mut buf[chunk_size..2*chunk_size]).await?;
+    }
+
+    Ok(())
+}
+
 #[actix_rt::main]
 async fn main() -> std::io::Result<()> {
     let mut listenfd = ListenFd::from_env();
     dotenv::dotenv().ok();
+    let ( tx_upload_worker
+        , rx_upload_worker ) :( Sender<String>
+                              , Receiver<String> ) = bounded(32);
+
+    let _upload_worker = actix_rt::spawn(async move {
+        while let Ok(filename) = rx_upload_worker.recv().await {
+            let mut f = File::open(&filename).await.unwrap();
+            let mut buf = vec!['.' as u8; 3 * 4096];
+            get_sample(&mut f, buf.as_mut()).await.unwrap();
+            println!("[UPLOAD WORKER] filename: {}", filename);
+            println!( "[UPLOAD WORKER] uuid: {}"
+                    , Uuid::get("some.unique.namespace", buf.as_mut()) );
+        }
+    });
+
     let database_url = std::env::var("DATABASE_URL").expect("NOT FOUND");
     let database_pool = Pool::builder()
         .build(ConnectionManager::new(database_url))
         .unwrap();
+    let database_pool = Arc::new(database_pool);
+    let app_data = AppData { database_pool, tx_upload_worker };
+
     let server = HttpServer::new(move || {
-        App::new() . data(database_pool.clone())
+        App::new() . data(app_data.clone())
                    . service(actix_files::Files::new("/static", "./static"))
                    . service( web::scope("/api/v0")
                             . service( web::resource("/upload")
--- a/server/src/routes/markdown.rs
View file @76eebfe
+++ b/server/src/routes/markdown.rs
View file @76eebfe
-use crate::models::markdown;
-use crate::Pool;
+use crate::{models::markdown, AppData};
 use actix_web::{Error, HttpResponse, web};
 use anyhow::Result;
@@ -11,36 +10,40 @@ pub struct Patchset {
     patch: Option<i32>,
 }
-pub async fn get_markdowns(pool: web::Data<Pool>)
+pub async fn get_markdowns(app_data: web::Data<AppData>)
     -> Result<HttpResponse, Error>
 {
-    Ok( web::block(move || markdown::get_markdowns(pool.into_inner()))
+    let pool = app_data.database_pool.clone();
+
+    Ok( web::block(move || markdown::get_markdowns(pool))
       . await
       . map(|markdowns| HttpResponse::Ok().json(markdowns))
       . map_err(|_| HttpResponse::InternalServerError())?
       )
 }
-pub async fn get_markdown( pool: web::Data<Pool>
+pub async fn get_markdown( app_data: web::Data<AppData>
                          , name: web::Path<String>
                          , patch: web::Query<Patchset>
                          ) -> Result<HttpResponse, Error>
 {
-    let pool = pool.into_inner();
+    let pool = app_data.database_pool.clone();
     let name = name.into_inner();
     let patch = patch.into_inner();
-    Ok( web::block(move || markdown::get_markdown(pool, name.as_str(), patch.patch))
+    Ok( web::block(move || markdown::get_markdown( pool
+                                                 , name.as_str()
+                                                 , patch.patch) )
       . await
       . map(|markdowns| HttpResponse::Ok().json(markdowns))
       . map_err(|_| HttpResponse::InternalServerError())?
       )
 }
-pub async fn get_patches( pool: web::Data<Pool>
+pub async fn get_patches( app_data: web::Data<AppData>
                         , name: web::Path<String>
                         ) -> Result<HttpResponse, Error> {
-    let pool = pool.into_inner();
+    let pool = app_data.database_pool.clone();
     let name = name.into_inner();
     Ok( web::block(move || markdown::get_patches(pool, name.as_str()))
@@ -50,12 +53,12 @@ pub async fn get_patches( pool: web::Data<Pool>
       )
 }
-pub async fn update_markdown( pool: web::Data<Pool>
+pub async fn update_markdown( app_data: web::Data<AppData>
                             , name: web::Path<String>
                             , item: web::Json<MarkdownJson> )
     -> Result<HttpResponse, Error>
 {
-    let pool = pool.into_inner();
+    let pool = app_data.database_pool.clone();
     let name = name.into_inner();
     let item = item.into_inner();
--- a/server/src/routes/upload.rs
View file @76eebfe
+++ b/server/src/routes/upload.rs
View file @76eebfe
@@ -4,19 +4,25 @@ use futures::stream::StreamExt;
 use async_std::{fs::OpenOptions, io::WriteExt};
 use uuid::Uuid;
-pub async fn upload(mut body: web::Payload) -> Result<HttpResponse, Error>
+use crate::AppData;
+
+pub async fn upload( app_data: web::Data<AppData>
+                   , mut body: web::Payload) -> Result<HttpResponse, Error>
 {
+    let worker = app_data.tx_upload_worker.clone();
+
+    let upload_filename = format!("/tmp/upload_{}", Uuid::new_v4());
     let mut output = OpenOptions::new();
     output . create(true)
            . write(true);
-    let mut output = output
-                   . open(format!("/tmp/upload_{}", Uuid::new_v4()))
-                   . await
-                   . unwrap();
+    let mut output = output.open(&upload_filename).await?;
     while let Some(item) = body.next().await {
-        output.write_all(&item?).await.unwrap();
+        output.write_all(&item?).await?;
     }
+    // TODO handle this as error response...
+    worker.send(upload_filename).await.unwrap();
+
     Ok(HttpResponse::Ok().finish())
 }
--- a/server/src/routes/user.rs
View file @76eebfe
+++ b/server/src/routes/user.rs
View file @76eebfe
 use crate::models::user::{self, Action};
-use crate::Pool;
+use crate::AppData;
 use actix_web::{Error, HttpResponse, web};
 use anyhow::Result;
-pub async fn create_user( pool: web::Data<Pool>
+pub async fn create_user( app_data: web::Data<AppData>
                         , item: web::Json<user::UserJson> )
     -> Result<HttpResponse, Error>
 {
-    let pool = pool.into_inner();
+    let pool = app_data.database_pool.clone();
     let item = item.into_inner();
     Ok(web::block(move || user::create_user(pool, item))
@@ -21,19 +21,21 @@ pub async fn create_user( pool: web::Data<Pool>
        . map_err(|_| HttpResponse::InternalServerError())?)
 }
-pub async fn get_users(pool: web::Data<Pool>)
+pub async fn get_users(app_data: web::Data<AppData>)
     -> Result<HttpResponse, Error>
 {
-    Ok(web::block(move || user::get_users(pool.into_inner()))
+    let pool = app_data.database_pool.clone();
+
+    Ok(web::block(move || user::get_users(pool))
        . await
        . map(|users| HttpResponse::Ok().json(users))
        . map_err(|_| HttpResponse::InternalServerError())?)
 }
-pub async fn get_user(pool: web::Data<Pool>, id: web::Path<i32>)
+pub async fn get_user(app_data: web::Data<AppData>, id: web::Path<i32>)
     -> Result<HttpResponse, Error>
 {
-    let pool = pool.into_inner();
+    let pool = app_data.database_pool.clone();
     let id   = id.into_inner();
     Ok(web::block(move || user::get_user(pool, id))
@@ -42,10 +44,10 @@ pub async fn get_user(pool: web::Data<Pool>, id: web::Path<i32>)
        . map_err(|_| HttpResponse::InternalServerError())?)
 }
-pub async fn delete_user(pool: web::Data<Pool>, id: web::Path<i32>)
+pub async fn delete_user(app_data: web::Data<AppData>, id: web::Path<i32>)
     -> Result<HttpResponse, Error>
 {
-    let pool = pool.into_inner();
+    let pool = app_data.database_pool.clone();
     let id   = id.into_inner();
     Ok(web::block(move || user::delete_user(pool, id))
@@ -54,12 +56,12 @@ pub async fn delete_user(pool: web::Data<Pool>, id: web::Path<i32>)
        . map_err(|_| HttpResponse::InternalServerError())?)
 }
-pub async fn update_user( pool: web::Data<Pool>
+pub async fn update_user( app_data: web::Data<AppData>
                         , id:   web::Path<i32>
                         , item: web::Json<user::UserJson> )
     -> Result<HttpResponse, Error>
 {
-    let pool = pool.into_inner();
+    let pool = app_data.database_pool.clone();
     let id   = id.into_inner();
     let item = item.into_inner();
--- a/server/src/uuid.rs 0 → 100644
View file @76eebfe
+++ b/server/src/uuid.rs 0 → 100644
View file @76eebfe
+use std::fmt::Display;
+
+#[derive(Clone,Copy,Debug)]
+pub struct Uuid(pub uuid::Uuid);
+
+impl Display for Uuid {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{}", self.0)
+    }
+}
+
+macro_rules! ns {
+    ($n:expr) => {
+        uuid::Uuid::new_v5(&uuid::Uuid::NAMESPACE_DNS, $n.as_bytes())
+    }
+}
+
+impl Uuid {
+    pub fn get(ns: &str, buf: &mut [u8]) -> Self {
+        Self(uuid::Uuid::new_v5(&ns!(ns), buf))
+    }
+}