Repository: AlexPikalov/cdrs (language: rust)

Issue:
Rid of frame exchange logic duplication
The expected outcome of this feature is reducing of frame exchange logic duplication. This logic is common and is just duplicated in following trait methods:

* `QueryExecutor:: query_with_params_tw`,
* `PrepareExecutor::prepare_tw`,
* `BatchExecutor:: batch_with_params_tw`
* `ExecExecutor::exec_with_params_tw`

The logic itself:

```rust
    let mut flags = vec![];
    if with_tracing {
      flags.push(Flag::Tracing);
    }
    if with_warnings {
      flags.push(Flag::Warning);
    }

    // here is a difference - different frames are used for different executors
    let options_frame = Frame::new_req_execute(prepared, query_parameters, flags).into_cbytes();
    let ref compression = self.get_compressor();

    self
      .get_connection()
      .ok_or(error::Error::from("Unable to get transport"))
      .and_then(|transport_cell| {
        let write_res = transport_cell
          .borrow_mut()
          .write(options_frame.as_slice())
          .map_err(error::Error::from);
        write_res.map(|_| transport_cell)
      })
      .and_then(|transport_cell| from_connection(&transport_cell, compression))
```

It should be extracted in some reusable form such as trait method or macro and then be integrated into those traits that were mentioned.


Discussion / hints:

Fix patch:
diff --git a/src/query/batch_executor.rs b/src/query/batch_executor.rs
index dcb62cb..722cff2 100644
--- a/src/query/batch_executor.rs
+++ b/src/query/batch_executor.rs
@@ -3,12 +3,13 @@ use std::cell::RefCell;
 
 use cluster::{GetCompressor, GetConnection};
 use error;
-use frame::parser::from_connection;
 use frame::traits::IntoBytes;
-use frame::{Flag, Frame};
+use frame::Frame;
 use query::batch_query_builder::QueryBatch;
 use transport::CDRSTransport;
 
+use super::utils::{prepare_flags, send_frame};
+
 pub trait BatchExecutor<
   T: CDRSTransport + 'static,
   M: r2d2::ManageConnection<Connection = RefCell<T>, Error = error::Error> + Sized,
@@ -19,34 +20,21 @@ pub trait BatchExecutor<
     batch: QueryBatch,
     with_tracing: bool,
     with_warnings: bool,
-  ) -> error::Result<Frame> {
-    let mut flags = vec![];
-
-    if with_tracing {
-      flags.push(Flag::Tracing);
-    }
-
-    if with_warnings {
-      flags.push(Flag::Warning);
-    }
+  ) -> error::Result<Frame>
+  where
+    Self: Sized,
+  {
+    let flags = prepare_flags(with_tracing, with_warnings);
 
     let query_frame = Frame::new_req_batch(batch, flags).into_cbytes();
-    let ref compression = self.get_compressor();
 
-    self
-      .get_connection()
-      .ok_or(error::Error::from("Unable to get transport"))
-      .and_then(|transport_cell| {
-        let write_res = transport_cell
-          .borrow_mut()
-          .write(query_frame.as_slice())
-          .map_err(error::Error::from);
-        write_res.map(|_| transport_cell)
-      })
-      .and_then(|transport_cell| from_connection(&transport_cell, compression))
+    send_frame(self, query_frame)
   }
 
-  fn batch_with_params(&self, batch: QueryBatch) -> error::Result<Frame> {
+  fn batch_with_params(&self, batch: QueryBatch) -> error::Result<Frame>
+  where
+    Self: Sized,
+  {
     self.batch_with_params_tw(batch, false, false)
   }
 }
diff --git a/src/query/exec_executor.rs b/src/query/exec_executor.rs
index 0d19003..5710b11 100644
--- a/src/query/exec_executor.rs
+++ b/src/query/exec_executor.rs
@@ -3,12 +3,13 @@ use std::cell::RefCell;
 
 use cluster::{GetCompressor, GetConnection};
 use error;
-use frame::parser::from_connection;
-use frame::{Flag, Frame, IntoBytes};
+use frame::{Frame, IntoBytes};
 use query::{QueryParams, QueryParamsBuilder, QueryValues};
 use transport::CDRSTransport;
 use types::CBytesShort;
 
+use super::utils::{prepare_flags, send_frame};
+
 pub type PreparedQuery = CBytesShort;
 
 pub trait ExecExecutor<
@@ -22,36 +23,24 @@ pub trait ExecExecutor<
     query_parameters: QueryParams,
     with_tracing: bool,
     with_warnings: bool,
