Minor non-breaking refactoring

Author: benashford

src/client/mod.rs

@@ -64,8 +64,7 @@ mod test {
                 connection
                     .send(resp_array!["PING", "TEST"])
                     .map_err(|e| e.into())
-            })
-            .and_then(|connection| connection.take(1).collect());
+            }).and_then(|connection| connection.take(1).collect());
 
         let values = run_and_wait(connection).unwrap();
 
@@ -89,12 +88,11 @@ mod test {
                     .send_all(stream::iter_ok::<_, io::Error>(ops))
                     .map(|(sender, _)| sender)
                     .map_err(|e| e.into())
-            })
-            .and_then(|connection| connection.skip(1001).take(1).collect());
+            }).and_then(|connection| connection.skip(1001).take(1).collect());
         let values = run_and_wait(connection).unwrap();
         assert_eq!(values.len(), 1);
         let values = match &values[0] {
-            &resp::RespValue::Array(ref values) => values.clone(),
+            resp::RespValue::Array(ref values) => values.clone(),
             _ => panic!("Not an array"),
         };
         assert_eq!(values.len(), 1000);

src/client/paired.rs

@@ -185,7 +185,7 @@ pub fn paired_connect(
     // It ensures that a Tokio executor runs the future.  This function would work correctly
     // without it, if we could be sure this function was only called by other futures that were
     // executed within the Tokio executor, but we cannot guarantee that.
-    let addr = addr.clone();
+    let addr = *addr;
     future::lazy(move || {
         reconnect(
             |con: &mpsc::UnboundedSender<SendPayload>, act| {
@@ -210,7 +210,7 @@ pub fn paired_connect(
             },
         )
     }).map(|out_tx_c| PairedConnection { out_tx_c })
-        .map_err(|()| error::Error::EndOfStream)
+    .map_err(|()| error::Error::EndOfStream)
 }
 
 impl PairedConnection {
@@ -231,7 +231,7 @@ impl PairedConnection {
         T: resp::FromResp,
     {
         match &msg {
-            &resp::RespValue::Array(_) => (),
+            resp::RespValue::Array(_) => (),
             _ => {
                 return Either::B(future::err(error::internal(
                     "Command must be a RespValue::Array",

src/client/pubsub.rs

@@ -216,7 +216,7 @@ pub struct PubsubConnection {
 pub fn pubsub_connect(
     addr: &SocketAddr,
 ) -> impl Future<Item = PubsubConnection, Error = error::Error> + Send {
-    let addr = addr.clone();
+    let addr = *addr;
     future::lazy(move || {
         reconnect(
             |con: &mpsc::UnboundedSender<PubsubEvent>, act| {
@@ -240,7 +240,7 @@ pub fn pubsub_connect(
             },
         )
     }).map(|out_tx_c| PubsubConnection { out_tx_c })
-        .map_err(|()| error::Error::EndOfStream)
+    .map_err(|()| error::Error::EndOfStream)
 }
 
 impl PubsubConnection {

src/reconnect.rs

@@ -10,7 +10,7 @@
 
 use std::error as std_error;
 use std::sync::{Arc, RwLock};
-use std::time::{Duration, Instant};
+use std::time::Duration;
 
 use futures::{
     future::{self, Either},
@@ -19,7 +19,7 @@ use futures::{
 };
 
 use tokio_executor::{DefaultExecutor, Executor};
-use tokio_timer::Deadline;
+use tokio_timer::Timeout;
 
 #[derive(Debug)]
 pub(crate) enum ReconnectError {
@@ -125,9 +125,7 @@ where
 
         let (tx, rx) = oneshot::channel();
 
-        let deadline = Instant::now() + Duration::from_secs(30); // TODO - review and make configurable
-
-        let connect_f = Deadline::new(connect_f, deadline).then(move |t| {
+        let connect_f = Timeout::new(connect_f, Duration::from_secs(30)).then(move |t| {
             let mut state = reconnect.state.write().expect("Cannot obtain write lock");
             let result = match *state {
                 NotConnected | Connecting => match t {

src/resp.rs

@@ -46,7 +46,7 @@ pub enum RespValue {
 }
 
 impl RespValue {
-    fn to_result(self) -> Result<RespValue, Error> {
+    fn into_result(self) -> Result<RespValue, Error> {
         match self {
             RespValue::Error(string) => Err(Error::Remote(string)),
             x => Ok(x),
@@ -78,7 +78,7 @@ pub trait FromResp: Sized {
     /// Return a `Result` containing either `Self` or `Error`.  Errors can occur due to either: a) the particular
     /// `RespValue` being incompatible with the required type, or b) a remote Redis error occuring.
     fn from_resp(resp: RespValue) -> Result<Self, Error> {
-        Self::from_resp_int(resp.to_result()?)
+        Self::from_resp_int(resp.into_result()?)
     }
 
     fn from_resp_int(resp: RespValue) -> Result<Self, Error>;
@@ -442,7 +442,7 @@ fn parse_error(message: String) -> Error {
 /// two bytes will not be returned)
 ///
 /// TODO - rename this function potentially, it's used for simple integers too
-fn scan_integer<'a>(buf: &'a mut BytesMut, idx: usize) -> Result<Option<(usize, &'a [u8])>, Error> {
+fn scan_integer(buf: &mut BytesMut, idx: usize) -> Result<Option<(usize, &[u8])>, Error> {
     let length = buf.len();
     let mut at_end = false;
     let mut pos = idx;
@@ -621,7 +621,7 @@ mod tests {
 
     #[test]
     fn test_bulk_string() {
-        let resp_object = RespValue::BulkString("THISISATEST".as_bytes().to_vec());
+        let resp_object = RespValue::BulkString(b"THISISATEST".to_vec());
         let mut bytes = BytesMut::new();
         let mut codec = RespCodec;
         codec.encode(resp_object.clone(), &mut bytes).unwrap();