implement Network receive

server/texture-sync-server/src/persistency/image_convert/mod.rs

@@ -16,4 +16,4 @@ pub fn generate_preview(
     config: &ConvertConfig,
 ) -> ::image::ImageResult<Vec<u8>> {
     unimplemented!()
-}
+}

server/texture-sync-server/src/persistency/mod.rs

@@ -11,10 +11,10 @@ use std::path::{Path, PathBuf};
 use std::sync::Arc;
 
 pub use self::search::Query;
-mod search;
 mod image_convert;
+mod search;
 
-pub type TextureFileResult = Result< Arc<Vec<u8>> , TextureFileError>;
+pub type TextureFileResult = Result<Arc<Vec<u8>>, TextureFileError>;
 pub enum TextureFileError {
     NotFound,
     IoError(io::Error),
@@ -72,4 +72,3 @@ impl DataStore {
         unimplemented!();
     }
 }
-

server/texture-sync-server/src/persistency/search/mod.rs

@@ -35,4 +35,4 @@ enum QueryFilter {
     InName(String),
     MinResolution(usize),
     BeforeDate { year: u16, month: u16, day: u16 },
-}
+}

server/texture-sync-server/src/protocol/implementation/mod.rs

@@ -0,0 +1,62 @@
+use super::*;
+
+use std::io::*;
+use std::net::*;
+
+use std::thread;
+use std::time::Duration;
+
+use crate::model::*;
+
+mod protocol_connection;
+use self::protocol_connection::*;
+
+pub fn listen_forever<H>(handler: H, config: &ProtocolConfig) -> io::Result<()>
+where
+    H: 'static + ProtocolHandler + Sized,
+{
+    let listener = TcpListener::bind((config.listen_addr.as_str(), config.port))?;
+
+    for mut connection in listener.incoming() {
+        // If there is an successful connection,
+        // set timeouts.
+        // We ignore errors here, so they will be caught in the clients thread.
+        let _ = connection.as_mut().map(|stream| {
+            stream.set_read_timeout(Duration::from_secs(config.read_timeout_s).into())?;
+            stream.set_write_timeout(Duration::from_secs(config.read_timeout_s).into())
+        });
+
+        let mut handler = handler.clone();
+        let _ = thread::spawn(move || client_loop(connection?, handler));
+    }
+
+    Ok(())
+}
+
+fn client_loop<H>(connection: TcpStream, handler: H) -> io::Result<()>
+where
+    H: 'static + ProtocolHandler + Sized,
+{
+    let mut connection = Connection::from_tcp(connection)?;
+    loop {
+        let package = connection.receive()?;
+
+        match package {
+            Package::Error(_, _) => {
+                // Just clone the connection.
+                break;
+            }
+            Package::Binary(_) => {
+                connection.send(&Package::Error(400, "Unexpected Binary".to_string()))?;
+                break;
+            }
+            Package::Json(Command::Ping {}) => {
+                connection.send(&Package::Json(Command::Pong {}))?;
+            }
+            // TODO: lots
+            _ => unimplemented!(),
+        }
+    }
+
+    Ok(())
+}

server/texture-sync-server/src/protocol/implementation/protocol_connection.rs

