Added GetLog command and KineticError implements Error

Author: Ignacio Corderi

Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 
 name = "kinetic-rust"
-version = "0.0.8"
+version = "0.0.9"
 authors = ["Ignacio Corderi <[email protected]>"]
 homepage = "https://github.com/icorderi/kinetic-rust/"
 repository = "https://github.com/icorderi/kinetic-rust/"

src/kinetic-bench/main.rs

@@ -40,6 +40,7 @@ static USAGE: &'static str = "
 Kinetic from Rust!
 
 Usage: kinetic-bench write <target> [<count>]
+       kinetic-bench config <target>
        kinetic-bench [options]
 
 Options:
@@ -50,9 +51,10 @@ Options:
 #[deriving(Decodable, Show)]
 struct Args {
    cmd_write: Option<WriteArgs>,
-   cmd_read: Option<ReadArgs>,
+   cmd_read: Option<TargetArgs>,
+   cmd_config: Option<TargetArgs>,
    flag_help: bool,
-   flag_version: bool
+   flag_version: bool,
 }
 
 #[deriving(Decodable, Show)]
@@ -62,7 +64,7 @@ struct WriteArgs{
 }
 
 #[deriving(Decodable, Show)]
-struct ReadArgs{
+struct TargetArgs{
     arg_target: String
 }
 
@@ -88,6 +90,12 @@ fn main() {
         return;
     }
 
+    if args.cmd_config.is_some() {
+        let c = kinetic::Client::connect(format!("{}:8123", args.cmd_config.unwrap().arg_target).as_slice()).unwrap();
+        println!("{}", c.get_config());
+        return;
+    }
+
     if args.cmd_write.is_none() {
         println!("{}", USAGE);
         return;

src/kinetic/client.rs

@@ -41,7 +41,9 @@ type KineticResponse = (::proto::Message, ::proto::Command, ::std::vec::Vec<u8>)
 #[experimental]
 struct KineticChannel {
     stream: io::TcpStream,
-    writer_tx: ::std::comm::SyncSender<KineticCommand>
+    writer_tx: ::std::comm::SyncSender<KineticCommand>,
+    pub configuration: ::proto::command::log::Configuration,
+    pub limits: ::proto::command::log::Limits,
 }
 
 #[experimental]
@@ -64,6 +66,9 @@ impl KineticChannel {
         // Handshake
         let (_, cmd, _) = ::network::recv(&mut s).unwrap();
         let connection_id = cmd.get_header().get_connectionID();
+        let mut the_log = cmd.unwrap_body().unwrap_getLog();
+        let configuration = the_log.take_configuration();
+        let limits = the_log.take_limits();
 
         // Other state like pending requests...
         let pending_mutex = Arc::new(Mutex::new(collections::HashMap::with_capacity(max_pending)));
@@ -128,7 +133,8 @@ impl KineticChannel {
                 hmac.input(&buffer);
                 hmac.input(cmd_bytes.as_slice());
 
-                auth.set_hmac(hmac.result().code().to_vec()); // TODO: Code is backed by a Vec, we should have a method to get it.
+                // TODO: Code is backed by a Vec, we should have an unwrap() method to get it.
+                auth.set_hmac(hmac.result().code().to_vec());
                 msg.set_hmacAuth(auth);
 
                 msg.set_commandBytes(cmd_bytes);
@@ -137,13 +143,15 @@ impl KineticChannel {
                     let mut pending = pending_mutex.lock();
                     pending.insert(seq, callback);
                 }
+
                 let value = value.unwrap_or(vec::Vec::new());
                 ::network::send(&mut writer, &msg, value.as_slice()).unwrap();
                 seq += 1;
             }
         });
 
-        Ok(KineticChannel { stream: s, writer_tx: w_tx})
+        Ok(KineticChannel { stream: s, writer_tx: w_tx,
+                            configuration: configuration, limits: limits })
     }
 
     #[experimental]
@@ -179,6 +187,16 @@ impl Client {
                      cluster_version: 0 })
     }
 
