Add ability to download raw and upload jpg from/to store

Cargo.toml

@@ -6,12 +6,19 @@ authors = ["Christoph Wurst <[email protected]>"]
 [dependencies]
 amq-protocol = "^0.16"
 dotenv = "0.8.0"
+env_logger = "0.4.3"
 futures = "^0.1"
+hyper = "^0.10"
 lapin-futures = "^0.8"
+log = "^0.3"
 resolve = "0.1.2"
 rocket = "0.2.8"
 rocket_codegen = "0.2.8"
 schani = { git = "https://github.com/schani-rs/schani.git" }
+serde_json="1.0.2"
+serde_urlencoded = "0.5.1"
+temporary = "0.6.3"
 tokio-core = "0.1.7"
 tokio-io = "^0.1"
 tokio-timer = "^0.1"
+url = "^1.4"

src/bin/processor.rs

@@ -1,8 +1,10 @@
 extern crate amq_protocol;
+extern crate env_logger;
 extern crate dotenv;
 extern crate futures;
 extern crate lapin_futures as lapin;
 extern crate resolve;
+extern crate schani_processor;
 extern crate tokio_core;
 
 use std::env;
@@ -13,61 +15,74 @@ use futures::Stream;
 use futures::future::Future;
 use lapin::client::ConnectionOptions;
 use lapin::channel::{BasicConsumeOptions, QueueDeclareOptions};
+use schani_processor::process_raw_image;
 use tokio_core::reactor::Core;
 use tokio_core::net::TcpStream;
 use resolve::resolve_host;
 
 fn main() {
+    env_logger::init().unwrap();
     dotenv().ok();
 
     // create the reactor
     let mut core = Core::new().unwrap();
     let handle = core.handle();
     let host = env::var("AMQP_ADDRESS").expect("AMQP_ADDRESS must be set");
-    let host_addr = resolve_host(&host).expect("could not lookup host").last().unwrap();
+    let host_addr = resolve_host(&host)
+        .expect("could not lookup host")
+        .last()
+        .unwrap();
     let addr = net::SocketAddr::new(host_addr, 5672);
 
     println!("connecting to AMQP service at {}", host_addr);
-    core.run(TcpStream::connect(&addr, &handle)
-                 .and_then(|stream| {
+    core.run(
+        TcpStream::connect(&addr, &handle)
+            .and_then(|stream| {
 
-                               // connect() returns a future of an AMQP Client
-                               // that resolves once the handshake is done
-                               lapin::client::Client::connect(stream, &ConnectionOptions::default())
-                           })
-                 .and_then(|client| {
+                // connect() returns a future of an AMQP Client
+                // that resolves once the handshake is done
+                lapin::client::Client::connect(stream, &ConnectionOptions::default())
+            })
+            .and_then(|client| {
 
-                               // create_channel returns a future that is resolved
-                               // once the channel is successfully created
-                               client.create_channel()
-                           })
-                 .and_then(|channel| {
-            let id = channel.id;
-            println!("created channel with id: {}", id);
+                // create_channel returns a future that is resolved
+                // once the channel is successfully created
+                client.create_channel()
+            })
+            .and_then(|channel| {
+                let id = channel.id;
+                println!("created channel with id: {}", id);
 
-            let ch = channel.clone();
-            channel
-                .queue_declare("raw", &QueueDeclareOptions::default(), FieldTable::new())
-                .and_then(move |_| {
-                    println!("channel {} declared queue {}", id, "hello");
+                let ch = channel.clone();
+                channel
+                    .queue_declare("raw", &QueueDeclareOptions::default(), FieldTable::new())
+                    .and_then(move |_| {
+                        println!("channel {} declared queue {}", id, "raw");
 
-                    // basic_consume returns a future of a message
-                    // stream. Any time a message arrives for this consumer,
-                    // the for_each method would be called
-                    channel.basic_consume("raw", "raw_processor", &BasicConsumeOptions::default())
-                })
-                .and_then(|stream| {
-                    println!("got consumer stream");
+                        // basic_consume returns a future of a message
+                        // stream. Any time a message arrives for this consumer,
+                        // the for_each method would be called
+                        channel.basic_consume(
+                            "raw",
+                            "raw_processor",
+                            &BasicConsumeOptions::default(),
+                        )
+                    })
+                    .and_then(|stream| {
+                        println!("got consumer stream");
 
-                    stream.for_each(move |message| {
-                        let file_id_str = std::str::from_utf8(&message.data).unwrap();
-                        let file_id = file_id_str.parse::<u64>().unwrap();
-                        println!("got message: {:?}", message);
-                        println!("file id: {:?}", file_id);
-                        ch.basic_ack(message.delivery_tag);
-                        Ok(())
+                        stream.for_each(move |message| {
+                            let file_id_str = std::str::from_utf8(&message.data).unwrap();
+                            let file_id = file_id_str.parse::<i32>().unwrap();
+                            println!("got message: {:?}", message);
+                            println!("file id: {:?}", file_id);
+                            try!(process_raw_image(file_id).map_err(|err| {
+                                std::io::Error::new(std::io::ErrorKind::Other, err)
+                            }));
+                            ch.basic_ack(message.delivery_tag);
+                            Ok(())
+                        })
                     })
-                })
-        }))
-        .unwrap();
-}
+            }),
+    ).unwrap();
+}

src/error.rs

@@ -0,0 +1,69 @@
+use hyper::error::Error as HyperError;
+use std::convert::From;
+use std::error;
+use std::fmt;
+use std::io;
+use url::ParseError;
+
+#[derive(Debug)]
+pub enum Error {
+    Generic(String),
+    HTTP(HyperError),
+    IO(io::Error),
+    URLError(ParseError),
+}
+
+impl fmt::Display for Error {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        match *self {
+            Error::Generic(ref msg) => write!(f, "{}", msg),
+            Error::HTTP(ref msg) => write!(f, "HTTP error: {}", msg),
+            Error::IO(ref err) => write!(f, "IO error: {}", err),
+            Error::URLError(ref err) => write!(f, "URL parsing error: {}", err),
+        }
+    }
+}
+
+impl error::Error for Error {
+    fn description(&self) -> &str {
+        match *self {
+            Error::Generic(ref msg) => &msg,
+            Error::HTTP(ref err) => err.description(),
+            Error::IO(ref err) => err.description(),
+            Error::URLError(ref err) => err.description(),
+        }
+    }
+
+    fn cause(&self) -> Option<&error::Error> {
+        match *self {
+            Error::Generic(_) => None,
+            Error::HTTP(ref err) => err.cause(),
+            Error::IO(ref err) => err.cause(),
+            Error::URLError(ref err) => err.cause(),
+        }
+    }
+}
+
+impl From<&'static str> for Error {
+    fn from(orig: &str) -> Self {
+        Error::Generic(orig.to_owned())
+    }
+}
+
+impl From<ParseError> for Error {
+    fn from(orig: ParseError) -> Self {
+        Error::URLError(orig)
+    }
+}
+
+impl From<HyperError> for Error {
+    fn from(orig: HyperError) -> Self {
+        Error::HTTP(orig)
+    }
+}
+
+impl From<io::Error> for Error {
+    fn from(orig: io::Error) -> Self {
+        Error::IO(orig)
+    }
+}