@@ -0,0 +1,304 @@
+use crate::model::*;
+
+use serde::{Deserialize, Serialize};
+
+#[derive(Eq, PartialEq, Clone, Debug)]
+pub enum Package {
+    JsonNull,
+    JsonFalse,
+    JsonTrue,
+    Json(Command),
+    Binary(Vec<u8>),
+    Error(u16, String),
+}
+
+#[derive(Serialize, Deserialize, Eq, PartialEq, Clone, Debug)]
+pub enum Command {
+    #[serde(rename = "ping")]
+    Ping {},
+    #[serde(rename = "pong")]
+    Pong {},
+    #[serde(rename = "query")]
+    Query { query: Vec<String> },
+    #[serde(rename = "get_texture")]
+    GetTexture {
+        id: Option<String>,
+        name: Option<String>,
+    },
+    #[serde(rename = "get_texture_file")]
+    GetTextureData { texture_hash: Sha256 },
+    #[serde(rename = "get_texture_preview")]
+    GetTexturePreview {
+        texture_hash: Sha256,
+        desired_format: TextureFormat,
+    },
+    #[serde(rename = "replace_texture")]
+    ReplaceTexture {
+        old: Option<Texture>,
+        new: Option<Texture>,
+    },
+}
+
+use std::io::*;
+use std::net::*;
+
+pub struct Connection<R: Read + Sized, W: Write + Sized> {
+    reader: R,
+    writer: W,
+}
+
+const KIB: u32 = 1024;
+const MIB: u32 = 1024 * 1024;
+
+const PACKAGE_TYPE_ERROR: u8 = 0;
+const PACKAGE_TYPE_JSON: u8 = 1;
+const PACKAGE_TYPE_BIN: u8 = 2;
+
+impl Connection<BufReader<TcpStream>, BufWriter<TcpStream>> {
+    pub fn from_tcp(connection: TcpStream) -> Result<Self> {
+        let reader = BufReader::new(connection.try_clone()?);
+        let writer = BufWriter::new(connection);
+
+        Ok(Connection { reader, writer })
+    }
+}
+
+impl<R: Read + Sized, W: Write + Sized> Connection<R, W> {
+    #[cfg(test)]
+    pub fn new(reader: R, writer: W) -> Self {
+        Connection { reader, writer }
+    }
+
+    pub fn receive(&mut self) -> Result<Package> {
+        let mut payload_type_buffer = [0u8; 1];
+        let mut reserved_buffer = [0u8; 3];
+        let mut payload_length_buffer = [0u8; 4];
+
+        self.reader.read_exact(&mut payload_type_buffer)?;
+        self.reader.read_exact(&mut reserved_buffer)?;
+        self.reader.read_exact(&mut payload_length_buffer)?;
+
+        let payload_type = payload_type_buffer[0];
+        let payload_length = u32::from_be_bytes(payload_length_buffer);
+
+        // Check length.
+        match payload_type {
+            PACKAGE_TYPE_ERROR => {
+                if payload_length > 1 * KIB {
+                    return Err(Error::new(
+                        ErrorKind::InvalidData,
+                        "Maximum length of Error Package is 1 KiB.",
+                    ));
+                }
+            }
+            PACKAGE_TYPE_JSON => {
+                if payload_length > 16 * MIB {
+                    return Err(Error::new(
+                        ErrorKind::InvalidData,
+                        "Maximum length of JSON Package is 16 MiB.",
+                    ));
+                }
+            }
+            PACKAGE_TYPE_BIN => {
+                if payload_length > 512 * MIB {
+                    return Err(Error::new(
+                        ErrorKind::InvalidData,
+                        "Maximum length of Binary Package is 512 MiB.",
+                    ));
+                }
+            }
+            _ => {
+                return Err(Error::new(ErrorKind::InvalidData, "Unknown Package Type."));
+            }
+        }
+
+        let mut payload = vec![0u8; payload_length as usize];
+        self.reader.read_exact(&mut payload[..])?;
+
+        match payload_type {
+            PACKAGE_TYPE_ERROR => {
+                let contents = String::from_utf8(payload)
+                    .map_err(|_| Error::new(ErrorKind::InvalidData, "Invalid UTF-8."))?;
+
+                let mut parts = contents.splitn(2, " ");
+
+                match (parts.next(), parts.next()) {
+                    (Some(code), Some(info)) => {
+                        let code: u16 = code.parse().map_err(|_| {
+                            Error::new(ErrorKind::InvalidData, "Status code in error expected!")
+                        })?;
+                        Ok(Package::Error(code, info.to_string()))
+                    }
+
+                    _ => Err(Error::new(
+                        ErrorKind::InvalidData,
+                        "Status code in error expected!",
+                    )),
+                }
+            }
+            PACKAGE_TYPE_JSON => {
+                // try special packages first.
+                match serde_json::from_slice::<Option<bool>>(&payload[..]) {
+                    Ok(Some(true)) => {
+                        return Ok(Package::JsonTrue);
+                    }
+
+                    Ok(Some(false)) => {
+                        return Ok(Package::JsonFalse);
+                    }
+
+                    Ok(None) => {
+                        return Ok(Package::JsonNull);
+                    }
+
+                    _ => (), // else try other
+                }
+
+                let json: Command = serde_json::from_slice(&payload[..]).map_err(|e| {
+                    #[cfg(test)]
+                    dbg!(&e);
+
+                    Error::new(ErrorKind::InvalidData, "Invalid JSON.")
+                })?;
+
+                Ok(Package::Json(json))
+            }
+            PACKAGE_TYPE_BIN => Ok(Package::Binary(payload)),
+            _ => {
+                // Covered in the match above.
+                unreachable!();
+            }
+        }
+    }
+
+    pub fn send(&mut self, pkg: &Package) -> Result<()> {
+        unimplemented!()
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+
+    #[test]
+    fn read_error_good() {
+        let mut read_data = Vec::new();
+
+        let msg = b"42 TEST";
+        read_data.extend_from_slice(&[0, 42, 42, 42, 0, 0, 0, msg.len() as u8]);
+        read_data.extend_from_slice(msg);
+
+        let mut c = Connection::new(&read_data[..], Vec::new());
+
+        assert_eq!(c.receive().unwrap(), Package::Error(42, "TEST".to_string()))
+    }
+
+    #[test]
+    fn read_error_bad() {
+        let mut read_data = Vec::new();
+
+        let msg = b"TEST XXXX";
+        read_data.extend_from_slice(&[0, 42, 42, 42, 0, 0, 0, msg.len() as u8]);
+        read_data.extend_from_slice(msg);
+
+        let mut c = Connection::new(&read_data[..], Vec::new());
+
+        assert_eq!(c.receive().is_err(), true)
+    }
+
+    #[test]
+    fn read_binary_good() {
+        let mut read_data = Vec::new();
+
+        let msg = b"Hello World";
+        read_data.extend_from_slice(&[2, 42, 42, 42, 0, 0, 0, msg.len() as u8]);
+        read_data.extend_from_slice(msg);
+
+        let mut c = Connection::new(&read_data[..], Vec::new());
+
+        assert_eq!(c.receive().unwrap(), Package::Binary(Vec::from(&msg[..])));
+    }
+
+    #[test]
+    fn read_binary_bad() {
+        let mut read_data = Vec::new();
+
+        let msg = b"Hello World";
+        // to large (size).
+        read_data.extend_from_slice(&[2, 42, 42, 42, 255, 0, 0, 0]);
+        read_data.extend_from_slice(msg);
+
+        let mut c = Connection::new(&read_data[..], Vec::new());
+
+        assert_eq!(c.receive().is_err(), true);
+    }
+
+    // we don't test the actual json parsing here,
+    // since serde_json is well tested.
+
+    // AndTesting the declaration seams not worth the effort.
+
+    #[test]
+    fn read_json_good() {
+        let mut read_data = Vec::new();
+
+        let msg = br#"{
+                "query" : {
+                    "query" : ["Hallo",   "Welt!"]
+                }
+            }"#;
+        read_data.extend_from_slice(&[1, 42, 42, 42, 0, 0, 0, msg.len() as u8]);
+        read_data.extend_from_slice(msg);
+
+        let mut c = Connection::new(&read_data[..], Vec::new());
+
+        assert_eq!(
+            c.receive().unwrap(),
+            Package::Json(Command::Query {
+                query: vec!["Hallo".to_string(), "Welt!".to_string()],
+            })
+        );
+    }
+
+    #[test]
+    fn read_json_multiple_special() {
+        let mut read_data = Vec::new();
+
+        let msg = br#"null"#;
+        read_data.extend_from_slice(&[1, 42, 42, 42, 0, 0, 0, msg.len() as u8]);
+        read_data.extend_from_slice(msg);
+
+        let msg = br#"true"#;
+        read_data.extend_from_slice(&[1, 42, 42, 42, 0, 0, 0, msg.len() as u8]);
+        read_data.extend_from_slice(msg);
+
+        let msg = br#"false"#;
+        read_data.extend_from_slice(&[1, 42, 42, 42, 0, 0, 0, msg.len() as u8]);
+        read_data.extend_from_slice(msg);
+
+        let mut c = Connection::new(&read_data[..], Vec::new());
+
+        assert_eq!(c.receive().unwrap(), Package::JsonNull);
+
+        assert_eq!(c.receive().unwrap(), Package::JsonTrue);
+
+        assert_eq!(c.receive().unwrap(), Package::JsonFalse);
+    }
+
+    #[test]
+    fn read_json_bad() {
+        let mut read_data = Vec::new();
+
+        let msg = br#"{
+                "big foot" : {
+                    "query" : ["Hallo",   "Welt!"]
+                }
+            }"#;
+        read_data.extend_from_slice(&[1, 42, 42, 42, 0, 0, 0, msg.len() as u8]);
+        read_data.extend_from_slice(msg);
+
+        let mut c = Connection::new(&read_data[..], Vec::new());
+
+        assert_eq!(c.receive().is_err(), true);
+    }
+}