+    // FIXME : Solve this without cloning
+    pub fn get_config(&self) -> ::proto::command::log::Configuration {
+        self.channel.configuration.clone()
+    }
+
+    // FIXME : Solve this without cloning
+    pub fn get_limits(&self) -> ::proto::command::log::Limits {
+        self.channel.limits.clone()
+    }
+
     #[inline]
     fn send_raw<R : Response, C: Command<R>> (&self, cmd: C) -> Receiver<KineticResponse> {
         // build specific command

src/kinetic/commands.rs

@@ -28,12 +28,12 @@ use core::Command;
 use std::vec;
 
 /// Requests the value stored with the given key
-#[experimental]
+#[unstable]
 pub struct Get {
     pub key: vec::Vec<u8>
 }
 
-#[experimental]
+#[unstable]
 impl Command<::responses::GetResponse> for Get {
 
     fn build_proto(self) -> (::proto::Command, Option<vec::Vec<u8>>) {
@@ -59,13 +59,13 @@ impl Command<::responses::GetResponse> for Get {
 }
 
 /// Stores the value asociated with the key
-#[experimental]
+#[unstable]
 pub struct Put {
     pub key: vec::Vec<u8>,
     pub value: vec::Vec<u8>
 }
 
-#[experimental]
+#[unstable]
 impl Command<()> for Put {
 
     fn build_proto(self) -> (::proto::Command, Option<vec::Vec<u8>>) {
@@ -93,3 +93,35 @@ impl Command<()> for Put {
     }
 
 }
+
+/// Stores the value asociated with the key
+#[unstable]
+pub struct GetLog {
+    // FIXME: The operation actually accepts a **set** of types
+    pub log_type: ::proto::command::LogType
+}
+
+#[unstable]
+impl Command<::proto::command::GetLog> for GetLog {
+
+    fn build_proto(self) -> (::proto::Command, Option<vec::Vec<u8>>) {
+        let mut cmd = ::proto::Command::new();
+        let mut header = ::proto::command::Header::new();
+
+        // Set command type
+        header.set_messageType(::proto::command::MessageType::GETLOG);
+        cmd.set_header(header);
+
+        // Build the actual command
+        let mut get_log = ::proto::command::GetLog::new();
+        get_log.set_types(vec![self.log_type]);
+
+        // Fill the body
+        let mut body = ::proto::command::Body::new();
+        body.set_getLog(get_log);
+        cmd.set_body(body);
+
+        (cmd, None) // return command
+    }
+
+}

src/kinetic/error.rs

@@ -22,19 +22,51 @@
 
 #![stable]
 
+use std::error::Error;
 use std::error::FromError;
-use protobuf::error::ProtobufError;
-use proto::StatusCode;
 use std::io::IoError;
+use protobuf::error::ProtobufError;
+use proto::command::Status;
 
-
-#[deriving(Show,Eq,PartialEq)]
 #[stable]
+#[deriving(Show)]
 pub enum KineticError {
     IoError(IoError),
     ProtobufError(ProtobufError),
     InvalidMagicNumber,
-    RemoteError(StatusCode, String)
+    RemoteError(Status)
+}
+
+impl Error for KineticError {
+    fn description(&self) -> &str {
+        match *self {
+            KineticError::IoError(_) => "An I/O error occurred",
+            KineticError::ProtobufError(_) => "There was an error with the protobuf library",
+            KineticError::InvalidMagicNumber => "Invalid magic number received",
+            KineticError::RemoteError(ref status) =>
+                format!("{}: {}", status.get_code(), status.get_statusMessage()).as_slice().clone(),
+        }
+    }
+
+    fn detail(&self) -> Option<String> {
+        match *self {
+            KineticError::IoError(ref err) => Some(err.description().to_string()),
+            KineticError::ProtobufError(ref err) => Some(err.description().to_string()),
+            KineticError::RemoteError(ref status) =>
+                if status.has_detailedMessage() {
+                    String::from_utf8(status.get_detailedMessage().to_vec()).ok() }
+                else { None },
+            _ => None,
+        }
+    }
+
+    fn cause(&self) -> Option<&Error> {
+        match *self {
+            KineticError::IoError(ref err) => Some(err as &Error),
+            KineticError::ProtobufError(ref err) => Some(err as &Error),
+            _ => None,
+        }
+    }
 }
 
 #[stable]

src/kinetic/proto/mod.rs

@@ -30,20 +30,18 @@ pub use proto::raw::Message;
 
 mod raw;
 
+#[stable]
 pub mod message {
 
-    #![stable]
-
     pub use proto::raw::Message_AuthType as AuthType;
     pub use proto::raw::Message_HMACauth as HmacAuth;
     pub use proto::raw::Message_PINauth as PinAuth;
 
 }
 
+#[stable]
 pub mod command {
 
-    #![stable]
-
     pub use proto::raw::Command_Header as Header;
     pub use proto::raw::Command_MessageType as MessageType;
     pub use proto::raw::Command_Body as Body;
@@ -53,6 +51,16 @@ pub mod command {
     pub use proto::raw::Command_Algorithm as Algorithm;
     pub use proto::raw::Command_Synchronization as Synchronization;
 
+    pub use proto::raw::Command_GetLog as GetLog;
+    pub use proto::raw::Command_GetLog_Type as LogType;
+
+    #[unstable]
+    pub mod log {
+
+        pub use proto::raw::Command_GetLog_Configuration as Configuration;
+        pub use proto::raw::Command_GetLog_Limits as Limits;
+
+    }
 }
 
 /// Returns the version of the Kinetic Protocol