Merge pull request #285 from golemfactory/feature/repo

Feature/repo

gu-hub/Cargo.toml

@@ -17,6 +17,7 @@ gu-model = { path = "../gu-model" }
 gu-net = { path = "../gu-net" }
 gu-persist = { path = "../gu-persist" }
 
+tempfile = "3.1.0"
 actix = "0.7"
 actix-web = { version = "0.7", default-features = false }
 actix_derive = "0.3.0"
@@ -32,7 +33,7 @@ prettytable-rs = "0.7"
 semver = { version = "0.9", features = ["serde"] }
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
-sha1 = "0.6.0"
+sha1 = { version = "0.6.0", features=["std"] }
 zip = "0.4"
 openssl = { version = "0.10", features = ["vendored"], optional=true }
 

gu-hub/src/main.rs

@@ -42,6 +42,7 @@ mod local_service;
 mod peer;
 mod plugins;
 mod proxy_service;
+mod repo;
 mod server;
 mod sessions;
 
@@ -65,6 +66,7 @@ fn main() {
             .chain(peer::PeerModule::new())
             .chain(AutocompleteModule::new())
             .chain(hub_info::module())
+            .chain(repo::module())
             .chain(server::ServerModule::new()),
     );
 }

gu-hub/src/repo.rs

@@ -0,0 +1,91 @@
+use actix_web::error::{ErrorBadRequest, ErrorInternalServerError};
+use actix_web::fs::NamedFile;
+use actix_web::{App, HttpMessage, HttpRequest, HttpResponse};
+use futures::prelude::*;
+use gu_base::{Decorator, Module};
+use std::io::prelude::*;
+use std::path::{Path, PathBuf};
+use std::rc::Rc;
+use std::sync::{Mutex, RwLock};
+
+use gu_persist::config::ConfigModule;
+use tempfile::NamedTempFile;
+
+struct RepoModule {
+    // repo, repo_cache
+    paths: Mutex<Option<(PathBuf, PathBuf)>>,
+}
+
+impl Module for RepoModule {
+    fn run<D: Decorator + Clone + 'static>(&self, decorator: D) {
+        let config_module: &ConfigModule = decorator.extract().unwrap();
+
+        let repo = config_module.cache_dir().join("repo");
+        let repo_cache = config_module.cache_dir().join("repo-temp");
+
+        let mut g = self.paths.lock().unwrap();
+        std::fs::create_dir_all(&repo).unwrap();
+        std::fs::create_dir_all(&repo_cache).unwrap();
+
+        *g = Some((repo, repo_cache))
+    }
+
+    fn decorate_webapp<S: 'static>(&self, app: App<S>) -> App<S> {
+        let (repo, repo_cache) = self.paths.lock().unwrap().clone().unwrap();
+
+        let repo_temp: Rc<Path> = repo_cache.into();
+
+        let cache_path: Rc<Path> = repo.into();
+        let cache_path_get = cache_path.clone();
+        std::fs::create_dir_all(&cache_path).unwrap();
+
+        app
+                .resource("/repo", |r| {
+                r.post().with_async(move |r: HttpRequest<S>| {
+                    let cache_path = cache_path.clone();
+                    let lob_file = Rc::new(RwLock::new(gu_actix::async_try!(NamedTempFile::new_in(repo_temp.as_ref()))));
+                    let sha1 = Rc::new(RwLock::new(sha1::Sha1::new()));
+
+                    let lob_file_f = lob_file.clone();
+                    let sha1_f = sha1.clone();
+
+                    gu_actix::async_result! {
+                    r.payload()
+                        .map_err(|e| ErrorBadRequest(format!("Couldn't get request body: {:?}", e)))
+                        .for_each(move |chunk| {
+                            lob_file.write().unwrap().write_all(chunk.as_ref())?;
+                            sha1.write().unwrap().update(chunk.as_ref());
+                            Ok(())
+                        })
+                        .and_then(move |()| {
+                            let hash = sha1_f.write().unwrap().hexdigest();
+
+                            Rc::try_unwrap(lob_file_f).unwrap()
+                                .into_inner()
+                                .unwrap()
+                                .persist(cache_path.join(&hash))
+                                .map_err(|e| ErrorInternalServerError(format!("Couldn't save image: {:?}", e)))?;
+
+                            Ok(HttpResponse::Created()
+                                .header("Location", format!("/repo/{}", hash))
+                                .json(hash))
+                        })
+                }
+                })
+            })
+            .resource("/repo/{hash}", move |r| {
+                let cache_path = cache_path_get.clone();
+                r.get().with(move |p: actix_web::Path<(String, )>| -> Result<NamedFile, actix_web::Error>{
+                    let hexhash = p.0.as_str();
+                    let file_path = cache_path.join(hexhash);
+                    Ok(NamedFile::open(file_path)?)
+                })
+            })
+    }
+}
+
+pub fn module() -> impl Module {
+    RepoModule {
+        paths: Mutex::new(None),
+    }
+}

gu-hub/src/sessions/manager.rs

@@ -339,11 +339,12 @@ impl Handler<CreateDeployment> for SessionsManager {
             ActorResponse::r#async(
                 fut::wrap_future(session.create_deployment(msg.node_id, msg.deployment_desc))
                     .and_then(move |deployment_id, act: &mut SessionsManager, _ctx| {
-                        act.sessions
-                            .get_mut(&session_id)
-                            .unwrap()
-                            .add_deployment(node_id, deployment_id.clone());
-                        fut::ok(deployment_id)
+                        if let Some(session) = act.sessions.get_mut(&session_id) {
+                            session.add_deployment(node_id, deployment_id.clone());
+                            fut::ok(deployment_id)
+                        } else {
+                            fut::err(SessionErr::SessionNotFoundError)
+                        }
                     }),
             )
         } else {
@@ -382,11 +383,9 @@ impl Handler<DeleteDeployment> for SessionsManager {
             ActorResponse::r#async(
                 fut::wrap_future(session.delete_deployment(msg.node_id, msg.deployment_id))
                     .and_then(move |_, act: &mut SessionsManager, _ctx| {
-                        if !(act
-                            .sessions
-                            .get_mut(&session_id)
-                            .unwrap()
-                            .remove_deployment(node_id, deployment_id.clone()))
+                        if act.sessions.get_mut(&session_id).map(|session| {
+                            session.remove_deployment(node_id, deployment_id.clone())
+                        }) != Some(true)
                         {
                             return fut::err(SessionErr::DeploymentNotFound(deployment_id));
                         }

gu-provider/src/exec_plugin.rs

@@ -119,6 +119,7 @@ impl Handler<CreateSession<<Self as EnvManService>::CreateOptions>> for PluginMa
             image_path
                 .into_actor(self)
                 .and_then(|image_path, act, _ctx| {
+                    log::debug!("validate: {}", image_path.display());
                     let async_status = match process::Command::new(&act.exec)
                         .args(&["validate-image".as_ref(), image_path.as_path().as_os_str()])
                         .status_async()
@@ -395,11 +396,13 @@ fn resolve_path(
         .output_async()
         .map_err(|e| format!("driver error: {}", e))
         .and_then(|output| {
-            eprintln!(
-                "stderr={}",
-                std::str::from_utf8(output.stderr.as_ref()).unwrap()
-            );
-            eprintln!(
+            if !output.stderr.is_empty() {
+                log::info!(
+                    "stderr={}",
+                    std::str::from_utf8(output.stderr.as_ref()).unwrap()
+                );
+            }
+            log::debug!(
                 "output={}",
                 std::str::from_utf8(output.stdout.as_ref()).unwrap()
             );