-  ) -> error::Result<Frame> {
-    let mut flags = vec![];
-    if with_tracing {
-      flags.push(Flag::Tracing);
-    }
-    if with_warnings {
-      flags.push(Flag::Warning);
-    }
-
+  ) -> error::Result<Frame>
+  where
+    Self: Sized,
+  {
+    let flags = prepare_flags(with_tracing, with_warnings);
     let options_frame = Frame::new_req_execute(prepared, query_parameters, flags).into_cbytes();
-    let ref compression = self.get_compressor();
 
-    self
-      .get_connection()
-      .ok_or(error::Error::from("Unable to get transport"))
-      .and_then(|transport_cell| {
-        let write_res = transport_cell
-          .borrow_mut()
-          .write(options_frame.as_slice())
-          .map_err(error::Error::from);
-        write_res.map(|_| transport_cell)
-      })
-      .and_then(|transport_cell| from_connection(&transport_cell, compression))
+    send_frame(self, options_frame)
   }
 
   fn exec_with_params(
     &self,
     prepared: &PreparedQuery,
     query_parameters: QueryParams,
-  ) -> error::Result<Frame> {
+  ) -> error::Result<Frame>
+  where
+    Self: Sized,
+  {
     self.exec_with_params_tw(prepared, query_parameters, false, false)
   }
 
@@ -61,7 +50,10 @@ pub trait ExecExecutor<
     values: V,
     with_tracing: bool,
     with_warnings: bool,
-  ) -> error::Result<Frame> {
+  ) -> error::Result<Frame>
+  where
+    Self: Sized,
+  {
     let query_params_builder = QueryParamsBuilder::new();
     let query_params = query_params_builder.values(values.into()).finalize();
     self.exec_with_params_tw(prepared, query_params, with_tracing, with_warnings)
@@ -71,7 +63,10 @@ pub trait ExecExecutor<
     &self,
     prepared: &PreparedQuery,
     values: V,
-  ) -> error::Result<Frame> {
+  ) -> error::Result<Frame>
+  where
+    Self: Sized,
+  {
     self.exec_with_values_tw(prepared, values, false, false)
   }
 
@@ -80,12 +75,18 @@ pub trait ExecExecutor<
     prepared: &PreparedQuery,
     with_tracing: bool,
     with_warnings: bool,
-  ) -> error::Result<Frame> {
+  ) -> error::Result<Frame>
+  where
+    Self: Sized,
+  {
     let query_params = QueryParamsBuilder::new().finalize();
     self.exec_with_params_tw(prepared, query_params, with_tracing, with_warnings)
   }
 
-  fn exec(&mut self, prepared: &PreparedQuery) -> error::Result<Frame> {
+  fn exec(&mut self, prepared: &PreparedQuery) -> error::Result<Frame>
+  where
+    Self: Sized,
+  {
     self.exec_tw(prepared, false, false)
   }
 }
diff --git a/src/query/mod.rs b/src/query/mod.rs
index c32744b..c275fab 100644
--- a/src/query/mod.rs
+++ b/src/query/mod.rs
@@ -8,6 +8,7 @@ mod query_flags;
 mod query_params;
 mod query_params_builder;
 mod query_values;
+mod utils;
 
 pub use query::batch_executor::BatchExecutor;
 pub use query::batch_query_builder::{BatchQueryBuilder, QueryBatch};
diff --git a/src/query/prepare_executor.rs b/src/query/prepare_executor.rs
index 60da28b..0d64385 100644
--- a/src/query/prepare_executor.rs
+++ b/src/query/prepare_executor.rs
@@ -3,11 +3,12 @@ use std::cell::RefCell;
 
 use cluster::{GetCompressor, GetConnection};
 use error;
-use frame::parser::from_connection;
-use frame::{Flag, Frame, IntoBytes};
+use frame::{Frame, IntoBytes};
 use transport::CDRSTransport;
 use types::CBytesShort;
 
+use super::utils::{prepare_flags, send_frame};
+
 pub type PreparedQuery = CBytesShort;
 
 pub trait PrepareExecutor<
@@ -23,29 +24,15 @@ pub trait PrepareExecutor<
     query: Q,
     with_tracing: bool,
     with_warnings: bool,