server/texture-sync-server/src/protocol/mod.rs

@@ -3,28 +3,18 @@
 #![allow(unused_variables)]
 #![allow(dead_code)]
 
+mod results;
+pub use self::results::*;
+
+mod implementation;
+pub use self::implementation::listen_forever;
+
 use crate::model::*;
 
-use std::sync::Arc;
 use std::io;
+use std::sync::Arc;
 
-pub enum ReplaceTextureStatus {
+pub trait ProtocolHandler: Send + Sync + Clone {
-    // Call Again With Texture Binary
-    NeedTextureData,
-    // Done.
-    Ok,
-}
-
-pub type ProtocolResult<T> = Result<T, ProtocolError>;
-pub enum ProtocolError {
-    BadRequest(String),
-    FileNotFound(String),
-    Conflict(String),
-    InternalServerError(std::io::Error),
-    NotImplemented,
-}
-
-pub trait ProtocolHandler: Send + Sync {
     fn query(&mut self, query: &[String]) -> ProtocolResult<Vec<Texture>>;
 
     fn get_texture_by_id(&mut self, id: String) -> ProtocolResult<Option<Texture>>;
@@ -43,10 +33,27 @@ pub trait ProtocolHandler: Send + Sync {
     ) -> ProtocolResult<ReplaceTextureStatus>;
 }
 
+#[derive(Clone, Debug)]
 pub struct ProtocolConfig {
     pub port: u16,
+    pub read_timeout_s: u64,
+    pub write_timeout_s: u64,
+    pub listen_addr: String,
 }
 
-pub fn listen_forever(handler: &ProtocolHandler) -> io::Result<()> {
+impl ProtocolConfig {
-    unimplemented!()
+    pub fn new() -> ProtocolConfig {
-}
+        ProtocolConfig::default()
+    }
+}
+
+impl Default for ProtocolConfig {
+    fn default() -> ProtocolConfig {
+        ProtocolConfig {
+            port: 10796,
+            read_timeout_s: 60,
+            write_timeout_s: 75,
+            listen_addr: "::1".to_owned(),
+        }
+    }
+}

server/texture-sync-server/src/protocol/results.rs

@@ -0,0 +1,15 @@
+pub enum ReplaceTextureStatus {
+    // Call Again With Texture Binary
+    NeedTextureData,
+    // Done.
+    Ok,
+}
+
+pub type ProtocolResult<T> = Result<T, ProtocolError>;
+pub enum ProtocolError {
+    BadRequest(String),
+    FileNotFound(String),
+    Conflict(String),
+    InternalServerError(std::io::Error),
+    NotImplemented,
+}

server/texture-sync-server/src/server_state.rs

@@ -8,6 +8,7 @@ use crate::protocol::*;
 
 use std::sync::Arc;
 
+#[derive(Clone)]
 pub struct ServerState {
     // ...
 }