meta_data now handles ls request and simple data node registration

.gitignore

@@ -7,3 +7,4 @@
 *.tar.gz
 *.db
 dfs_skel/
+venv/

sm

@@ -0,0 +1 @@
+echo $1 | nc -v -N localhost 6770

src/bin/copy.rs

@@ -1,7 +1,13 @@
 extern crate a03;
+extern crate serde;
+extern crate serde_json;
+#[macro_use]
+extern crate serde_derive;
 
 use a03::*;
+use std::net::{TcpListener, TcpStream, Shutdown};
+use std::thread;
+use std::io::Read;
+use std::io::Write;
 
-fn main() {
+fn main() {}
-    println!("Hello, world!");
-}

src/bin/data_node.rs

@@ -1,25 +1,28 @@
 extern crate a03;
-
-use a03::*;
 extern crate serde;
 extern crate serde_json;
 #[macro_use]
 extern crate serde_derive;
 
-use std::net::{TcpListener,TcpStream,Shutdown};
+use a03::*;
+use std::net::{TcpStream, Shutdown};
 use std::thread;
 use std::io::Read;
 use std::io::Write;
 
 fn main() {
     let mut stream = TcpStream::connect("localhost:6770").unwrap();
-    let writer = serde_json::to_writer(
+    serde_json::to_writer(
         &mut stream,
-        &TestObj { message : String::from("I can't think of something clever")})
+        &Packet {
+            p_type: PacketType::RegisterNode,
+            json: Some(serde_json::to_string(
+                &RegisterNode { ip: String::from("localhost"), port: 6770 }).unwrap()),
+        })
         .unwrap();
-    println!("Message sent!");
+    println!("Registered myself");
     stream.flush().unwrap();
-//    stream.shutdown(Shutdown::Write);
+    stream.shutdown(Shutdown::Write).unwrap();
-    let test_obj : TestObj = serde_json::from_reader(&mut stream).unwrap();
+    let result: Packet = serde_json::from_reader(&mut stream).unwrap();
-    println!("Received message back! {}", test_obj.message);
+    println!("{:?}", result);
 }

src/bin/ls.rs

@@ -1,3 +1,31 @@
+extern crate a03;
+extern crate serde;
+extern crate serde_json;
+#[macro_use]
+extern crate serde_derive;
+
+use a03::*;
+use std::net::{TcpListener, TcpStream, Shutdown};
+use std::borrow::Cow;
+use std::thread;
+use std::io::Read;
+use std::io::Write;
+
 fn main() {
-    println!("Hello, world!");
+    let mut stream = TcpStream::connect("localhost:6770").unwrap();
+    serde_json::to_writer(
+        &mut stream,
+        &Packet {
+            p_type: PacketType::ListFiles,
+            json: None,
+        })
+        .unwrap();
+    println!("Message sent!");
+    stream.flush().unwrap();
+    stream.shutdown(Shutdown::Write).unwrap();
+    let files: FilePaths = serde_json::from_reader(&mut stream).unwrap();
+    for path in files.paths.iter() {
+        println!("Path: {}", path);
     }
+}
+

src/bin/meta_data.rs

@@ -8,47 +8,74 @@ extern crate serde_derive;
 use a03::*;
 use rusqlite::types::ToSql;
 use rusqlite::{Connection, NO_PARAMS};
-use std::net::{TcpListener,TcpStream};
+use std::borrow::Cow;
-use std::thread;
 use std::io::Read;
+use std::io::Write;
+use std::net::{Shutdown, TcpListener, TcpStream};
+use std::thread;
 
 fn main() {
+    let mut data_nodes: Vec<DataNode> = Vec::new();
+    let mut file_list: Vec<String> = Vec::new();
+
+    file_list.push(String::from("/"));
+    file_list.push(String::from("/home"));
+    file_list.push(String::from("/home/joe"));
+    file_list.push(String::from("/bin"));
     let listener = TcpListener::bind("localhost:6770").unwrap();
-    println!("Binding!");
     for stream in listener.incoming() {
         let mut stream = stream.unwrap();
-        println!("Got here!");
+        match serde_json::from_reader(&mut stream) {
-        let test_obj : TestObj = serde_json::from_reader(&mut stream).unwrap();
+            Ok(packet @ Packet { .. }) => match packet.p_type {
+                PacketType::ListFiles => list(&mut stream, &file_list[..]),
+                PacketType::PutFiles => put(&mut stream, &packet.json.unwrap(), &mut file_list),
+                PacketType::RegisterNode =>
+                    register_node(&mut stream, &packet.json.unwrap(), &mut data_nodes),
+                _ => (),
+            },
+            Err(e) => println!("Error parsing json {}", e.to_string()),
+        };
+        stream.flush().unwrap();
+    }
+}
+
+fn list(stream: &mut TcpStream, files: &[String]) {
     match serde_json::to_writer(
-            &mut stream,
+        stream,
-            &TestObj { message : String::from("I got it dude") })
+        &FilePaths {
-        {
+            paths: Cow::from(files),
-            Ok(_) => println!("{}", test_obj.message),
+        },
+    ) {
+        Ok(_) => println!("{}", "Sent file paths"),
         Err(e) => println!("{}", e),
     };
 }