-  ) -> error::Result<PreparedQuery> {
-    let mut flags = vec![];
-    if with_tracing {
-      flags.push(Flag::Tracing);
-    }
-    if with_warnings {
-      flags.push(Flag::Warning);
-    }
+  ) -> error::Result<PreparedQuery>
+  where
+    Self: Sized,
+  {
+    let flags = prepare_flags(with_tracing, with_warnings);
 
     let query_frame = Frame::new_req_prepare(query.to_string(), flags).into_cbytes();
-    let ref compression = self.get_compressor();
 
-    self
-      .get_connection()
-      .ok_or(error::Error::from("Unable to get transport"))
-      .and_then(|transport_cell| {
-        let write_res = transport_cell
-          .borrow_mut()
-          .write(query_frame.as_slice())
-          .map_err(error::Error::from);
-        write_res.map(|_| transport_cell)
-      })
-      .and_then(|transport_cell| from_connection(&transport_cell, compression))
+    send_frame(self, query_frame)
       .and_then(|response| response.get_body())
       .and_then(|body| {
         Ok(
@@ -58,7 +45,10 @@ pub trait PrepareExecutor<
   }
 
   /// It prepares query without additional tracing information and warnings.
-  fn prepare<Q: ToString>(&self, query: Q) -> error::Result<PreparedQuery> {
+  fn prepare<Q: ToString>(&self, query: Q) -> error::Result<PreparedQuery>
+  where
+    Self: Sized,
+  {
     self.prepare_tw(query, false, false)
   }
 }
diff --git a/src/query/query_executor.rs b/src/query/query_executor.rs
index 6f11967..2473814 100644
--- a/src/query/query_executor.rs
+++ b/src/query/query_executor.rs
@@ -3,11 +3,12 @@ use std::cell::RefCell;
 
 use cluster::{GetCompressor, GetConnection};
 use error;
-use frame::parser::from_connection;
-use frame::{Flag, Frame, IntoBytes};
+use frame::{Frame, IntoBytes};
 use query::{Query, QueryParams, QueryParamsBuilder, QueryValues};
 use transport::CDRSTransport;
 
+use super::utils::{prepare_flags, send_frame};
+
 pub trait QueryExecutor<
   T: CDRSTransport + 'static,
   M: r2d2::ManageConnection<Connection = RefCell<T>, Error = error::Error> + Sized,
@@ -19,41 +20,28 @@ pub trait QueryExecutor<
     query_params: QueryParams,
     with_tracing: bool,
     with_warnings: bool,
-  ) -> error::Result<Frame> {
+  ) -> error::Result<Frame>
+  where
+    Self: Sized,
+  {
     let query = Query {
       query: query.to_string(),
       params: query_params,
     };
 
-    let mut flags = vec![];
-
-    if with_tracing {
-      flags.push(Flag::Tracing);
-    }
-
-    if with_warnings {
-      flags.push(Flag::Warning);
-    }
+    let flags = prepare_flags(with_tracing, with_warnings);
 
     let query_frame = Frame::new_query(query, flags).into_cbytes();
-    let ref compression = self.get_compressor();
 
-    self
-      .get_connection()
-      .ok_or(error::Error::from("Unable to get transport"))
-      .and_then(|transport_cell| {
-        let write_res = transport_cell
-          .borrow_mut()
-          .write(query_frame.as_slice())
-          .map_err(error::Error::from);
-        write_res.map(|_| transport_cell)
-      })
-      .and_then(|transport_cell| from_connection(&transport_cell, compression))
+    send_frame(self, query_frame)
   }
 
   /// Executes a query with default parameters:
   /// * TDB
-  fn query<Q: ToString>(&self, query: Q) -> error::Result<Frame> {
+  fn query<Q: ToString>(&self, query: Q) -> error::Result<Frame>
+  where
+    Self: Sized,
+  {
     self.query_tw(query, false, false)
   }
 
@@ -64,7 +52,10 @@ pub trait QueryExecutor<
     query: Q,
     with_tracing: bool,
     with_warnings: bool,
-  ) -> error::Result<Frame> {
+  ) -> error::Result<Frame>
+  where
+    Self: Sized,
+  {
     let query_params = QueryParamsBuilder::new().finalize();
     self.query_with_params_tw(query, query_params, with_tracing, with_warnings)
   }
@@ -74,7 +65,10 @@ pub trait QueryExecutor<
     &self,
     query: Q,
     values: V,
-  ) -> error::Result<Frame> {
+  ) -> error::Result<Frame>
+  where
+    Self: Sized,
+  {
     self.query_with_values_tw(query, values, false, false)
   }
 
@@ -86,7 +80,10 @@ pub trait QueryExecutor<
     values: V,
     with_tracing: bool,
     with_warnings: bool,
-  ) -> error::Result<Frame> {
+  ) -> error::Result<Frame>
+  where
+    Self: Sized,
+  {
     let query_params_builder = QueryParamsBuilder::new();
     let query_params = query_params_builder.values(values.into()).finalize();
     self.query_with_params_tw(query, query_params, with_tracing, with_warnings)
@@ -97,7 +94,10 @@ pub trait QueryExecutor<
     &self,
     query: Q,
     query_params: QueryParams,
-  ) -> error::Result<Frame> {
+  ) -> error::Result<Frame>
+  where
+    Self: Sized,
+  {
     self.query_with_params_tw(query, query_params, false, false)
   }
 }