src/lib.rs

@@ -1,3 +1,35 @@
+extern crate hyper;
+#[macro_use]
+extern crate log;
 extern crate schani;
+extern crate serde_json;
+extern crate serde_urlencoded;
+extern crate temporary;
+extern crate url;
 
-pub mod rawtherapee;
+pub mod rawtherapee;
+pub mod error;
+mod store;
+
+use rawtherapee::process_raw;
+use store::{load_raw_file, upload_image_file};
+use temporary::Directory;
+
+pub fn process_raw_image(image_id: i32) -> Result<(), error::Error> {
+    // Create a temporary directory.
+    info!("processing image {} …", image_id);
+    let directory = Directory::new("raw_images").unwrap();
+
+    let tmp_path = directory.join(image_id.to_string() + &".NEF".to_string());
+    println!("{:?}", tmp_path);
+    let target_path = directory.into_path();
+
+    try!(load_raw_file(image_id, &tmp_path));
+    info!("loaded image {} …", image_id);
+    try!(process_raw(&tmp_path, &target_path));
+    info!("processed image {} …", image_id);
+    try!(upload_image_file(image_id, &target_path));
+    info!("uploaded image {} …", image_id);
+
+    Ok(())
+}

src/rawtherapee.rs

@@ -1,17 +1,21 @@
-use schani::images::RawtherapeeImage;
+use std::path::PathBuf;
 use std::process::Command;
 use std::result::Result;
 
-pub fn process_raw(img: &RawtherapeeImage) -> Result<(), &'static str> {
-    let status = try!(Command::new("rawtherapee-cli")
+pub fn process_raw(path: &PathBuf, out: &PathBuf) -> Result<(), &'static str> {
+    let output = try!(Command::new("rawtherapee")
                           .arg("-j90")
                           .arg("-Y")
+                          .arg("-O")
+                          .arg(out)
                           .arg("-c")
-                          .arg(&img.raw)
-                          .status()
-                          .map_err(|_| "could not process raw image"));
+                          .arg(path)
+                          .output()
+                          .map_err(|_| "could not start rawtherapee"));
 
-    if status.success() {
+    println!("{:?}", output);
+
+    if output.status.success() {
         Ok(())
     } else {
         Err("rawtherapee process returned error status code")
@@ -20,8 +24,8 @@ pub fn process_raw(img: &RawtherapeeImage) -> Result<(), &'static str> {
 
 #[cfg(test)]
 mod test {
-    use super::super::schani::images::RawtherapeeImage;
     use super::process_raw;
+    use std::path::PathBuf;
     use std::process::Command;
 
     fn is_rawtherapee_installed() -> bool {
@@ -35,13 +39,10 @@ mod test {
             return ();
         }
 
-        let raw1 = RawtherapeeImage {
-            name: "raw1".to_string(),
-            raw: "resources/RAW1.NEF".to_string(),
-            sidecar: "resources/RAW1.NEF.pp3".to_string(),
-        };
+        let path = PathBuf::from("resources/RAW1.NEF");
+        let out = PathBuf::from("/tmp/");
 
-        assert!(process_raw(&raw1).is_ok());
+        assert!(process_raw(&path, &out).is_ok());
     }
 
     #[test]
@@ -51,12 +52,9 @@ mod test {
             return ();
         }
 
-        let raw1 = RawtherapeeImage {
-            name: "x".to_string(),
-            raw: "resources/x".to_string(),
-            sidecar: "resources/x".to_string(),
-        };
+        let path = PathBuf::from("resources/nope.NEF");
+        let out = PathBuf::from("/tmp/");
 
-        assert!(process_raw(&raw1).is_err());
+        assert!(process_raw(&path, &out).is_err());
     }
 }