+
+fn put(stream: &mut TcpStream, json: &String, files: &mut Vec<String>) {
+    let files: PutFiles = serde_json::from_str(json).unwrap();
+    report_success(stream);
 }
 
-#[derive(Debug)]
+fn register_node(stream: &mut TcpStream, json: &String, data_nodes: &mut Vec<DataNode>) {
-struct DataNode {
+    let endpoint : RegisterNode = serde_json::from_str(json).unwrap();
-    id: i32,
+    data_nodes.push(DataNode { address: endpoint.ip, port: endpoint.port, id: 1 });
-    address: String,
+    for dn in data_nodes {
-    port: i32,
+        println!("{:?}", dn);
+    }
+    report_success(stream);
 }
 
-#[derive(Debug)]
+fn report_success(stream: &mut TcpStream) {
-struct INode {
+    match serde_json::to_writer(
-    id: i32,
+        stream,
-    name: String,
+        &Packet {
-    size: i32,
+            p_type: PacketType::Success,
-}
+            json: None,
-
+        },
-#[derive(Debug)]
+    ) {
-struct Block {
+        Ok(_) => println!("{}", "Success Registering Data Node"),
-    id : i32,
+        Err(e) => println!("{}", e),
-    file_id : i32,
+    };
-    node_id : i32,
-    c_id : String,
 }
 
 fn create_tables(conn: &Connection) {
@@ -96,14 +123,11 @@ fn check_node(conn: &Connection, address: &str, port: i32) -> DataNode {
     let mut stmt = conn
         .prepare("SELECT nid, address, port FROM dnode WHERE address=?1 AND port=?2")
         .unwrap();
-    stmt.query_row(
+    stmt.query_row(&[&address as &ToSql, &port], |row| DataNode {
-        &[&address as &ToSql, &port],
-        |row| DataNode {
         id: row.get(0),
         address: row.get(1),
         port: row.get(2),
-        })
+    }).unwrap()
-        .unwrap()
 }
 
 fn get_data_nodes(conn: &Connection) {
@@ -142,5 +166,4 @@ mod tests {
         assert_eq!(dnode.address, ip);
         assert_eq!(dnode.port, 65533);
     }
-
 }

src/lib.rs

@@ -3,8 +3,66 @@ extern crate serde_json;
 #[macro_use]
 extern crate serde_derive;
 
+use std::borrow::Cow;
+use std::net::Ipv4Addr;
+
 #[derive(Serialize, Deserialize, Debug)]
-pub struct TestObj {
+pub enum PacketType {
-    pub message : String
+    RegisterNode,
+    ListFiles,
+    PutFiles,
+    GetFiles,
+    AddDataBlocks,
+    Success,
+}
+
+#[derive(Serialize, Deserialize, Debug)]
+pub struct Packet {
+    pub p_type: PacketType,
+    pub json: Option<String>,
+}
+
+#[derive(Serialize, Deserialize, Debug)]
+pub struct FilePaths<'a> {
+    pub paths: Cow<'a, [String]>,
+}
+
+#[derive(Serialize, Deserialize, Debug)]
+pub struct RegisterNode {
+    pub ip: String,
+    pub port: u32,
+}
+
+#[derive(Serialize, Deserialize, Debug)]
+pub struct PutFiles {
+    pub files: Vec<String>
+}
+
+#[derive(Serialize, Deserialize, Debug)]
+pub struct GetFiles {}
+
+#[derive(Serialize, Deserialize, Debug)]
+pub struct AddDataBlocks {}
+
+#[derive(Debug)]
+pub struct DataNode {
+    pub id: u32,
+    pub address: String,
+    pub port: u32,
+}
+
+#[derive(Debug)]
+pub struct INode {
+    pub id: u32,
+    pub name: String,
+    pub size: u32,
+}
+
+#[derive(Debug)]
+pub struct Block {
+    pub id: u32,
+    pub file_id: u32,
+    pub node_id: u32,
+    pub c_id: String,
 }