diff --git a/src/query/query_params_builder.rs b/src/query/query_params_builder.rs
index 9a07eff..47aa0c6 100644
--- a/src/query/query_params_builder.rs
+++ b/src/query/query_params_builder.rs
@@ -1,6 +1,6 @@
+use super::{QueryFlags, QueryParams, QueryValues};
 use consistency::Consistency;
 use types::CBytes;
-use super::{QueryFlags, QueryParams, QueryValues};
 
 #[derive(Debug, Default)]
 pub struct QueryParamsBuilder {
@@ -38,12 +38,12 @@ impl QueryParamsBuilder {
     self.with_names = Some(with_names);
     self.values = Some(values);
     self.flags = self.flags.or(Some(vec![])).map(|mut flags| {
-                                                   flags.push(QueryFlags::Value);
-                                                   if with_names {
-                                                     flags.push(QueryFlags::WithNamesForValues);
-                                                   }
-                                                   flags
-                                                 });
+      flags.push(QueryFlags::Value);
+      if with_names {
+        flags.push(QueryFlags::WithNamesForValues);
+      }
+      flags
+    });
 
     self
   }
@@ -56,9 +56,9 @@ impl QueryParamsBuilder {
   pub fn page_size(mut self, size: i32) -> Self {
     self.page_size = Some(size);
     self.flags = self.flags.or(Some(vec![])).map(|mut flags| {
-                                                   flags.push(QueryFlags::PageSize);
-                                                   flags
-                                                 });
+      flags.push(QueryFlags::PageSize);
+      flags
+    });
 
     self
   }
@@ -68,9 +68,9 @@ impl QueryParamsBuilder {
   pub fn paging_state(mut self, state: CBytes) -> Self {
     self.paging_state = Some(state);
     self.flags = self.flags.or(Some(vec![])).map(|mut flags| {
-                                                   flags.push(QueryFlags::WithPagingState);
-                                                   flags
-                                                 });
+      flags.push(QueryFlags::WithPagingState);
+      flags
+    });
 
     self
   }
@@ -83,13 +83,15 @@ impl QueryParamsBuilder {
 
   /// Finalizes query building process and returns query itself
   pub fn finalize(self) -> QueryParams {
-    QueryParams { consistency: self.consistency,
-                  flags: self.flags.unwrap_or(vec![]),
-                  values: self.values,
-                  with_names: self.with_names,
-                  page_size: self.page_size,
-                  paging_state: self.paging_state,
-                  serial_consistency: self.serial_consistency,
-                  timestamp: self.timestamp, }
+    QueryParams {
+      consistency: self.consistency,
+      flags: self.flags.unwrap_or(vec![]),
+      values: self.values,
+      with_names: self.with_names,
+      page_size: self.page_size,
+      paging_state: self.paging_state,
+      serial_consistency: self.serial_consistency,
+      timestamp: self.timestamp,
+    }
   }
 }
diff --git a/src/query/utils.rs b/src/query/utils.rs
new file mode 100644
index 0000000..ac344d7
--- /dev/null
+++ b/src/query/utils.rs
@@ -0,0 +1,57 @@
+use std::cell::RefCell;
+
+use cluster::{GetCompressor, GetConnection};
+use error;
+use frame::parser::from_connection;
+use frame::{Flag, Frame};
+use transport::CDRSTransport;
+
+pub fn prepare_flags(with_tracing: bool, with_warnings: bool) -> Vec<Flag> {
+  let mut flags = vec![];
+
+  if with_tracing {
+    flags.push(Flag::Tracing);
+  }
+
+  if with_warnings {
+    flags.push(Flag::Warning);
+  }
+
+  flags
+}
+
+pub fn send_frame<S, T, M>(sender: &S, frame_bytes: Vec<u8>) -> error::Result<Frame>
+where
+  S: GetConnection<T, M> + GetCompressor<'static> + Sized,
+  T: CDRSTransport + 'static,
+  M: r2d2::ManageConnection<Connection = RefCell<T>, Error = error::Error> + Sized,
+{
+  let ref compression = sender.get_compressor();
+
+  sender
+    .get_connection()
+    .ok_or(error::Error::from("Unable to get transport"))
+    .and_then(|transport_cell| {
+      let write_res = transport_cell
+        .borrow_mut()
+        .write(frame_bytes.as_slice())
+        .map_err(error::Error::from);
+      write_res.map(|_| transport_cell)
+    })
+    .and_then(|transport_cell| from_connection(&transport_cell, compression))
+}
+
+#[cfg(test)]
+mod test {
+  use super::*;
+
+  #[test]
+  fn prepare_flags_test() {
+    assert_eq!(prepare_flags(true, false), vec![Flag::Tracing]);
+    assert_eq!(prepare_flags(false, true), vec![Flag::Warning]);
+    assert_eq!(
+      prepare_flags(true, true),
+      vec![Flag::Tracing, Flag::Warning]
+    );
+  }
+}


Test patch: