Repository: kafka-rust/kafka-rust (language: rust)

Issue:
Consider formatting with rustfmt
It keeps the code nice and clean :) Currently, it would fix quite a lot of formatting:

```
 src/client/metadata.rs     |  20 +++----
 src/client/mod.rs          | 278 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++---------------------------------
 src/client/network.rs      |  61 +++++++++++---------
 src/client/state.rs        |  36 ++++++------
 src/client_internals.rs    |  11 ++--
 src/codecs.rs              |  75 ++++++++++++++-----------
 src/compression/gzip.rs    |   9 +--
 src/compression/snappy.rs  |  31 ++++++----
 src/consumer/assignment.rs |   5 +-
 src/consumer/builder.rs    |  16 ++++--
 src/consumer/mod.rs        | 113 ++++++++++++++++++++++++-------------
 src/consumer/state.rs      | 179 +++++++++++++++++++++++++++++++++++-----------------------
 src/error.rs               |  32 ++++++-----
 src/lib.rs                 |  11 ++--
 src/producer.rs            |  32 +++++------
 src/protocol/consumer.rs   | 162 ++++++++++++++++++++++++----------------------------
 src/protocol/fetch.rs      | 186 +++++++++++++++++++++++++++++++++++++-----------------------
 src/protocol/metadata.rs   |  60 ++++++++------------
 src/protocol/mod.rs        |  58 ++++++++++---------
 src/protocol/offset.rs     |  69 +++++++++--------------
 src/protocol/produce.rs    |  82 +++++++++++++--------------
 src/protocol/zreader.rs    |  17 +++---
 22 files changed, 877 insertions(+), 666 deletions(-)
```



Discussion / hints:

Fix patch:
diff --git a/rustfmt.toml b/rustfmt.toml
new file mode 100644
index 000000000..f1cba768d
--- /dev/null
+++ b/rustfmt.toml
@@ -0,0 +1,3 @@
+max_width = 100
+ideal_with = 80
+fn_call_width = 80
diff --git a/src/client/metadata.rs b/src/client/metadata.rs
index a9cf8d74e..c84449e9c 100644
--- a/src/client/metadata.rs
+++ b/src/client/metadata.rs
@@ -77,8 +77,8 @@ impl<'a> fmt::Debug for Topics<'a> {
 }
 
 impl<'a> IntoIterator for &'a Topics<'a> {
-    type Item=Topic<'a>;
-    type IntoIter=TopicIter<'a>;
+    type Item = Topic<'a>;
+    type IntoIter = TopicIter<'a>;
 
     fn into_iter(self) -> Self::IntoIter {
         self.iter()
@@ -86,8 +86,8 @@ impl<'a> IntoIterator for &'a Topics<'a> {
 }
 
 impl<'a> IntoIterator for Topics<'a> {
-    type Item=Topic<'a>;
-    type IntoIter=TopicIter<'a>;
+    type Item = Topic<'a>;
+    type IntoIter = TopicIter<'a>;
 
     fn into_iter(self) -> Self::IntoIter {
         TopicIter::new(self.state)
@@ -110,7 +110,7 @@ impl<'a> TopicIter<'a> {
 }
 
 impl<'a> Iterator for TopicIter<'a> {
-    type Item=Topic<'a>;
+    type Item = Topic<'a>;
 
     #[inline]
     fn next(&mut self) -> Option<Self::Item> {
@@ -150,10 +150,7 @@ impl<'a> Topic<'a> {
 
 impl<'a> fmt::Debug for Topic<'a> {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        write!(f,
-               "Topic {{ name: {}, partitions: {:?} }}",
-               self.name,
-               self.partitions())
+        write!(f, "Topic {{ name: {}, partitions: {:?} }}", self.name, self.partitions())
     }
 }
 
@@ -215,8 +212,8 @@ impl<'a> fmt::Debug for Partitions<'a> {
 }
 
 impl<'a> IntoIterator for &'a Partitions<'a> {
-    type Item=Partition<'a>;
-    type IntoIter=PartitionIter<'a>;
+    type Item = Partition<'a>;
+    type IntoIter = PartitionIter<'a>;
 
     fn into_iter(self) -> Self::IntoIter {
         self.iter()
@@ -224,8 +221,8 @@ impl<'a> IntoIterator for &'a Partitions<'a> {
 }
 
 impl<'a> IntoIterator for Partitions<'a> {
-    type Item=Partition<'a>;
-    type IntoIter=PartitionIter<'a>;
+    type Item = Partition<'a>;
+    type IntoIter = PartitionIter<'a>;
 
     fn into_iter(self) -> Self::IntoIter {
         PartitionIter::new(self.state, self.tp)
@@ -248,7 +245,7 @@ impl<'a> PartitionIter<'a> {
 }
 
 impl<'a> Iterator for PartitionIter<'a> {
-    type Item=Partition<'a>;
+    type Item = Partition<'a>;
 
     #[inline]
     fn next(&mut self) -> Option<Self::Item> {
@@ -302,9 +299,6 @@ impl<'a> Partition<'a> {
 
 impl<'a> fmt::Debug for Partition<'a> {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        write!(f,
-               "Partition {{ id: {}, leader: {:?} }}",
-               self.id(),
-               self.leader())
+        write!(f, "Partition {{ id: {}, leader: {:?} }}", self.id(), self.leader())
     }
 }
diff --git a/src/client/mod.rs b/src/client/mod.rs
index 481077c97..625149d20 100644
--- a/src/client/mod.rs
+++ b/src/client/mod.rs
@@ -132,8 +132,8 @@ pub enum FetchOffset {
     /// Receive the latest offset.
     Latest,
     /// Used to ask for all messages before a certain time (ms); unix
-    /// timestamp in milliseconds.  See also
-    /// https://cwiki.apache.org/confluence/display/KAFKA/Writing+a+Driver+for+Kafka#WritingaDriverforKafka-Offsets
+    /// timestamp in milliseconds.
+    /// See https://cwiki.apache.org/confluence/display/KAFKA/Writing+a+Driver+for+Kafka#WritingaDriverforKafka-Offsets
     ByTime(i64),
 }
 
@@ -192,7 +192,10 @@ pub struct FetchGroupOffset<'a> {
 impl<'a> FetchGroupOffset<'a> {
     #[inline]
     pub fn new(topic: &'a str, partition: i32) -> Self {
-        FetchGroupOffset { topic: topic, partition: partition }
+        FetchGroupOffset {
+            topic: topic,
+            partition: partition,
+        }
     }
 }
 
@@ -219,7 +222,11 @@ pub struct CommitOffset<'a> {
 
 impl<'a> CommitOffset<'a> {
     pub fn new(topic: &'a str, partition: i32, offset: i64) -> Self {
-        CommitOffset { topic: topic, partition: partition, offset: offset }
+        CommitOffset {
+            topic: topic,
+            partition: partition,
+            offset: offset,
+        }
     }
 }
 
@@ -279,8 +286,17 @@ impl<'a, 'b> AsRef<ProduceMessage<'a, 'b>> for ProduceMessage<'a, 'b> {
 impl<'a, 'b> ProduceMessage<'a, 'b> {
     /// A convenient constructor method to create a new produce
     /// message with all attributes specified.
-    pub fn new(topic: &'a str, partition: i32, key: Option<&'b [u8]>, value: Option<&'b [u8]>) -> Self {
-        ProduceMessage { key: key, value: value, topic: topic, partition: partition }
+    pub fn new(topic: &'a str,
+               partition: i32,
+               key: Option<&'b [u8]>,
+               value: Option<&'b [u8]>)
+               -> Self {
+        ProduceMessage {
+            key: key,
+            value: value,
+            topic: topic,
+            partition: partition,
+        }
     }
 }
 
@@ -310,7 +326,6 @@ pub struct FetchPartition<'a> {
 }
 
 impl<'a> FetchPartition<'a> {
-
     /// Creates a new "fetch messages" request structure with an
     /// unspecified `max_bytes`.
     pub fn new(topic: &'a str, partition: i32, offset: i64) -> Self {
@@ -338,7 +353,6 @@ impl<'a> AsRef<FetchPartition<'a>> for FetchPartition<'a> {
 // --------------------------------------------------------------------
 
 impl KafkaClient {
-
     /// Creates a new instance of KafkaClient. Before being able to
     /// successfully use the new client, you'll have to load metadata.
     ///
@@ -406,7 +420,8 @@ impl KafkaClient {
     /// `KafkaClient::load_metadata` methods, the creates
     /// [openssl](https://crates.io/crates/openssl)
     /// and [openssl_verify](https://crates.io/crates/openssl-verify),
-    /// as well as [Kafka's documentation](https://kafka.apache.org/documentation.html#security_ssl).
+    /// as well as
+    /// [Kafka's documentation](https://kafka.apache.org/documentation.html#security_ssl).
     #[cfg(feature = "security")]
     pub fn new_secure(hosts: Vec<String>, security: SecurityConfig) -> KafkaClient {
         KafkaClient {
@@ -731,17 +746,24 @@ impl KafkaClient {
 
     /// Fetches metadata about the specified topics from all of the
     /// underlying brokers (`self.hosts`).
-    fn fetch_metadata<T: AsRef<str>>(&mut self, topics: &[T]) -> Result<protocol::MetadataResponse> {
+    fn fetch_metadata<T: AsRef<str>>(&mut self,
+                                     topics: &[T])
+                                     -> Result<protocol::MetadataResponse> {
         let correlation = self.state.next_correlation_id();
         let now = Instant::now();
         for host in &self.config.hosts {
             debug!("fetch_metadata: requesting metadata from {}", host);
             match self.conn_pool.get_conn(host, now) {
                 Ok(conn) => {
-                    let req = protocol::MetadataRequest::new(correlation, &self.config.client_id, topics);
+                    let req =
+                        protocol::MetadataRequest::new(correlation, &self.config.client_id, topics);
                     match __send_request(conn, req) {
                         Ok(_) => return __get_response::<protocol::MetadataResponse>(conn),
-                        Err(e) => debug!("fetch_metadata: failed to request metadata from {}: {}", host, e),
+                        Err(e) => {
+                            debug!("fetch_metadata: failed to request metadata from {}: {}",
+                                   host,
+                                   e)
+                        }
                     }
                 }
                 Err(e) => {
@@ -767,9 +789,10 @@ impl KafkaClient {
     ///
     /// Returns a mapping of topic name to `PartitionOffset`s for each
     /// currently available partition of the corresponding topic.
-    pub fn fetch_offsets<T: AsRef<str>>(&mut self, topics: &[T], offset: FetchOffset)
-                                        -> Result<HashMap<String, Vec<PartitionOffset>>>
-    {
+    pub fn fetch_offsets<T: AsRef<str>>(&mut self,
+                                        topics: &[T],
+                                        offset: FetchOffset)
+                                        -> Result<HashMap<String, Vec<PartitionOffset>>> {
         let time = offset.to_kafka_value();
         let n_topics = topics.len();
 
@@ -782,9 +805,12 @@ impl KafkaClient {
         for topic in topics {
             let topic = topic.as_ref();
             if let Some(ps) = state.partitions_for(topic) {
-                for (id, host) in ps.iter().filter_map(|(id, p)| p.broker(&state).map(|b| (id, b.host()))) {
+                for (id, host) in ps.iter()
+                    .filter_map(|(id, p)| p.broker(&state).map(|b| (id, b.host()))) {
                     let entry = reqs.entry(host)
-                        .or_insert_with(|| protocol::OffsetRequest::new(correlation, &config.client_id));
+                        .or_insert_with(|| {
+                            protocol::OffsetRequest::new(correlation, &config.client_id)
+                        });
                     entry.add(topic, id, time);
                 }
             }
@@ -794,9 +820,12 @@ impl KafkaClient {
         let now = Instant::now();
         let mut res: HashMap<String, Vec<PartitionOffset>> = HashMap::with_capacity(n_topics);
         for (host, req) in reqs {
-            let resp = try!(__send_receive::<_, protocol::OffsetResponse>(&mut self.conn_pool, &host, now, req));
+            let resp = try!(__send_receive::<_, protocol::OffsetResponse>(&mut self.conn_pool,
+                                                                          &host,
+                                                                          now,
+                                                                          req));
             for tp in resp.topic_partitions {
-                let e = res.entry(tp.topic).or_insert(vec!());
+                let e = res.entry(tp.topic).or_insert(vec![]);
                 for p in tp.partitions {
                     e.push(p.into_offset());
                 }
@@ -819,13 +848,14 @@ impl KafkaClient {
     ///
     /// Returns a vector of the offset data for each available partition.
     /// See also `KafkaClient::fetch_offsets`.
-    pub fn fetch_topic_offsets<T: AsRef<str>>(&mut self, topic: T, offset: FetchOffset)
-                                              -> Result<Vec<PartitionOffset>>
-    {
+    pub fn fetch_topic_offsets<T: AsRef<str>>(&mut self,
+                                              topic: T,
+                                              offset: FetchOffset)
+                                              -> Result<Vec<PartitionOffset>> {
         let topic = topic.as_ref();
 
         let mut m = try!(self.fetch_offsets(&[topic], offset));
-        let offs = m.remove(topic).unwrap_or(vec!());
+        let offs = m.remove(topic).unwrap_or(vec![]);
         if offs.is_empty() {
             Err(Error::Kafka(KafkaCode::UnknownTopicOrPartition))
         } else {
@@ -901,7 +931,8 @@ impl KafkaClient {
     /// See also `kafka::consumer`.
     /// See also `KafkaClient::set_fetch_max_bytes_per_partition`.
     pub fn fetch_messages<'a, I, J>(&mut self, input: I) -> Result<Vec<fetch::Response>>
-        where J: AsRef<FetchPartition<'a>>, I: IntoIterator<Item=J>
+        where J: AsRef<FetchPartition<'a>>,
+              I: IntoIterator<Item = J>
     {
         let state = &mut self.state;
         let config = &self.config;
@@ -915,11 +946,14 @@ impl KafkaClient {
             if let Some(broker) = state.find_broker(inp.topic, inp.partition) {
                 reqs.entry(broker)
                     .or_insert_with(|| {
-                        protocol::FetchRequest::new(
-                            correlation, &config.client_id,
-                            config.fetch_max_wait_time, config.fetch_min_bytes)
+                        protocol::FetchRequest::new(correlation,
+                                                    &config.client_id,
+                                                    config.fetch_max_wait_time,
+                                                    config.fetch_min_bytes)
                     })
-                    .add(inp.topic, inp.partition, inp.offset,
+                    .add(inp.topic,
+                         inp.partition,
+                         inp.offset,
                          if inp.max_bytes > 0 {
                              inp.max_bytes
                          } else {
@@ -934,9 +968,9 @@ impl KafkaClient {
     /// Fetch messages from a single kafka partition.
     ///
     /// See `KafkaClient::fetch_messages`.
-    pub fn fetch_messages_for_partition<'a>(&mut self, req: &FetchPartition<'a>)
-                                            -> Result<Vec<fetch::Response>>
-    {
+    pub fn fetch_messages_for_partition<'a>(&mut self,
+                                            req: &FetchPartition<'a>)
+                                            -> Result<Vec<fetch::Response>> {
         self.fetch_messages(&[req])
     }
 
@@ -969,7 +1003,8 @@ impl KafkaClient {
     /// client.load_metadata_all().unwrap();
     /// let req = vec![ProduceMessage::new("my-topic", 0, None, Some("a".as_bytes())),
     ///                ProduceMessage::new("my-topic-2", 0, None, Some("b".as_bytes()))];
-    /// println!("{:?}", client.produce_messages(RequiredAcks::One, Duration::from_millis(100), req));
+    /// let resp = client.produce_messages(RequiredAcks::One, Duration::from_millis(100), req);
+    /// println!("{:?}", resp);
     /// ```
     ///
     /// The return value will contain a vector of topic, partition,
@@ -978,13 +1013,17 @@ impl KafkaClient {
     // XXX rework signaling an error; note that we need to either return the
     // messages which kafka failed to accept or otherwise tell the client about them
 
-    pub fn produce_messages<'a, 'b, I, J>(
-        &mut self, acks: RequiredAcks, ack_timeout: Duration, messages: I)
-        -> Result<Vec<TopicPartitionOffset>>
-        where J: AsRef<ProduceMessage<'a, 'b>>, I: IntoIterator<Item=J>
+    pub fn produce_messages<'a, 'b, I, J>(&mut self,
+                                          acks: RequiredAcks,
+                                          ack_timeout: Duration,
+                                          messages: I)
+                                          -> Result<Vec<TopicPartitionOffset>>
+        where J: AsRef<ProduceMessage<'a, 'b>>,
+              I: IntoIterator<Item = J>
     {
-        self.internal_produce_messages(
-            acks as i16, try!(protocol::to_millis_i32(ack_timeout)), messages)
+        self.internal_produce_messages(acks as i16,
+                                       try!(protocol::to_millis_i32(ack_timeout)),
+                                       messages)
     }
 
     /// Commit offset for a topic partitions on behalf of a consumer group.
@@ -1009,11 +1048,13 @@ impl KafkaClient {
     /// process or at much later point in time to resume comusing the
     /// topic partitions as of these offsets.
     pub fn commit_offsets<'a, J, I>(&mut self, group: &str, offsets: I) -> Result<()>
-        where J: AsRef<CommitOffset<'a>>, I: IntoIterator<Item=J>
+        where J: AsRef<CommitOffset<'a>>,
+              I: IntoIterator<Item = J>
     {
-        let mut req = protocol::OffsetCommitRequest::new(
-            group, self.config.offset_commit_version,
-            self.state.next_correlation_id(), &self.config.client_id);
+        let mut req = protocol::OffsetCommitRequest::new(group,
+                                                         self.config.offset_commit_version,
+                                                         self.state.next_correlation_id(),
+                                                         &self.config.client_id);
         for o in offsets {
             let o = o.as_ref();
             if self.state.contains_topic_partition(o.topic, o.partition) {
@@ -1044,9 +1085,12 @@ impl KafkaClient {
     /// ```
     ///
     /// See also `KafkaClient::commit_offsets`.
-    pub fn commit_offset(&mut self, group: &str, topic: &str, partition: i32, offset: i64)
-                         -> Result<()>
-    {
+    pub fn commit_offset(&mut self,
+                         group: &str,
+                         topic: &str,
+                         partition: i32,
+                         offset: i64)
+                         -> Result<()> {
         self.commit_offsets(group, &[CommitOffset::new(topic, partition, offset)])
     }
 
@@ -1068,13 +1112,17 @@ impl KafkaClient {
     /// ```
     ///
     /// See also `KafkaClient::fetch_group_topic_offsets`.
-    pub fn fetch_group_offsets<'a, J, I>(&mut self, group: &str, partitions: I)
+    pub fn fetch_group_offsets<'a, J, I>(&mut self,
+                                         group: &str,
+                                         partitions: I)
                                          -> Result<Vec<TopicPartitionOffset>>
-        where J: AsRef<FetchGroupOffset<'a>>, I: IntoIterator<Item=J>
+        where J: AsRef<FetchGroupOffset<'a>>,
+              I: IntoIterator<Item = J>
     {
-        let mut req = protocol::OffsetFetchRequest::new(
-            group, self.config.offset_fetch_version,
-            self.state.next_correlation_id(), &self.config.client_id);
+        let mut req = protocol::OffsetFetchRequest::new(group,
+                                                        self.config.offset_fetch_version,
+                                                        self.state.next_correlation_id(),
+                                                        &self.config.client_id);
         let mut num = 0;
         for p in partitions {
             let p = p.as_ref();
@@ -1099,12 +1147,14 @@ impl KafkaClient {
     /// client.load_metadata_all().unwrap();
     /// let offsets = client.fetch_group_topic_offsets("my-group", "my-topic").unwrap();
     /// ```
-    pub fn fetch_group_topic_offsets(&mut self, group: &str, topic: &str)
-                                     -> Result<Vec<TopicPartitionOffset>>
-    {
-        let mut req = protocol::OffsetFetchRequest::new(
-            group, self.config.offset_fetch_version,
-            self.state.next_correlation_id(), &self.config.client_id);
+    pub fn fetch_group_topic_offsets(&mut self,
+                                     group: &str,
+                                     topic: &str)
+                                     -> Result<Vec<TopicPartitionOffset>> {
+        let mut req = protocol::OffsetFetchRequest::new(group,
+                                                        self.config.offset_fetch_version,
+                                                        self.state.next_correlation_id(),
+                                                        &self.config.client_id);
         let num;
         match self.state.partitions_for(topic) {
             None => return Err(Error::Kafka(KafkaCode::UnknownTopicOrPartition)),
@@ -1120,10 +1170,13 @@ impl KafkaClient {
 }
 
 impl KafkaClientInternals for KafkaClient {
-    fn internal_produce_messages<'a, 'b, I, J>(
-        &mut self, required_acks: i16, ack_timeout: i32, messages: I)
-        -> Result<Vec<TopicPartitionOffset>>
-        where J: AsRef<ProduceMessage<'a, 'b>>, I: IntoIterator<Item=J>
+    fn internal_produce_messages<'a, 'b, I, J>(&mut self,
+                                               required_acks: i16,
+                                               ack_timeout: i32,
+                                               messages: I)
+                                               -> Result<Vec<TopicPartitionOffset>>
+        where J: AsRef<ProduceMessage<'a, 'b>>,
+              I: IntoIterator<Item = J>
     {
         let state = &mut self.state;
         let correlation = state.next_correlation_id();
@@ -1135,17 +1188,21 @@ impl KafkaClientInternals for KafkaClient {
             let msg = msg.as_ref();
             match state.find_broker(msg.topic, msg.partition) {
                 None => return Err(Error::Kafka(KafkaCode::UnknownTopicOrPartition)),
-                Some(broker) => reqs.entry(broker)
-                    .or_insert_with(
-                        || protocol::ProduceRequest::new(
-                            required_acks, ack_timeout, correlation,
-                            &config.client_id, config.compression))
-                    .add(msg.topic, msg.partition, msg.key, msg.value),
+                Some(broker) => {
+                    reqs.entry(broker)
+                        .or_insert_with(|| {
+                            protocol::ProduceRequest::new(required_acks,
+                                                          ack_timeout,
+                                                          correlation,
+                                                          &config.client_id,
+                                                          config.compression)
+                        })
+                        .add(msg.topic, msg.partition, msg.key, msg.value)
+                }
             }
         }
         __produce_messages(&mut self.conn_pool, reqs, required_acks == 0)
     }
-
 }
 
 fn __get_group_coordinator<'a>(group: &str,
@@ -1153,8 +1210,7 @@ fn __get_group_coordinator<'a>(group: &str,
                                conn_pool: &mut network::Connections,
                                config: &ClientConfig,
                                now: Instant)
-                               -> Result<&'a str>
-{
+                               -> Result<&'a str> {
     if let Some(host) = state.group_coordinator(group) {
         // ~ decouple the lifetimes to make borrowck happy;
         // this is actually safe since we're immediatelly
@@ -1178,7 +1234,7 @@ fn __get_group_coordinator<'a>(group: &str,
             Ok(r) => {
                 return Ok(state.set_group_coordinator(group, &r));
             }
-            Err(Error::Kafka(e@KafkaCode::GroupCoordinatorNotAvailable)) => {
+            Err(Error::Kafka(e @ KafkaCode::GroupCoordinatorNotAvailable)) => {
                 retry_code = e;
             }
             Err(e) => {
@@ -1187,7 +1243,8 @@ fn __get_group_coordinator<'a>(group: &str,
         }
         if attempt < config.retry_max_attempts {
             debug!("get_group_coordinator: will retry request (c: {}) due to: {:?}",
-                   req.header.correlation_id, retry_code);
+                   req.header.correlation_id,
+                   retry_code);
             attempt += 1;
             __retry_sleep(config);
         } else {
@@ -1200,8 +1257,7 @@ fn __commit_offsets(req: protocol::OffsetCommitRequest,
                     state: &mut state::ClientState,
                     conn_pool: &mut network::Connections,
                     config: &ClientConfig)
-                    -> Result<()>
-{
+                    -> Result<()> {
     let mut attempt = 1;
     loop {
         let now = Instant::now();
@@ -1209,7 +1265,9 @@ fn __commit_offsets(req: protocol::OffsetCommitRequest,
         let tps = {
             let host = try!(__get_group_coordinator(req.group, state, conn_pool, config, now));
             debug!("commit_offsets: committing on protocol v{} for '{}' to: {}",
-                   req.header.api_version, req.group, host);
+                   req.header.api_version,
+                   req.group,
+                   host);
             try!(__send_receive::<_, protocol::OffsetCommitResponse>(conn_pool, host, now, &req))
                 .topic_partitions
         };
@@ -1219,14 +1277,13 @@ fn __commit_offsets(req: protocol::OffsetCommitRequest,
         'rproc: for tp in tps {
             for p in tp.partitions {
                 match p.to_error() {
-                    None => {},
-                    Some(e@KafkaCode::GroupLoadInProgress) => {
+                    None => {}
+                    Some(e @ KafkaCode::GroupLoadInProgress) => {
                         retry_code = Some(e);
                         break 'rproc;
                     }
-                    Some(e@KafkaCode::NotCoordinatorForGroup) => {
-                        debug!("commit_offsets: resetting group coordinator for '{}'",
-                               req.group);
+                    Some(e @ KafkaCode::NotCoordinatorForGroup) => {
+                        debug!("commit_offsets: resetting group coordinator for '{}'", req.group);
                         state.remove_group_coordinator(&req.group);
                         retry_code = Some(e);
                         break 'rproc;
@@ -1242,7 +1299,8 @@ fn __commit_offsets(req: protocol::OffsetCommitRequest,
             Some(e) => {
                 if attempt < config.retry_max_attempts {
                     debug!("commit_offsets: will retry request (c: {}) due to: {:?}",
-                           req.header.correlation_id, e);
+                           req.header.correlation_id,
+                           e);
                     attempt += 1;
                     __retry_sleep(config);
                 }
@@ -1255,11 +1313,12 @@ fn __commit_offsets(req: protocol::OffsetCommitRequest,
 }
 
 // ~ 'num_partitions' is merely a hint to pre-allocation inside this method
-fn __fetch_group_offsets(
-    req: protocol::OffsetFetchRequest, num_partitions: usize,
-    state: &mut state::ClientState, conn_pool: &mut network::Connections,
-    config: &ClientConfig) -> Result<Vec<TopicPartitionOffset>>
-{
+fn __fetch_group_offsets(req: protocol::OffsetFetchRequest,
+                         num_partitions: usize,
+                         state: &mut state::ClientState,
+                         conn_pool: &mut network::Connections,
+                         config: &ClientConfig)
+                         -> Result<Vec<TopicPartitionOffset>> {
     let mut attempt = 1;
     loop {
         let now = Instant::now();
@@ -1267,7 +1326,9 @@ fn __fetch_group_offsets(
         let r = {
             let host = try!(__get_group_coordinator(req.group, state, conn_pool, config, now));
             debug!("fetch_group_offsets: fetching on protocol v{} for '{}' from: {}",
-                   req.header.api_version, req.group, host);
+                   req.header.api_version,
+                   req.group,
+                   host);
             try!(__send_receive::<_, protocol::OffsetFetchResponse>(conn_pool, host, now, &req))
         };
 
@@ -1288,11 +1349,11 @@ fn __fetch_group_offsets(
                         o.offset = Ok(-1);
                         v.push(o);
                     }
-                    Err(Error::Kafka(e@KafkaCode::GroupLoadInProgress)) => {
+                    Err(Error::Kafka(e @ KafkaCode::GroupLoadInProgress)) => {
                         retry_code = Some(e);
                         break 'rproc;
                     }
-                    Err(Error::Kafka(e@KafkaCode::NotCoordinatorForGroup)) => {
+                    Err(Error::Kafka(e @ KafkaCode::NotCoordinatorForGroup)) => {
                         debug!("fetch_group_offsets: resetting group coordinator for '{}'",
                                req.group);
                         state.remove_group_coordinator(&req.group);
@@ -1312,7 +1373,8 @@ fn __fetch_group_offsets(
             Some(e) => {
                 if attempt < config.retry_max_attempts {
                     debug!("fetch_group_offsets: will retry request (c: {}) due to: {:?}",
-                           req.header.correlation_id, e);
+                           req.header.correlation_id,
+                           e);
                     attempt += 1;
                     __retry_sleep(config)
                 } else {
@@ -1330,8 +1392,7 @@ fn __fetch_group_offsets(
 fn __fetch_messages(conn_pool: &mut network::Connections,
                     config: &ClientConfig,
                     reqs: HashMap<&str, protocol::FetchRequest>)
-                    -> Result<Vec<fetch::Response>>
-{
+                    -> Result<Vec<fetch::Response>> {
     let now = Instant::now();
     let mut res = Vec::with_capacity(reqs.len());
     for (host, req) in reqs {
@@ -1348,18 +1409,18 @@ fn __fetch_messages(conn_pool: &mut network::Connections,
 fn __produce_messages(conn_pool: &mut network::Connections,
                       reqs: HashMap<&str, protocol::ProduceRequest>,
                       no_acks: bool)
-                      -> Result<Vec<TopicPartitionOffset>>
-{
+                      -> Result<Vec<TopicPartitionOffset>> {
     let now = Instant::now();
     if no_acks {
         for (host, req) in reqs {
             try!(__send_noack::<_, protocol::ProduceResponse>(conn_pool, host, now, req));
         }
-        Ok(vec!())
+        Ok(vec![])
     } else {
         let mut res: Vec<TopicPartitionOffset> = vec![];
         for (host, req) in reqs {
-            let resp = try!(__send_receive::<_, protocol::ProduceResponse>(conn_pool, &host, now, req));
+            let resp =
+                try!(__send_receive::<_, protocol::ProduceResponse>(conn_pool, &host, now, req));
             for tpo in resp.get_response() {
                 res.push(tpo);
             }
@@ -1368,31 +1429,38 @@ fn __produce_messages(conn_pool: &mut network::Connections,
     }
 }
 
-fn __send_receive<T, V>(conn_pool: &mut network::Connections, host: &str, now: Instant, req: T)
+fn __send_receive<T, V>(conn_pool: &mut network::Connections,
+                        host: &str,
+                        now: Instant,
+                        req: T)
                         -> Result<V::R>
-    where T: ToByte, V: FromByte
+    where T: ToByte,
+          V: FromByte
 {
     __send_receive_conn::<T, V>(try!(conn_pool.get_conn(host, now)), req)
 }
 
 fn __send_receive_conn<T, V>(conn: &mut network::KafkaConnection, req: T) -> Result<V::R>
-    where T: ToByte, V: FromByte
+    where T: ToByte,
+          V: FromByte
 {
     try!(__send_request(conn, req));
     __get_response::<V>(conn)
 }
 
-fn __send_noack<T, V>(conn_pool: &mut network::Connections, host: &str, now: Instant, req: T)
+fn __send_noack<T, V>(conn_pool: &mut network::Connections,
+                      host: &str,
+                      now: Instant,
+                      req: T)
                       -> Result<usize>
-    where T: ToByte, V: FromByte
+    where T: ToByte,
+          V: FromByte
 {
     let mut conn = try!(conn_pool.get_conn(host, now));
     __send_request(&mut conn, req)
 }
 
-fn __send_request<T: ToByte>(conn: &mut network::KafkaConnection, request: T)
-                             -> Result<usize>
-{
+fn __send_request<T: ToByte>(conn: &mut network::KafkaConnection, request: T) -> Result<usize> {
     // ~ buffer to receive data to be sent
     let mut buffer = Vec::with_capacity(4);
     // ~ reserve bytes for the actual request size (we'll fill in that later)
@@ -1406,9 +1474,7 @@ fn __send_request<T: ToByte>(conn: &mut network::KafkaConnection, request: T)
     conn.send(&buffer)
 }
 
-fn __get_response<T: FromByte>(conn: &mut network::KafkaConnection)
-                             -> Result<T::R>
-{
+fn __get_response<T: FromByte>(conn: &mut network::KafkaConnection) -> Result<T::R> {
     let size = try!(__get_response_size(conn));
     let resp = try!(conn.read_exact_alloc(size as u64));
 
@@ -1427,9 +1493,14 @@ fn __get_response<T: FromByte>(conn: &mut network::KafkaConnection)
     T::decode_new(&mut Cursor::new(resp))
 }
 
-fn __z_send_receive<R, P>(conn_pool: &mut network::Connections, host: &str, now: Instant, req: R, parser: &P)
+fn __z_send_receive<R, P>(conn_pool: &mut network::Connections,
+                          host: &str,
+                          now: Instant,
+                          req: R,
+                          parser: &P)
                           -> Result<P::T>
-    where R: ToByte, P: ResponseParser
+    where R: ToByte,
+          P: ResponseParser
 {
     let mut conn = try!(conn_pool.get_conn(host, now));
     try!(__send_request(&mut conn, req));
diff --git a/src/client/network.rs b/src/client/network.rs
index 40917ea8d..7bbc793dc 100644
--- a/src/client/network.rs
+++ b/src/client/network.rs
@@ -41,14 +41,16 @@ struct Pooled<T> {
 
 impl<T> Pooled<T> {
     fn new(last_checkout: Instant, item: T) -> Self {
-        Pooled { last_checkout: last_checkout, item: item }
+        Pooled {
+            last_checkout: last_checkout,
+            item: item,
+        }
     }
 }
 
 impl<T: fmt::Debug> fmt::Debug for Pooled<T> {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        write!(f, "Pooled {{ last_checkout: {:?}, item: {:?} }}",
-               self.last_checkout, self.item)
+        write!(f, "Pooled {{ last_checkout: {:?}, item: {:?} }}", self.last_checkout, self.item)
     }
 }
 
@@ -63,19 +65,15 @@ pub struct Config {
 impl Config {
     #[cfg(not(feature = "security"))]
     fn new_conn(&self, id: u32, host: &str) -> Result<KafkaConnection> {
-        KafkaConnection::new(id, host, self.rw_timeout)
-            .map(|c| {
-                debug!("Established: {:?}", c);
-                c
-            })
+        KafkaConnection::new(id, host, self.rw_timeout).map(|c| {
+            debug!("Established: {:?}", c);
+            c
+        })
     }
 
     #[cfg(feature = "security")]
     fn new_conn(&self, id: u32, host: &str) -> Result<KafkaConnection> {
-        KafkaConnection::new(id,
-                             host,
-                             self.rw_timeout,
-                             self.security_config.as_ref().map(|c| &c.0))
+        KafkaConnection::new(id, host, self.rw_timeout, self.security_config.as_ref().map(|c| &c.0))
             .map(|c| {
                 debug!("Established: {:?}", c);
                 c
@@ -129,8 +127,7 @@ impl Connections {
     pub fn new_with_security(rw_timeout: Option<Duration>,
                              idle_timeout: Duration,
                              security: Option<SecurityConfig>)
-                             -> Connections
-    {
+                             -> Connections {
         Connections {
             conns: HashMap::new(),
             state: State::new(),
@@ -150,9 +147,7 @@ impl Connections {
         self.config.idle_timeout
     }
 
-    pub fn get_conn<'a>(&'a mut self, host: &str, now: Instant)
-                        -> Result<&'a mut KafkaConnection>
-    {
+    pub fn get_conn<'a>(&'a mut self, host: &str, now: Instant) -> Result<&'a mut KafkaConnection> {
         if let Some(conn) = self.conns.get_mut(host) {
             if now.duration_since(conn.last_checkout) >= self.config.idle_timeout {
                 debug!("Idle timeout reached: {:?}", conn.item);
@@ -169,8 +164,7 @@ impl Connections {
             return Ok(unsafe { mem::transmute(kconn) });
         }
         let cid = self.state.next_conn_id();
-        self.conns.insert(host.to_owned(),
-                          Pooled::new(now, try!(self.config.new_conn(cid, host))));
+        self.conns.insert(host.to_owned(), Pooled::new(now, try!(self.config.new_conn(cid, host))));
         Ok(&mut self.conns.get_mut(host).unwrap().item)
     }
 
@@ -230,7 +224,7 @@ mod openssled {
 
     pub enum KafkaStream {
         Plain(TcpStream),
-        Ssl(SslStream<TcpStream>)
+        Ssl(SslStream<TcpStream>),
     }
 
     impl IsSecured for KafkaStream {
@@ -246,7 +240,7 @@ mod openssled {
         fn get_ref(&self) -> &TcpStream {
             match *self {
                 KafkaStream::Plain(ref s) => s,
-                KafkaStream::Ssl(ref s) => s.get_ref()
+                KafkaStream::Ssl(ref s) => s.get_ref(),
             }
         }
 
@@ -301,13 +295,15 @@ pub struct KafkaConnection {
 
 impl fmt::Debug for KafkaConnection {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        write!(f, "KafkaConnection {{ id: {}, secured: {}, host: \"{}\" }}",
-               self.id, self.stream.is_secured(), self.host)
+        write!(f,
+               "KafkaConnection {{ id: {}, secured: {}, host: \"{}\" }}",
+               self.id,
+               self.stream.is_secured(),
+               self.host)
     }
 }
 
 impl KafkaConnection {
-
     pub fn send(&mut self, msg: &[u8]) -> Result<usize> {
         let r = self.stream.write(&msg[..]).map_err(From::from);
         trace!("Sent {} bytes to: {:?} => {:?}", msg.len(), self, r);
@@ -338,12 +334,18 @@ impl KafkaConnection {
         r.map_err(From::from)
     }
 
-    fn from_stream(stream: KafkaStream, id: u32, host: &str, rw_timeout: Option<Duration>)
-                   -> Result<KafkaConnection>
-    {
+    fn from_stream(stream: KafkaStream,
+                   id: u32,
+                   host: &str,
+                   rw_timeout: Option<Duration>)
+                   -> Result<KafkaConnection> {
         try!(stream.set_read_timeout(rw_timeout));
         try!(stream.set_write_timeout(rw_timeout));
-        Ok(KafkaConnection { id: id, host: host.to_owned(), stream: stream })
+        Ok(KafkaConnection {
+            id: id,
+            host: host.to_owned(),
+            stream: stream,
+        })
     }
 
     #[cfg(not(feature = "security"))]
@@ -352,9 +354,11 @@ impl KafkaConnection {
     }
 
     #[cfg(feature = "security")]
-    fn new(id: u32, host: &str, rw_timeout: Option<Duration>, security: Option<&SslContext>)
-           -> Result<KafkaConnection>
-    {
+    fn new(id: u32,
+           host: &str,
+           rw_timeout: Option<Duration>,
+           security: Option<&SslContext>)
+           -> Result<KafkaConnection> {
         let stream = try!(TcpStream::connect(host));
         let stream = match security {
             Some(ctx) => KafkaStream::Ssl(try!(SslStream::connect(ctx, stream))),
diff --git a/src/client/state.rs b/src/client/state.rs
index 400adaa55..3e2b690d9 100644
--- a/src/client/state.rs
+++ b/src/client/state.rs
@@ -187,7 +187,7 @@ pub struct TopicNames<'a> {
 }
 
 impl<'a> Iterator for TopicNames<'a> {
-    type Item=&'a str;
+    type Item = &'a str;
 
     #[inline]
     fn next(&mut self) -> Option<Self::Item> {
@@ -281,7 +281,7 @@ impl ClientState {
                 }
                 Entry::Vacant(e) => {
                     &mut e.insert(TopicPartitions::new_with_partitions(t.partitions.len()))
-                          .partitions
+                        .partitions
                 }
             };
             // ~ sync the partitions vector with the new information
@@ -353,12 +353,11 @@ impl ClientState {
     /// ~ Updates the coordinator for the specified group and returns
     /// the coordinator host as if `group_coordinator` would have
     /// been called subsequently.
-    pub fn set_group_coordinator<'a>(
-        &'a mut self, group: &str, gc: &protocol::GroupCoordinatorResponse)
-        -> &'a str
-    {
-        debug!("set_group_coordinator: registering coordinator for '{}': {:?}",
-               group, gc);
+    pub fn set_group_coordinator<'a>(&'a mut self,
+                                     group: &str,
+                                     gc: &protocol::GroupCoordinatorResponse)
+                                     -> &'a str {
+        debug!("set_group_coordinator: registering coordinator for '{}': {:?}", group, gc);
 
         let group_host = format!("{}:{}", gc.host, gc.port);
         // ~ try to find an already existing broker
@@ -366,8 +365,11 @@ impl ClientState {
         for (i, broker) in (0u32..).zip(self.brokers.iter()) {
             if gc.broker_id == broker.node_id {
                 if group_host != broker.host {
-                    warn!("set_group_coordinator: coord_host({}) != broker_host({}) for broker_id({})!",
-                          group_host, broker.host, broker.node_id);
+                    warn!("set_group_coordinator: coord_host({}) != broker_host({}) for \
+                           broker_id({})!",
+                          group_host,
+                          broker.host,
+                          broker.node_id);
                 }
                 broker_ref._index = i;
                 break;
@@ -466,13 +468,13 @@ mod tests {
         assert_eq!(expected.len() == 0, partitions.is_empty());
         assert_eq!(expected,
                    &partitions.iter()
-                              .map(|(id, tp)| {
-                                  let broker = tp.broker(&state).map(|b| (b.id(), b.host()));
-                                  // ~ verify that find_broker delivers the same information
-                                  assert_eq!(broker.map(|b| b.1), state.find_broker(topic, id));
-                                  (id, broker)
-                              })
-                              .collect::<Vec<_>>()[..]);
+                       .map(|(id, tp)| {
+                           let broker = tp.broker(&state).map(|b| (b.id(), b.host()));
+                           // ~ verify that find_broker delivers the same information
+                           assert_eq!(broker.map(|b| b.1), state.find_broker(topic, id));
+                           (id, broker)
+                       })
+                       .collect::<Vec<_>>()[..]);
     }
 
     fn assert_initial_metadata_load(state: &ClientState) {
diff --git a/src/client_internals.rs b/src/client_internals.rs
index 0ced10a89..69c90fde9 100644
--- a/src/client_internals.rs
+++ b/src/client_internals.rs
@@ -5,8 +5,11 @@ use client;
 use error::Result;
 
 pub trait KafkaClientInternals {
-    fn internal_produce_messages<'a, 'b, I, J>(
-        &mut self, required_acks: i16, ack_timeout: i32, messages: I)
-        -> Result<Vec<client::TopicPartitionOffset>>
-        where J: AsRef<client::ProduceMessage<'a, 'b>>, I: IntoIterator<Item=J>;
+    fn internal_produce_messages<'a, 'b, I, J>(&mut self,
+                                               required_acks: i16,
+                                               ack_timeout: i32,
+                                               messages: I)
+                                               -> Result<Vec<client::TopicPartitionOffset>>
+        where J: AsRef<client::ProduceMessage<'a, 'b>>,
+              I: IntoIterator<Item = J>;
 }
diff --git a/src/codecs.rs b/src/codecs.rs
index e45199744..9978ad932 100644
--- a/src/codecs.rs
+++ b/src/codecs.rs
@@ -34,25 +34,25 @@ impl<'a, T: ToByte + 'a + ?Sized> ToByte for &'a T {
 }
 
 impl ToByte for i8 {
-    fn encode<T:Write>(&self, buffer: &mut T) -> Result<()> {
+    fn encode<T: Write>(&self, buffer: &mut T) -> Result<()> {
         buffer.write_i8(*self).or_else(|e| Err(From::from(e)))
     }
 }
 
 impl ToByte for i16 {
-    fn encode<T:Write>(&self, buffer: &mut T) -> Result<()> {
+    fn encode<T: Write>(&self, buffer: &mut T) -> Result<()> {
         buffer.write_i16::<BigEndian>(*self).or_else(|e| Err(From::from(e)))
     }
 }
 
 impl ToByte for i32 {
-    fn encode<T:Write>(&self, buffer: &mut T) -> Result<()> {
+    fn encode<T: Write>(&self, buffer: &mut T) -> Result<()> {
         buffer.write_i32::<BigEndian>(*self).or_else(|e| Err(From::from(e)))
     }
 }
 
 impl ToByte for i64 {
-    fn encode<T:Write>(&self, buffer: &mut T) -> Result<()> {
+    fn encode<T: Write>(&self, buffer: &mut T) -> Result<()> {
         buffer.write_i64::<BigEndian>(*self).or_else(|e| Err(From::from(e)))
     }
 }
@@ -76,8 +76,8 @@ fn test_string_too_long() {
     assert!(buf.is_empty());
 }
 
-impl <V: ToByte> ToByte for [V] {
-    fn encode<T:Write>(&self, buffer: &mut T) -> Result<()> {
+impl<V: ToByte> ToByte for [V] {
+    fn encode<T: Write>(&self, buffer: &mut T) -> Result<()> {
         encode_as_array(buffer, self, |buffer, x| x.encode(buffer))
     }
 }
@@ -96,18 +96,16 @@ pub struct AsStrings<'a, T: 'a>(pub &'a [T]);
 
 impl<'a, T: AsRef<str> + 'a> ToByte for AsStrings<'a, T> {
     fn encode<W: Write>(&self, buffer: &mut W) -> Result<()> {
-        encode_as_array(buffer, self.0, |buffer, x| {
-            x.as_ref().encode(buffer)
-        })
+        encode_as_array(buffer, self.0, |buffer, x| x.as_ref().encode(buffer))
     }
 }
 
 /// ~ Renders the length of `xs` to `buffer` as the start of a
 /// protocol array and then for each element of `xs` invokes `f`
 /// assuming that function will render the element to the buffer.
-pub fn encode_as_array<T, F, W>(buffer: &mut W, xs: &[T], mut f: F)
-                                -> Result<()>
-    where F: FnMut(&mut W, &T) -> Result<()>, W: Write
+pub fn encode_as_array<T, F, W>(buffer: &mut W, xs: &[T], mut f: F) -> Result<()>
+    where F: FnMut(&mut W, &T) -> Result<()>,
+          W: Write
 {
     let l = try_usize_to_int!(xs.len(), i32);
     try!(buffer.write_i32::<BigEndian>(l));
@@ -127,7 +125,7 @@ pub trait FromByte {
         let mut temp: Self::R = Default::default();
         match temp.decode(buffer) {
             Ok(_) => Ok(temp),
-            Err(e) => Err(e)
+            Err(e) => Err(e),
         }
     }
 }
@@ -193,7 +191,9 @@ impl FromByte for String {
         if let Err(e) = decode!(buffer, read_i16, &mut length) {
             return Err(e);
         }
-        if length <= 0 { return Ok(()); }
+        if length <= 0 {
+            return Ok(());
+        }
         self.reserve(length as usize);
         let _ = buffer.take(length as u64).read_to_string(self);
         if self.len() != length as usize {
@@ -203,7 +203,7 @@ impl FromByte for String {
     }
 }
 
-impl <V: FromByte + Default> FromByte for Vec<V>{
+impl<V: FromByte + Default> FromByte for Vec<V> {
     type R = Vec<V>;
 
     fn decode<T: Read>(&mut self, buffer: &mut T) -> Result<()> {
@@ -211,7 +211,9 @@ impl <V: FromByte + Default> FromByte for Vec<V>{
         if let Err(e) = decode!(buffer, read_i32, &mut length) {
             return Err(e);
         }
-        if length <= 0 { return Ok(()); }
+        if length <= 0 {
+            return Ok(());
+        }
         self.reserve(length as usize);
         for _ in 0..length {
             let mut e: V = Default::default();
@@ -222,26 +224,28 @@ impl <V: FromByte + Default> FromByte for Vec<V>{
     }
 }
 
-impl FromByte for Vec<u8>{
+impl FromByte for Vec<u8> {
     type R = Vec<u8>;
 
     fn decode<T: Read>(&mut self, buffer: &mut T) -> Result<()> {
         let mut length: i32 = 0;
         match decode!(buffer, read_i32, &mut length) {
-            Ok(_) => {},
-            Err(e) => return Err(e)
+            Ok(_) => {}
+            Err(e) => return Err(e),
+        }
+        if length <= 0 {
+            return Ok(());
         }
-        if length <= 0 {return Ok(());}
         self.reserve(length as usize);
         match buffer.take(length as u64).read_to_end(self) {
             Ok(size) => {
                 if size < length as usize {
-                    return Err(Error::UnexpectedEOF)
+                    return Err(Error::UnexpectedEOF);
                 } else {
                     Ok(())
                 }
-            },
-            Err(e) => Err(From::from(e))
+            }
+            Err(e) => Err(From::from(e)),
         }
     }
 }
@@ -249,7 +253,7 @@ impl FromByte for Vec<u8>{
 #[test]
 fn codec_i8() {
     use std::io::Cursor;
-    let mut buf = vec!();
+    let mut buf = vec![];
     let orig: i8 = 5;
 
     // Encode into buffer
@@ -269,7 +273,7 @@ fn codec_i8() {
 #[test]
 fn codec_i16() {
     use std::io::Cursor;
-    let mut buf = vec!();
+    let mut buf = vec![];
     let orig: i16 = 5;
 
     // Encode into buffer
@@ -289,7 +293,7 @@ fn codec_i16() {
 #[test]
 fn codec_32() {
     use std::io::Cursor;
-    let mut buf = vec!();
+    let mut buf = vec![];
     let orig: i32 = 5;
 
     // Encode into buffer
@@ -309,7 +313,7 @@ fn codec_32() {
 #[test]
 fn codec_i64() {
     use std::io::Cursor;
-    let mut buf = vec!();
+    let mut buf = vec![];
     let orig: i64 = 5;
 
     // Encode into buffer
@@ -329,7 +333,7 @@ fn codec_i64() {
 #[test]
 fn codec_string() {
     use std::io::Cursor;
-    let mut buf = vec!();
+    let mut buf = vec![];
     let orig = "test".to_owned();
 
     // Encode into buffer
@@ -349,15 +353,15 @@ fn codec_string() {
 #[test]
 fn codec_vec_u8() {
     use std::io::Cursor;
-    let mut buf = vec!();
-    let orig: Vec<u8> = vec!(1, 2, 3);
+    let mut buf = vec![];
+    let orig: Vec<u8> = vec![1, 2, 3];
 
     // Encode into buffer
     orig.encode(&mut buf).unwrap();
     assert_eq!(buf, [0, 0, 0, 3, 1, 2, 3]);
 
     // Read from buffer into existing variable
-    let mut dec1: Vec<u8> = vec!();
+    let mut dec1: Vec<u8> = vec![];
     dec1.decode(&mut Cursor::new(&buf[..])).unwrap();
     assert_eq!(dec1, orig);
 
@@ -397,17 +401,20 @@ fn codec_as_strings() {
         }}
     }
 
-    { // slice of &str
+    {
+        // slice of &str
         let orig: &[&str] = &["abc", "defg"];
         enc_dec_cmp!(orig);
     }
 
-    { // vec of &str
+    {
+        // vec of &str
         let orig: Vec<&str> = vec!["abc", "defg"];
         enc_dec_cmp!(orig);
     }
 
-    { // vec of String
+    {
+        // vec of String
         let orig: Vec<String> = vec!["abc".to_owned(), "defg".to_owned()];
         enc_dec_cmp!(orig);
     }
diff --git a/src/compression/gzip.rs b/src/compression/gzip.rs
index 2ef768d54..ab9e2ea30 100644
--- a/src/compression/gzip.rs
+++ b/src/compression/gzip.rs
@@ -19,8 +19,8 @@ pub fn uncompress<T: Read>(src: T) -> Result<Vec<u8>> {
 
     let mut buffer: Vec<u8> = Vec::new();
     match d.read_to_end(&mut buffer) {
-        Err(err) =>  Err(From::from(err)),
-        Ok(_) => Ok(buffer)
+        Err(err) => Err(From::from(err)),
+        Ok(_) => Ok(buffer),
     }
 }
 
@@ -28,7 +28,8 @@ pub fn uncompress<T: Read>(src: T) -> Result<Vec<u8>> {
 fn test_uncompress() {
     use std::io::Cursor;
     // The vector should uncompress to "test"
-    let msg: Vec<u8> = vec!(31, 139, 8, 0, 192, 248, 79, 85, 2, 255, 43, 73, 45, 46, 1, 0, 12, 126, 127, 216, 4, 0, 0, 0);
+    let msg: Vec<u8> = vec![31, 139, 8, 0, 192, 248, 79, 85, 2, 255, 43, 73, 45, 46, 1, 0, 12,
+                            126, 127, 216, 4, 0, 0, 0];
     let uncomp_msg = String::from_utf8(uncompress(Cursor::new(msg)).unwrap()).unwrap();
     assert_eq!(&uncomp_msg[..], "test");
 }
@@ -37,7 +38,7 @@ fn test_uncompress() {
 #[should_panic]
 fn test_uncompress_panic() {
     use std::io::Cursor;
-    let msg: Vec<u8> = vec!(12, 42, 84, 104, 105, 115, 32, 105, 115, 32, 116, 101, 115, 116);
+    let msg: Vec<u8> = vec![12, 42, 84, 104, 105, 115, 32, 105, 115, 32, 116, 101, 115, 116];
     let uncomp_msg = String::from_utf8(uncompress(Cursor::new(msg)).unwrap()).unwrap();
     assert_eq!(&uncomp_msg[..], "This is test");
 }
diff --git a/src/compression/snappy.rs b/src/compression/snappy.rs
index 6f198dd62..1260642e3 100644
--- a/src/compression/snappy.rs
+++ b/src/compression/snappy.rs
@@ -39,15 +39,23 @@ macro_rules! next_i32 {
 /// stream following the validated header.
 fn validate_stream(mut stream: &[u8]) -> Result<&[u8]> {
     // ~ check the "header magic"
-    if stream.len() < MAGIC.len() { return Err(Error::UnexpectedEOF); }
-    if &stream[..MAGIC.len()] != MAGIC { return Err(Error::InvalidInputSnappy); }
+    if stream.len() < MAGIC.len() {
+        return Err(Error::UnexpectedEOF);
+    }
+    if &stream[..MAGIC.len()] != MAGIC {
+        return Err(Error::InvalidInputSnappy);
+    }
     stream = &stream[MAGIC.len()..];
     // ~ let's be assertive and (for the moment) restrict ourselves to
     // version == 1 and compatibility == 1.
     let version = next_i32!(stream);
-    if version != 1 { return Err(Error::InvalidInputSnappy); }
+    if version != 1 {
+        return Err(Error::InvalidInputSnappy);
+    }
     let compat = next_i32!(stream);
-    if compat != 1 { return Err(Error::InvalidInputSnappy); }
+    if compat != 1 {
+        return Err(Error::InvalidInputSnappy);
+    }
     Ok(stream)
 }
 
@@ -200,13 +208,15 @@ mod tests {
         let compressed = &[12, 42, 84, 104, 105, 115, 32, 105, 115, 32, 116, 101, 115, 116];
         let uncompressed = uncompress(compressed);
         assert!(uncompressed.is_err());
-        assert!(if let Some(Error::InvalidInputSnappy) = uncompressed.err() { true } else { false });
+        assert!(if let Some(Error::InvalidInputSnappy) = uncompressed.err() {
+            true
+        } else {
+            false
+        });
     }
 
-    static ORIGINAL: &'static str =
-        include_str!("../../test-data/fetch1.txt");
-    static COMPRESSED: &'static [u8] =
-        include_bytes!("../../test-data/fetch1.snappy.chunked.4k");
+    static ORIGINAL: &'static str = include_str!("../../test-data/fetch1.txt");
+    static COMPRESSED: &'static [u8] = include_bytes!("../../test-data/fetch1.snappy.chunked.4k");
 
     #[test]
     fn test_snappy_reader_read() {
diff --git a/src/consumer/assignment.rs b/src/consumer/assignment.rs
index ecff054b5..7fc1b357a 100644
--- a/src/consumer/assignment.rs
+++ b/src/consumer/assignment.rs
@@ -14,7 +14,7 @@ pub struct Assignment {
     /// available partitions
     /// ~ kept in ascending order
     /// ~ no duplicates
-    partitions: Vec<i32>, 
+    partitions: Vec<i32>,
 }
 
 impl Assignment {
@@ -41,7 +41,8 @@ impl Assignments {
     }
 
     pub fn topic_ref(&self, topic: &str) -> Option<AssignmentRef> {
-        self.0.binary_search_by(|x| x.topic.as_str().cmp(topic))
+        self.0
+            .binary_search_by(|x| x.topic.as_str().cmp(topic))
             .ok()
             .map(|i| AssignmentRef(i as u32))
     }
diff --git a/src/consumer/builder.rs b/src/consumer/builder.rs
index 01e51b296..f946c40bb 100644
--- a/src/consumer/builder.rs
+++ b/src/consumer/builder.rs
@@ -219,9 +219,7 @@ impl Builder {
         // ~ create the client if necessary
         let (mut client, need_metadata) = match self.client {
             Some(client) => (client, false),
-            None => {
-                (Self::new_kafka_client(self.hosts, self.security_config), true)
-            }
+            None => (Self::new_kafka_client(self.hosts, self.security_config), true),
         };
         // ~ apply configuration settings
         try!(client.set_fetch_max_wait_time(self.fetch_max_wait_time));
@@ -237,10 +235,14 @@ impl Builder {
         let config = Config {
             group: self.group,
             fallback_offset: self.fallback_offset,
-            retry_max_bytes_limit: self.retry_max_bytes_limit
+            retry_max_bytes_limit: self.retry_max_bytes_limit,
         };
         let state = try!(State::new(&mut client, &config, assignment::from_map(self.assignments)));
         debug!("initialized: Consumer {{ config: {:?}, state: {:?} }}", config, state);
-        Ok(Consumer{ client: client, state: state, config: config })
+        Ok(Consumer {
+            client: client,
+            state: state,
+            config: config,
+        })
     }
 }
diff --git a/src/consumer/mod.rs b/src/consumer/mod.rs
index 177d1f127..c7d469392 100644
--- a/src/consumer/mod.rs
+++ b/src/consumer/mod.rs
@@ -98,7 +98,6 @@ pub struct Consumer {
 // XXX 2) Issue IO in a separate (background) thread and pre-fetch messagesets
 
 impl Consumer {
-
     /// Starts building a consumer using the given kafka client.
     pub fn from_client(client: KafkaClient) -> Builder {
         builder::new(Some(client), Vec::new())
@@ -146,16 +145,21 @@ impl Consumer {
                 };
                 let topic = self.state.topic_name(tp.topic_ref);
                 debug!("fetching messages: (fetch-offset: {{\"{}:{}\": {:?}}})",
-                       topic, tp.partition, s);
-                (1, self.client.fetch_messages_for_partition(
-                    &FetchPartition::new(topic, tp.partition, s.offset)
-                        .with_max_bytes(s.max_bytes)))
+                       topic,
+                       tp.partition,
+                       s);
+                (1,
+                 self.client.fetch_messages_for_partition(&FetchPartition::new(topic,
+                                                                               tp.partition,
+                                                                               s.offset)
+                    .with_max_bytes(s.max_bytes)))
             }
             None => {
                 let client = &mut self.client;
                 let state = &self.state;
                 debug!("fetching messages: (fetch-offsets: {:?})", state.fetch_offsets_debug());
-                let reqs = state.fetch_offsets.iter()
+                let reqs = state.fetch_offsets
+                    .iter()
                     .map(|(tp, s)| {
                         let topic = state.topic_name(tp.topic_ref);
                         FetchPartition::new(topic, tp.partition, s.offset)
@@ -172,16 +176,19 @@ impl Consumer {
     // ~ num_partitions_queried: the original number of partitions requested/queried for
     //   the responses
     // ~ resps: the responses to post process
-    fn process_fetch_responses(&mut self, num_partitions_queried: u32, resps: Vec<fetch::Response>)
-                               -> Result<MessageSets>
-    {
+    fn process_fetch_responses(&mut self,
+                               num_partitions_queried: u32,
+                               resps: Vec<fetch::Response>)
+                               -> Result<MessageSets> {
         let single_partition_consumer = self.single_partition_consumer();
         let mut empty = true;
         let mut retry_partitions = &mut self.state.retry_partitions;
 
         for resp in &resps {
             for t in resp.topics() {
-                let topic_ref = self.state.assignments.topic_ref(t.topic())
+                let topic_ref = self.state
+                    .assignments
+                    .topic_ref(t.topic())
                     .expect("unknown topic in response");
 
                 for p in t.partitions() {
@@ -202,9 +209,10 @@ impl Consumer {
                         &Ok(ref data) => data,
                     };
 
-                    let mut fetch_state = self.state.fetch_offsets
-                            .get_mut(&tp)
-                            .expect("non-requested partition");
+                    let mut fetch_state = self.state
+                        .fetch_offsets
+                        .get_mut(&tp)
+                        .expect("non-requested partition");
                     // ~ book keeping
                     if let Some(last_msg) = data.messages().last() {
                         fetch_state.offset = last_msg.offset + 1;
@@ -217,11 +225,19 @@ impl Consumer {
                             let prev_max_bytes = fetch_state.max_bytes;
                             fetch_state.max_bytes = self.client.fetch_max_bytes_per_partition();
                             debug!("reset max_bytes for {}:{} from {} to {}",
-                                   t.topic(), tp.partition, prev_max_bytes, fetch_state.max_bytes);
+                                   t.topic(),
+                                   tp.partition,
+                                   prev_max_bytes,
+                                   fetch_state.max_bytes);
                         }
                     } else {
-                        debug!("no data received for {}:{} (max_bytes: {} / fetch_offset: {} / highwatermark_offset: {})",
-                               t.topic(), tp.partition, fetch_state.max_bytes, fetch_state.offset, data.highwatermark_offset());
+                        debug!("no data received for {}:{} (max_bytes: {} / fetch_offset: {} / \
+                                highwatermark_offset: {})",
+                               t.topic(),
+                               tp.partition,
+                               fetch_state.max_bytes,
+                               fetch_state.offset,
+                               data.highwatermark_offset());
 
                         // ~ when a partition is empty but has a
                         // highwatermark-offset equal to or greater
@@ -239,7 +255,10 @@ impl Consumer {
                                     fetch_state.max_bytes = incr_max_bytes;
                                 }
                                 debug!("increased max_bytes for {}:{} from {} to {}",
-                                       t.topic(), tp.partition, prev_max_bytes, fetch_state.max_bytes);
+                                       t.topic(),
+                                       tp.partition,
+                                       prev_max_bytes,
+                                       fetch_state.max_bytes);
                             } else if num_partitions_queried == 1 {
                                 // ~ this was a single partition
                                 // request and we didn't get anything
@@ -269,16 +288,24 @@ impl Consumer {
         // request's response will likely be already ready for
         // consumption
 
-        Ok(MessageSets{ responses: resps, empty: empty })
+        Ok(MessageSets {
+            responses: resps,
+            empty: empty,
+        })
     }
 
     /// Retrieves the offset of the last "consumed" message in the
     /// specified partition. Results in `None` if there is no such
     /// "consumed" message.
     pub fn last_consumed_message(&self, topic: &str, partition: i32) -> Option<i64> {
-        self.state.topic_ref(topic)
-            .and_then(|tref| self.state.consumed_offsets.get(
-                &state::TopicPartition { topic_ref: tref, partition: partition }))
+        self.state
+            .topic_ref(topic)
+            .and_then(|tref| {
+                self.state.consumed_offsets.get(&state::TopicPartition {
+                    topic_ref: tref,
+                    partition: partition,
+                })
+            })
             .map(|co| co.offset)
     }
 
@@ -298,10 +325,16 @@ impl Consumer {
             None => return Err(Error::Kafka(KafkaCode::UnknownTopicOrPartition)),
             Some(topic_ref) => topic_ref,
         };
-        let tp = state::TopicPartition {topic_ref: topic_ref, partition: partition };
+        let tp = state::TopicPartition {
+            topic_ref: topic_ref,
+            partition: partition,
+        };
         match self.state.consumed_offsets.entry(tp) {
             Entry::Vacant(v) => {
-                v.insert(state::ConsumedOffset { offset: offset, dirty: true });
+                v.insert(state::ConsumedOffset {
+                    offset: offset,
+                    dirty: true,
+                });
             }
             Entry::Occupied(mut v) => {
                 let o = v.get_mut();
@@ -336,16 +369,17 @@ impl Consumer {
             return Ok(());
         }
         debug!("commit_consumed: commiting dirty-only consumer offsets (group: {} / offsets: {:?}",
-               self.config.group, self.state.consumed_offsets_debug());
+               self.config.group,
+               self.state.consumed_offsets_debug());
         let (client, state) = (&mut self.client, &mut self.state);
-        try!(client.commit_offsets(
-            &self.config.group,
-            state.consumed_offsets.iter()
-                .filter(|&(_, o)| o.dirty)
-                .map(|(tp, o)| {
-                    let topic = state.topic_name(tp.topic_ref);
-                    CommitOffset::new(topic, tp.partition, o.offset)
-                })));
+        try!(client.commit_offsets(&self.config.group,
+                                   state.consumed_offsets
+                                       .iter()
+                                       .filter(|&(_, o)| o.dirty)
+                                       .map(|(tp, o)| {
+                                           let topic = state.topic_name(tp.topic_ref);
+                                           CommitOffset::new(topic, tp.partition, o.offset)
+                                       })));
         for (_, co) in &mut state.consumed_offsets {
             if co.dirty {
                 co.dirty = false;
@@ -366,7 +400,7 @@ pub struct MessageSets {
 
     /// Precomputed; Says whether there are some messages or whether
     /// the responses actually contain consumeable messages
-    empty: bool
+    empty: bool,
 }
 
 impl MessageSets {
@@ -381,8 +415,7 @@ impl MessageSets {
     pub fn iter(&self) -> MessageSetsIter {
         let mut responses = self.responses.iter();
         let mut topics = responses.next().map(|r| r.topics().iter());
-        let (curr_topic, partitions) =
-            topics.as_mut()
+        let (curr_topic, partitions) = topics.as_mut()
             .and_then(|t| t.next())
             .map_or((None, None), |t| (Some(t.topic()), Some(t.partitions().iter())));
         MessageSetsIter {
diff --git a/src/consumer/state.rs b/src/consumer/state.rs
index 569c9c1fd..b89ccb0fd 100644
--- a/src/consumer/state.rs
+++ b/src/consumer/state.rs
@@ -60,21 +60,24 @@ pub struct State {
 
 impl<'a> fmt::Debug for State {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        write!(f, "State {{ assignments: {:?}, \
-                   fetch_offsets: {:?}, \
-                   retry_partitions: {:?}, \
-                   consumed_offsets: {:?} }}",
+        write!(f,
+               "State {{ assignments: {:?}, fetch_offsets: {:?}, retry_partitions: {:?}, \
+                consumed_offsets: {:?} }}",
                self.assignments,
                self.fetch_offsets_debug(),
-               TopicPartitionsDebug { state: self, tps: &self.retry_partitions },
+               TopicPartitionsDebug {
+                   state: self,
+                   tps: &self.retry_partitions,
+               },
                self.consumed_offsets_debug())
     }
 }
 
 impl State {
-    pub fn new(client: &mut KafkaClient, config: &Config, assignments: Assignments)
-               -> Result<State>
-    {
+    pub fn new(client: &mut KafkaClient,
+               config: &Config,
+               assignments: Assignments)
+               -> Result<State> {
         let (consumed_offsets, fetch_offsets) = {
             let subscriptions = {
                 let xs = assignments.as_slice();
@@ -85,8 +88,10 @@ impl State {
                 subs
             };
             let n = subscriptions.iter().map(|s| s.partitions.len()).fold(0, |acc, n| acc + n);
-            let consumed = try!(load_consumed_offsets(client, &config.group, &assignments, &subscriptions, n));
-            let fetch_next = try!(load_fetch_states(client, config, &assignments, &subscriptions, &consumed, n));
+            let consumed =
+                try!(load_consumed_offsets(client, &config.group, &assignments, &subscriptions, n));
+            let fetch_next =
+                try!(load_fetch_states(client, config, &assignments, &subscriptions, &consumed, n));
             (consumed, fetch_next)
         };
         Ok(State {
@@ -108,11 +113,17 @@ impl State {
     /// Returns a wrapper around `self.fetch_offsets` for nice dumping
     /// in debug messages
     pub fn fetch_offsets_debug<'a>(&'a self) -> OffsetsMapDebug<'a, FetchState> {
-        OffsetsMapDebug { state: self, offsets: &self.fetch_offsets }
+        OffsetsMapDebug {
+            state: self,
+            offsets: &self.fetch_offsets,
+        }
     }
 
     pub fn consumed_offsets_debug<'a>(&'a self) -> OffsetsMapDebug<'a, ConsumedOffset> {
-        OffsetsMapDebug { state: self, offsets: &self.consumed_offsets }
+        OffsetsMapDebug {
+            state: self,
+            offsets: &self.consumed_offsets,
+        }
     }
 }
 
@@ -125,16 +136,15 @@ struct Subscription<'a> {
 /// Determines the partitions to be consumed according to the
 /// specified topic and requested partitions configuration. Returns an
 /// ordered list of the partition ids to consume.
-fn determine_partitions<'a>(assignment: &'a Assignment, metadata: Topics)
-                            -> Result<Subscription<'a>>
-{
+fn determine_partitions<'a>(assignment: &'a Assignment,
+                            metadata: Topics)
+                            -> Result<Subscription<'a>> {
     let topic = assignment.topic();
     let req_partitions = assignment.partitions();
     let avail_partitions = match metadata.partitions(topic) {
         // ~ fail if the underlying topic is unkonwn to the given client
         None => {
-            debug!("determine_partitions: no such topic: {} (all metadata: {:?})",
-                   topic, metadata);
+            debug!("determine_partitions: no such topic: {} (all metadata: {:?})", topic, metadata);
             return Err(Error::Kafka(KafkaCode::UnknownTopicOrPartition));
         }
         Some(tp) => tp,
@@ -154,7 +164,10 @@ fn determine_partitions<'a>(assignment: &'a Assignment, metadata: Topics)
             match avail_partitions.partition(p) {
                 None => {
                     debug!("determine_partitions: no such partition: \"{}:{}\" \
-                            (all metadata: {:?})", topic, p, metadata);
+                            (all metadata: {:?})",
+                           topic,
+                           p,
+                           metadata);
                     return Err(Error::Kafka(KafkaCode::UnknownTopicOrPartition));
                 }
                 Some(_) => ps.push(p),
@@ -162,16 +175,20 @@ fn determine_partitions<'a>(assignment: &'a Assignment, metadata: Topics)
         }
         ps
     };
-    Ok(Subscription { assignment: assignment, partitions: ps })
+    Ok(Subscription {
+        assignment: assignment,
+        partitions: ps,
+    })
 }
 
 // Fetches the so-far commited/consumed offsets for the configured
 // group/topic/partitions.
-fn load_consumed_offsets(
-    client: &mut KafkaClient, group: &str, assignments: &Assignments,
-    subscriptions: &[Subscription], result_capacity: usize)
-    -> Result<HashMap<TopicPartition, ConsumedOffset, PartitionHasher>>
-{
+fn load_consumed_offsets(client: &mut KafkaClient,
+                         group: &str,
+                         assignments: &Assignments,
+                         subscriptions: &[Subscription],
+                         result_capacity: usize)
+                         -> Result<HashMap<TopicPartition, ConsumedOffset, PartitionHasher>> {
     assert!(!subscriptions.is_empty());
     // ~ pre-allocate the right size
     let mut offs = HashMap::with_capacity_and_hasher(result_capacity, PartitionHasher::default());
@@ -180,25 +197,26 @@ fn load_consumed_offsets(
         return Ok(offs);
     }
     // ~ otherwise try load them for the group
-    let tpos = try!(client.fetch_group_offsets(
-        group,
-        subscriptions.iter()
-            .flat_map(|s| {
-                let topic = s.assignment.topic();
-                s.partitions.iter().map(move |&p| FetchGroupOffset::new(topic, p))
-            })));
+    let tpos = try!(client.fetch_group_offsets(group,
+                                               subscriptions.iter()
+                                                   .flat_map(|s| {
+                                                       let topic = s.assignment.topic();
+                                                       s.partitions.iter().map(move |&p| {
+                                                           FetchGroupOffset::new(topic, p)
+                                                       })
+                                                   })));
     for tpo in tpos {
         match tpo.offset {
             Ok(o) if o != -1 => {
-                offs.insert(
-                    TopicPartition {
-                        topic_ref: assignments.topic_ref(&tpo.topic).expect("non-assigned topic"),
-                        partition: tpo.partition,
-                    },
-                    ConsumedOffset {
-                        offset: o,
-                        dirty: false,
-                    });
+                offs.insert(TopicPartition {
+                                topic_ref: assignments.topic_ref(&tpo.topic)
+                                    .expect("non-assigned topic"),
+                                partition: tpo.partition,
+                            },
+                            ConsumedOffset {
+                                offset: o,
+                                dirty: false,
+                            });
             }
             _ => {}
         }
@@ -208,20 +226,22 @@ fn load_consumed_offsets(
 
 /// Fetches the "next fetch" offsets/states based on the specified
 /// subscriptions and the given consumed offsets.
-fn load_fetch_states(
-    client: &mut KafkaClient, config: &Config,
-    assignments: &Assignments, subscriptions: &[Subscription],
-    consumed_offsets: &HashMap<TopicPartition, ConsumedOffset, PartitionHasher>,
-    result_capacity: usize)
-    -> Result<HashMap<TopicPartition, FetchState, PartitionHasher>>
-{
-    fn load_partition_offsets(client: &mut KafkaClient, topics: &[&str], offset: FetchOffset)
-                              -> Result<HashMap<String, HashMap<i32, i64, PartitionHasher>>>
-    {
+fn load_fetch_states(client: &mut KafkaClient,
+                     config: &Config,
+                     assignments: &Assignments,
+                     subscriptions: &[Subscription],
+                     consumed_offsets: &HashMap<TopicPartition, ConsumedOffset, PartitionHasher>,
+                     result_capacity: usize)
+                     -> Result<HashMap<TopicPartition, FetchState, PartitionHasher>> {
+    fn load_partition_offsets(client: &mut KafkaClient,
+                              topics: &[&str],
+                              offset: FetchOffset)
+                              -> Result<HashMap<String, HashMap<i32, i64, PartitionHasher>>> {
         let toffs = try!(client.fetch_offsets(topics, offset));
         let mut m = HashMap::with_capacity(toffs.len());
         for (topic, poffs) in toffs {
-            let mut pidx = HashMap::with_capacity_and_hasher(poffs.len(), PartitionHasher::default());
+            let mut pidx = HashMap::with_capacity_and_hasher(poffs.len(),
+                                                             PartitionHasher::default());
             for poff in poffs {
                 pidx.insert(poff.partition, poff.offset.unwrap_or(-1));
             }
@@ -230,15 +250,15 @@ fn load_fetch_states(
         Ok(m)
     }
 
-    let mut fetch_offsets = HashMap::with_capacity_and_hasher(
-        result_capacity, PartitionHasher::default());
+    let mut fetch_offsets = HashMap::with_capacity_and_hasher(result_capacity,
+                                                              PartitionHasher::default());
     let max_bytes = client.fetch_max_bytes_per_partition();
     let subscription_topics: Vec<_> = subscriptions.iter().map(|s| s.assignment.topic()).collect();
     if consumed_offsets.is_empty() {
         // ~ if there are no offsets on behalf of the consumer
         // group - if any - we can directly use the fallback offsets.
-        let offsets = try!(load_partition_offsets(
-            client, &subscription_topics, config.fallback_offset));
+        let offsets =
+            try!(load_partition_offsets(client, &subscription_topics, config.fallback_offset));
         for s in subscriptions {
             let topic_ref = assignments.topic_ref(s.assignment.topic())
                 .expect("unassigned subscription");
@@ -250,23 +270,24 @@ fn load_fetch_states(
                 }
                 Some(offsets) => {
                     for p in &s.partitions {
-                        fetch_offsets.insert(
-                            TopicPartition {
-                                topic_ref: topic_ref,
-                                partition: *p,
-                            },
-                            FetchState {
-                                offset: *offsets.get(p).unwrap_or(&-1),
-                                max_bytes: max_bytes,
-                            });
+                        fetch_offsets.insert(TopicPartition {
+                                                 topic_ref: topic_ref,
+                                                 partition: *p,
+                                             },
+                                             FetchState {
+                                                 offset: *offsets.get(p).unwrap_or(&-1),
+                                                 max_bytes: max_bytes,
+                                             });
                     }
                 }
             }
         }
     } else {
         // fetch the earliest and latest available offsets
-        let latest = try!(load_partition_offsets(client, &subscription_topics, FetchOffset::Latest));
-        let earliest = try!(load_partition_offsets(client, &subscription_topics, FetchOffset::Earliest));
+        let latest =
+            try!(load_partition_offsets(client, &subscription_topics, FetchOffset::Latest));
+        let earliest =
+            try!(load_partition_offsets(client, &subscription_topics, FetchOffset::Earliest));
         // ~ for each subscribed partition if we have a
         // consumed_offset verify it is in the earliest/latest range
         // and use that consumed_offset+1 as the fetch_message.
@@ -274,10 +295,15 @@ fn load_fetch_states(
             let topic_ref = assignments.topic_ref(s.assignment.topic())
                 .expect("unassigned subscription");
             for p in &s.partitions {
-                let l_off = *latest.get(s.assignment.topic()).and_then(|ps| ps.get(p)).unwrap_or(&-1);
-                let e_off = *earliest.get(s.assignment.topic()).and_then(|ps| ps.get(p)).unwrap_or(&-1);
+                let l_off =
+                    *latest.get(s.assignment.topic()).and_then(|ps| ps.get(p)).unwrap_or(&-1);
+                let e_off =
+                    *earliest.get(s.assignment.topic()).and_then(|ps| ps.get(p)).unwrap_or(&-1);
 
-                let tp = TopicPartition { topic_ref: topic_ref, partition: *p };
+                let tp = TopicPartition {
+                    topic_ref: topic_ref,
+                    partition: *p,
+                };
                 let offset = match consumed_offsets.get(&tp) {
                     Some(co) if co.offset >= e_off && co.offset < l_off => co.offset + 1,
                     _ => {
@@ -287,13 +313,19 @@ fn load_fetch_states(
                             _ => {
                                 debug!("cannot determine fetch offset \
                                         (group: {} / topic: {} / partition: {})",
-                                       &config.group, s.assignment.topic(), p);
+                                       &config.group,
+                                       s.assignment.topic(),
+                                       p);
                                 return Err(Error::Kafka(KafkaCode::Unknown));
                             }
                         }
                     }
                 };
-                fetch_offsets.insert(tp, FetchState { offset: offset, max_bytes: max_bytes });
+                fetch_offsets.insert(tp,
+                                     FetchState {
+                                         offset: offset,
+                                         max_bytes: max_bytes,
+                                     });
             }
         }
     }
diff --git a/src/error.rs b/src/error.rs
index d2ab4ddae..fb10fab6c 100644
--- a/src/error.rs
+++ b/src/error.rs
@@ -163,7 +163,7 @@ pub enum KafkaCode {
     InvalidTimestamp = 32,
     /// The broker does not support the requested SASL mechanism.
     UnsupportedSaslMechanism = 33,
-    ///	Request is not valid given the current SASL state.
+    /// Request is not valid given the current SASL state.
     IllegalSaslState = 34,
     /// The version of API is not supported.
     UnsupportedVersion = 35,
@@ -181,7 +181,9 @@ impl From<io::Error> for Error {
 
 #[cfg(feature = "security")]
 impl From<ssl::error::SslError> for Error {
-    fn from(err: ssl::error::SslError) -> Error { Error::Ssl(err) }
+    fn from(err: ssl::error::SslError) -> Error {
+        Error::Ssl(err)
+    }
 }
 
 impl Clone for Error {
@@ -190,16 +192,20 @@ impl Clone for Error {
             &Error::Io(ref err) => Error::Io(io::Error::new(err.kind(), "Io Error")),
             &Error::Kafka(x) => Error::Kafka(x),
             #[cfg(feature = "security")]
-            &Error::Ssl(ref x) => match x {
-                &ssl::error::SslError::StreamError(ref e) => 
-                    Error::Ssl(ssl::error::SslError::StreamError(
-                            io::Error::new(e.kind(), "Stream Error"))
-                    ),
-                &ssl::error::SslError::SslSessionClosed => 
-                    Error::Ssl(ssl::error::SslError::SslSessionClosed),
-                &ssl::error::SslError::OpenSslErrors(ref v) => 
-                    Error::Ssl(ssl::error::SslError::OpenSslErrors(v.clone())),
-            },
+            &Error::Ssl(ref x) => {
+                match x {
+                    &ssl::error::SslError::StreamError(ref e) => {
+                        Error::Ssl(ssl::error::SslError::StreamError(io::Error::new(e.kind(),
+                                                                                    "Stream Error")))
+                    }
+                    &ssl::error::SslError::SslSessionClosed => {
+                        Error::Ssl(ssl::error::SslError::SslSessionClosed)
+                    }
+                    &ssl::error::SslError::OpenSslErrors(ref v) => {
+                        Error::Ssl(ssl::error::SslError::OpenSslErrors(v.clone()))
+                    }
+                }
+            }
             &Error::UnsupportedProtocol => Error::UnsupportedProtocol,
             &Error::UnsupportedCompression => Error::UnsupportedCompression,
             #[cfg(feature = "snappy")]
@@ -237,7 +243,7 @@ impl error::Error for Error {
     fn cause(&self) -> Option<&error::Error> {
         match *self {
             Error::Io(ref err) => err.cause(),
-            _ => None
+            _ => None,
         }
     }
 }
diff --git a/src/lib.rs b/src/lib.rs
index 5a22c559e..a65da04ad 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -3,7 +3,8 @@
 //!
 //! - `kafka::producer::Producer` - for sending message to Kafka
 //! - `kafka::consumer::Consumer` - for retrieving/consuming messages from Kafka
-//! - `kafka::client::KafkaClient` - a lower-level, general purpose client leaving you with more power but also more resposibility
+//! - `kafka::client::KafkaClient` - a lower-level, general purpose client leaving
+//!   you with more power but also more resposibility
 //!
 //! See module level documentation corresponding to each client individually.
 
@@ -42,8 +43,11 @@ mod compression;
 pub use self::error::{Error, Result};
 
 trait KafkaClientInternals {
-    fn internal_produce_messages<'a, 'b, I, J>(
-        &mut self, ack: client::RequiredAcks, ack_timeout: i32, messages: I)
-        -> Result<Vec<client::TopicPartitionOffset>>
-        where J: AsRef<client::ProduceMessage<'a, 'b>>, I: IntoIterator<Item=J>;
+    fn internal_produce_messages<'a, 'b, I, J>(&mut self,
+                                               ack: client::RequiredAcks,
+                                               ack_timeout: i32,
+                                               messages: I)
+                                               -> Result<Vec<client::TopicPartitionOffset>>
+        where J: AsRef<client::ProduceMessage<'a, 'b>>,
+              I: IntoIterator<Item = J>;
 }
diff --git a/src/producer.rs b/src/producer.rs
index d0271304a..b7d48025c 100644
--- a/src/producer.rs
+++ b/src/producer.rs
@@ -276,28 +276,23 @@ impl<P: Partitioner> Producer<P> {
         let client = &mut self.client;
         let config = &self.config;
 
-        client.internal_produce_messages(
-            config.required_acks,
-            config.ack_timeout,
-            recs.into_iter().map(|r| {
-                let mut m = client::ProduceMessage {
-                    key: to_option(r.key.as_bytes()),
-                    value: to_option(r.value.as_bytes()),
-                    topic: r.topic,
-                    partition: r.partition,
-                };
-                partitioner.partition(Topics::new(partitions), &mut m);
-                m
-            }))
+        client.internal_produce_messages(config.required_acks,
+                                         config.ack_timeout,
+                                         recs.into_iter().map(|r| {
+            let mut m = client::ProduceMessage {
+                key: to_option(r.key.as_bytes()),
+                value: to_option(r.value.as_bytes()),
+                topic: r.topic,
+                partition: r.partition,
+            };
+            partitioner.partition(Topics::new(partitions), &mut m);
+            m
+        }))
     }
 }
 
 fn to_option(data: &[u8]) -> Option<&[u8]> {
-    if data.is_empty() {
-        None
-    } else {
-        Some(data)
-    }
+    if data.is_empty() { None } else { Some(data) }
 }
 
 // --------------------------------------------------------------------
@@ -343,7 +338,8 @@ impl Builder {
             hosts: hosts,
             compression: client::DEFAULT_COMPRESSION,
             ack_timeout: Duration::from_millis(DEFAULT_ACK_TIMEOUT_MILLIS),
-            conn_idle_timeout: Duration::from_millis(client::DEFAULT_CONNECTION_IDLE_TIMEOUT_MILLIS),
+            conn_idle_timeout:
+                Duration::from_millis(client::DEFAULT_CONNECTION_IDLE_TIMEOUT_MILLIS),
             required_acks: DEFAULT_REQUIRED_ACKS,
             partitioner: DefaultPartitioner::default(),
             security_config: None,
@@ -433,9 +429,7 @@ impl<P> Builder<P> {
         // ~ create the client if necessary
         let (mut client, need_metadata) = match self.client {
             Some(client) => (client, false),
-            None => {
-                (Self::new_kafka_client(self.hosts, self.security_config), true)
-            }
+            None => (Self::new_kafka_client(self.hosts, self.security_config), true),
         };
         // ~ apply configuration settings
         client.set_compression(self.compression);
diff --git a/src/protocol/consumer.rs b/src/protocol/consumer.rs
index bb2c97e63..9a2a85acd 100644
--- a/src/protocol/consumer.rs
+++ b/src/protocol/consumer.rs
@@ -16,12 +16,15 @@ pub struct GroupCoordinatorRequest<'a, 'b> {
 }
 
 impl<'a, 'b> GroupCoordinatorRequest<'a, 'b> {
-    pub fn new(group: &'b str, correlation_id: i32, client_id: &'a str)
-               -> GroupCoordinatorRequest<'a, 'b>
-    {
+    pub fn new(group: &'b str,
+               correlation_id: i32,
+               client_id: &'a str)
+               -> GroupCoordinatorRequest<'a, 'b> {
         GroupCoordinatorRequest {
-            header: HeaderRequest::new(
-                API_KEY_GROUP_COORDINATOR, API_VERSION, correlation_id, client_id),
+            header: HeaderRequest::new(API_KEY_GROUP_COORDINATOR,
+                                       API_VERSION,
+                                       correlation_id,
+                                       client_id),
             group: group,
         }
     }
@@ -29,9 +32,7 @@ impl<'a, 'b> GroupCoordinatorRequest<'a, 'b> {
 
 impl<'a, 'b> ToByte for GroupCoordinatorRequest<'a, 'b> {
     fn encode<W: Write>(&self, buffer: &mut W) -> Result<()> {
-        try_multi!(
-            self.header.encode(buffer),
-            self.group.encode(buffer))
+        try_multi!(self.header.encode(buffer), self.group.encode(buffer))
     }
 }
 
@@ -57,12 +58,11 @@ impl FromByte for GroupCoordinatorResponse {
     type R = GroupCoordinatorResponse;
 
     fn decode<T: Read>(&mut self, buffer: &mut T) -> Result<()> {
-        try_multi!(
-            self.header.decode(buffer),
-            self.error.decode(buffer),
-            self.broker_id.decode(buffer),
-            self.host.decode(buffer),
-            self.port.decode(buffer))
+        try_multi!(self.header.decode(buffer),
+                   self.error.decode(buffer),
+                   self.broker_id.decode(buffer),
+                   self.host.decode(buffer),
+                   self.port.decode(buffer))
     }
 }
 
@@ -81,18 +81,18 @@ pub enum OffsetFetchVersion {
 pub struct OffsetFetchRequest<'a, 'b, 'c> {
     pub header: HeaderRequest<'a>,
     pub group: &'b str,
-    pub topic_partitions: Vec<TopicPartitionOffsetFetchRequest<'c>>
+    pub topic_partitions: Vec<TopicPartitionOffsetFetchRequest<'c>>,
 }
 
 #[derive(Debug)]
 pub struct TopicPartitionOffsetFetchRequest<'a> {
     pub topic: &'a str,
-    pub partitions: Vec<PartitionOffsetFetchRequest>
+    pub partitions: Vec<PartitionOffsetFetchRequest>,
 }
 
 #[derive(Debug)]
 pub struct PartitionOffsetFetchRequest {
-    pub partition: i32
+    pub partition: i32,
 }
 
 impl<'a, 'b, 'c> OffsetFetchRequest<'a, 'b, 'c> {
@@ -100,13 +100,14 @@ impl<'a, 'b, 'c> OffsetFetchRequest<'a, 'b, 'c> {
                version: OffsetFetchVersion,
                correlation_id: i32,
                client_id: &'a str)
-               -> OffsetFetchRequest<'a, 'b, 'c>
-    {
+               -> OffsetFetchRequest<'a, 'b, 'c> {
         OffsetFetchRequest {
-            header: HeaderRequest::new(
-                API_KEY_OFFSET_FETCH, version as i16, correlation_id, client_id),
+            header: HeaderRequest::new(API_KEY_OFFSET_FETCH,
+                                       version as i16,
+                                       correlation_id,
+                                       client_id),
             group: group,
-            topic_partitions: vec!()
+            topic_partitions: vec![],
         }
     }
 
@@ -125,7 +126,10 @@ impl<'a, 'b, 'c> OffsetFetchRequest<'a, 'b, 'c> {
 
 impl<'a> TopicPartitionOffsetFetchRequest<'a> {
     pub fn new(topic: &'a str) -> TopicPartitionOffsetFetchRequest<'a> {
-        TopicPartitionOffsetFetchRequest{topic: topic, partitions: vec!()}
+        TopicPartitionOffsetFetchRequest {
+            topic: topic,
+            partitions: vec![],
+        }
     }
 
     pub fn add(&mut self, partition: i32) {
@@ -135,26 +139,21 @@ impl<'a> TopicPartitionOffsetFetchRequest<'a> {
 
 impl PartitionOffsetFetchRequest {
     pub fn new(partition: i32) -> PartitionOffsetFetchRequest {
-        PartitionOffsetFetchRequest{partition: partition}
+        PartitionOffsetFetchRequest { partition: partition }
     }
 }
 
 impl<'a, 'b, 'c> ToByte for OffsetFetchRequest<'a, 'b, 'c> {
     fn encode<W: Write>(&self, buffer: &mut W) -> Result<()> {
-        try_multi!(
-            self.header.encode(buffer),
-            self.group.encode(buffer),
-            self.topic_partitions.encode(buffer)
-        )
+        try_multi!(self.header.encode(buffer),
+                   self.group.encode(buffer),
+                   self.topic_partitions.encode(buffer))
     }
 }
 
 impl<'a> ToByte for TopicPartitionOffsetFetchRequest<'a> {
     fn encode<W: Write>(&self, buffer: &mut W) -> Result<()> {
-        try_multi!(
-            self.topic.encode(buffer),
-            self.partitions.encode(buffer)
-        )
+        try_multi!(self.topic.encode(buffer), self.partitions.encode(buffer))
     }
 }
 
@@ -169,13 +168,13 @@ impl ToByte for PartitionOffsetFetchRequest {
 #[derive(Default, Debug)]
 pub struct OffsetFetchResponse {
     pub header: HeaderResponse,
-    pub topic_partitions: Vec<TopicPartitionOffsetFetchResponse>
+    pub topic_partitions: Vec<TopicPartitionOffsetFetchResponse>,
 }
 
 #[derive(Default, Debug)]
 pub struct TopicPartitionOffsetFetchResponse {
     pub topic: String,
-    pub partitions: Vec<PartitionOffsetFetchResponse>
+    pub partitions: Vec<PartitionOffsetFetchResponse>,
 }
 
 #[derive(Default, Debug)]
@@ -183,7 +182,7 @@ pub struct PartitionOffsetFetchResponse {
     pub partition: i32,
     pub offset: i64,
     pub metadata: String,
-    pub error: i16
+    pub error: i16,
 }
 
 impl PartitionOffsetFetchResponse {
@@ -194,7 +193,7 @@ impl PartitionOffsetFetchResponse {
             offset: match Error::from_protocol(self.error) {
                 None => Ok(self.offset),
                 Some(e) => Err(e),
-            }
+            },
         }
     }
 }
@@ -203,10 +202,7 @@ impl FromByte for OffsetFetchResponse {
     type R = OffsetFetchResponse;
 
     fn decode<T: Read>(&mut self, buffer: &mut T) -> Result<()> {
-        try_multi!(
-            self.header.decode(buffer),
-            self.topic_partitions.decode(buffer)
-        )
+        try_multi!(self.header.decode(buffer), self.topic_partitions.decode(buffer))
     }
 }
 
@@ -214,10 +210,7 @@ impl FromByte for TopicPartitionOffsetFetchResponse {
     type R = TopicPartitionOffsetFetchResponse;
 
     fn decode<T: Read>(&mut self, buffer: &mut T) -> Result<()> {
-        try_multi!(
-            self.topic.decode(buffer),
-            self.partitions.decode(buffer)
-        )
+        try_multi!(self.topic.decode(buffer), self.partitions.decode(buffer))
     }
 }
 
@@ -225,12 +218,10 @@ impl FromByte for PartitionOffsetFetchResponse {
     type R = PartitionOffsetFetchResponse;
 
     fn decode<T: Read>(&mut self, buffer: &mut T) -> Result<()> {
-        try_multi!(
-            self.partition.decode(buffer),
-            self.offset.decode(buffer),
-            self.metadata.decode(buffer),
-            self.error.decode(buffer)
-        )
+        try_multi!(self.partition.decode(buffer),
+                   self.offset.decode(buffer),
+                   self.metadata.decode(buffer),
+                   self.error.decode(buffer))
     }
 }
 
@@ -263,13 +254,13 @@ impl OffsetCommitVersion {
 pub struct OffsetCommitRequest<'a, 'b> {
     pub header: HeaderRequest<'a>,
     pub group: &'b str,
-    pub topic_partitions: Vec<TopicPartitionOffsetCommitRequest<'b>>
+    pub topic_partitions: Vec<TopicPartitionOffsetCommitRequest<'b>>,
 }
 
 #[derive(Debug)]
 pub struct TopicPartitionOffsetCommitRequest<'a> {
     pub topic: &'a str,
-    pub partitions: Vec<PartitionOffsetCommitRequest<'a>>
+    pub partitions: Vec<PartitionOffsetCommitRequest<'a>>,
 }
 
 #[derive(Debug)]
@@ -284,13 +275,14 @@ impl<'a, 'b> OffsetCommitRequest<'a, 'b> {
                version: OffsetCommitVersion,
                correlation_id: i32,
                client_id: &'a str)
-               -> OffsetCommitRequest<'a, 'b>
-    {
-        OffsetCommitRequest{
-            header: HeaderRequest::new(
-                API_KEY_OFFSET_COMMIT, version as i16, correlation_id, client_id),
+               -> OffsetCommitRequest<'a, 'b> {
+        OffsetCommitRequest {
+            header: HeaderRequest::new(API_KEY_OFFSET_COMMIT,
+                                       version as i16,
+                                       correlation_id,
+                                       client_id),
             group: group,
-            topic_partitions: vec!()
+            topic_partitions: vec![],
         }
     }
 
@@ -311,7 +303,7 @@ impl<'a> TopicPartitionOffsetCommitRequest<'a> {
     pub fn new(topic: &'a str) -> TopicPartitionOffsetCommitRequest<'a> {
         TopicPartitionOffsetCommitRequest {
             topic: topic,
-            partitions: vec!()
+            partitions: vec![],
         }
     }
 
@@ -321,13 +313,11 @@ impl<'a> TopicPartitionOffsetCommitRequest<'a> {
 }
 
 impl<'a> PartitionOffsetCommitRequest<'a> {
-    pub fn new(partition: i32, offset: i64, metadata: &'a str)
-               -> PartitionOffsetCommitRequest<'a>
-    {
+    pub fn new(partition: i32, offset: i64, metadata: &'a str) -> PartitionOffsetCommitRequest<'a> {
         PartitionOffsetCommitRequest {
             partition: partition,
             offset: offset,
-            metadata: metadata
+            metadata: metadata,
         }
     }
 }
@@ -347,20 +337,20 @@ impl<'a, 'b> ToByte for OffsetCommitRequest<'a, 'b> {
                 try!("".encode(buffer));
                 try!((-1i64).encode(buffer));
             }
-            _ => { /* nothing to do */ }
+            _ => {
+                // nothing to do
+            }
         }
         codecs::encode_as_array(buffer, &self.topic_partitions, |buffer, tp| {
-            try_multi!(
-                tp.topic.encode(buffer),
-                codecs::encode_as_array(buffer, &tp.partitions, |buffer, p| {
-                    try!(p.partition.encode(buffer));
-                    try!(p.offset.encode(buffer));
-                    if v == OffsetCommitVersion::V1 {
-                        try!((-1i64).encode(buffer));
-                    }
-                    p.metadata.encode(buffer)
-                })
-            )
+            try_multi!(tp.topic.encode(buffer),
+                       codecs::encode_as_array(buffer, &tp.partitions, |buffer, p| {
+                try!(p.partition.encode(buffer));
+                try!(p.offset.encode(buffer));
+                if v == OffsetCommitVersion::V1 {
+                    try!((-1i64).encode(buffer));
+                }
+                p.metadata.encode(buffer)
+            }))
         })
     }
 }
@@ -370,39 +360,35 @@ impl<'a, 'b> ToByte for OffsetCommitRequest<'a, 'b> {
 #[derive(Default, Debug)]
 pub struct OffsetCommitResponse {
     pub header: HeaderResponse,
-    pub topic_partitions: Vec<TopicPartitionOffsetCommitResponse>
+    pub topic_partitions: Vec<TopicPartitionOffsetCommitResponse>,
 }
 
 impl FromByte for OffsetCommitResponse {
     type R = OffsetCommitResponse;
 
     fn decode<T: Read>(&mut self, buffer: &mut T) -> Result<()> {
-        try_multi!(
-            self.header.decode(buffer),
-            self.topic_partitions.decode(buffer))
+        try_multi!(self.header.decode(buffer), self.topic_partitions.decode(buffer))
     }
 }
 
 #[derive(Default, Debug)]
 pub struct TopicPartitionOffsetCommitResponse {
     pub topic: String,
-    pub partitions: Vec<PartitionOffsetCommitResponse>
+    pub partitions: Vec<PartitionOffsetCommitResponse>,
 }
 
 impl FromByte for TopicPartitionOffsetCommitResponse {
     type R = TopicPartitionOffsetCommitResponse;
 
     fn decode<T: Read>(&mut self, buffer: &mut T) -> Result<()> {
-        try_multi!(
-            self.topic.decode(buffer),
-            self.partitions.decode(buffer))
+        try_multi!(self.topic.decode(buffer), self.partitions.decode(buffer))
     }
 }
 
 #[derive(Default, Debug)]
 pub struct PartitionOffsetCommitResponse {
     pub partition: i32,
-    pub error: i16
+    pub error: i16,
 }
 
 impl PartitionOffsetCommitResponse {
@@ -415,8 +401,6 @@ impl FromByte for PartitionOffsetCommitResponse {
     type R = PartitionOffsetCommitResponse;
 
     fn decode<T: Read>(&mut self, buffer: &mut T) -> Result<()> {
-        try_multi!(
-            self.partition.decode(buffer),
-            self.error.decode(buffer))
+        try_multi!(self.partition.decode(buffer), self.error.decode(buffer))
     }
 }
diff --git a/src/protocol/fetch.rs b/src/protocol/fetch.rs
index 4ad803e33..ca4be712d 100644
--- a/src/protocol/fetch.rs
+++ b/src/protocol/fetch.rs
@@ -36,22 +36,23 @@ pub struct FetchRequest<'a, 'b> {
 #[derive(Debug)]
 pub struct TopicPartitionFetchRequest {
     // partition-id -> partition-data
-    pub partitions: HashMap<i32, PartitionFetchRequest, PartitionHasher>
+    pub partitions: HashMap<i32, PartitionFetchRequest, PartitionHasher>,
 }
 
 #[derive(Debug)]
 pub struct PartitionFetchRequest {
     pub offset: i64,
-    pub max_bytes: i32
+    pub max_bytes: i32,
 }
 
 impl<'a, 'b> FetchRequest<'a, 'b> {
-
-    pub fn new(correlation_id: i32, client_id: &'a str, max_wait_time: i32, min_bytes: i32)
+    pub fn new(correlation_id: i32,
+               client_id: &'a str,
+               max_wait_time: i32,
+               min_bytes: i32)
                -> FetchRequest<'a, 'b> {
         FetchRequest {
-            header: HeaderRequest::new(
-                API_KEY_FETCH, API_VERSION, correlation_id, client_id),
+            header: HeaderRequest::new(API_KEY_FETCH, API_VERSION, correlation_id, client_id),
             replica: -1,
             max_wait_time: max_wait_time,
             min_bytes: min_bytes,
@@ -60,7 +61,8 @@ impl<'a, 'b> FetchRequest<'a, 'b> {
     }
 
     pub fn add(&mut self, topic: &'b str, partition: i32, offset: i64, max_bytes: i32) {
-        self.topic_partitions.entry(topic)
+        self.topic_partitions
+            .entry(topic)
             .or_insert_with(|| TopicPartitionFetchRequest::new())
             .add(partition, offset, max_bytes)
     }
@@ -121,19 +123,18 @@ impl TopicPartitionFetchRequest {
 }
 
 impl PartitionFetchRequest {
-    fn encode<T:Write>(&self, partition: i32, buffer: &mut T) -> Result<()> {
-        try_multi!(
-            partition.encode(buffer),
-            self.offset.encode(buffer),
-            self.max_bytes.encode(buffer)
-        )
+    fn encode<T: Write>(&self, partition: i32, buffer: &mut T) -> Result<()> {
+        try_multi!(partition.encode(buffer),
+                   self.offset.encode(buffer),
+                   self.max_bytes.encode(buffer))
     }
 }
 
 // ~ response related -------------------------------------------------
 
 pub struct ResponseParser<'a, 'b, 'c>
-    where 'a: 'c, 'b: 'c
+    where 'a: 'c,
+          'b: 'c
 {
     pub validate_crc: bool,
     pub requests: Option<&'c FetchRequest<'a, 'b>>,
@@ -180,9 +181,10 @@ pub struct Response {
 impl Response {
     /// Parses a Response from binary data as defined by the
     /// Kafka Protocol.
-    fn from_vec(response: Vec<u8>, reqs: Option<&FetchRequest>, validate_crc: bool)
-                -> Result<Response>
-    {
+    fn from_vec(response: Vec<u8>,
+                reqs: Option<&FetchRequest>,
+                validate_crc: bool)
+                -> Result<Response> {
         let slice = unsafe { mem::transmute(&response[..]) };
         let mut r = ZReader::new(slice);
         let correlation_id = try!(r.read_i32());
@@ -190,7 +192,7 @@ impl Response {
         Ok(Response {
             raw_data: response,
             correlation_id: correlation_id,
-            topics: topics
+            topics: topics,
         })
     }
 
@@ -219,9 +221,10 @@ pub struct Topic<'a> {
 }
 
 impl<'a> Topic<'a> {
-    fn read(r: &mut ZReader<'a>, reqs: Option<&FetchRequest>, validate_crc: bool)
-            -> Result<Topic<'a>>
-    {
+    fn read(r: &mut ZReader<'a>,
+            reqs: Option<&FetchRequest>,
+            validate_crc: bool)
+            -> Result<Topic<'a>> {
         let name = try!(r.read_str());
         let preqs = reqs.and_then(|reqs| reqs.get(name));
         let partitions = array_of!(r, Partition::read(r, preqs, validate_crc));
@@ -261,9 +264,10 @@ pub struct Partition<'a> {
 }
 
 impl<'a> Partition<'a> {
-    fn read(r: &mut ZReader<'a>, preqs: Option<&TopicPartitionFetchRequest>, validate_crc: bool)
-            -> Result<Partition<'a>>
-    {
+    fn read(r: &mut ZReader<'a>,
+            preqs: Option<&TopicPartitionFetchRequest>,
+            validate_crc: bool)
+            -> Result<Partition<'a>> {
         let partition = try!(r.read_i32());
         let proffs = preqs.and_then(|preqs| preqs.get(partition))
             .map(|preq| preq.offset)
@@ -277,11 +281,13 @@ impl<'a> Partition<'a> {
             partition: partition,
             data: match error {
                 Some(error) => Err(error),
-                None => Ok(Data {
-                    highwatermark_offset: highwatermark,
-                    message_set: msgset,
-                    first_message_idx: Cell::new(0),
-                }),
+                None => {
+                    Ok(Data {
+                        highwatermark_offset: highwatermark,
+                        message_set: msgset,
+                        first_message_idx: Cell::new(0),
+                    })
+                }
             },
         })
     }
@@ -356,18 +362,19 @@ impl<'a> MessageSet<'a> {
         // further modifying it and providing
         // publicly no mutability possibilities
         // this is safe
-        let ms = try!(MessageSet::from_slice(unsafe {
-            mem::transmute(&data[..])
-        }, req_offset, validate_crc));
+        let ms = try!(MessageSet::from_slice(unsafe { mem::transmute(&data[..]) },
+                                             req_offset,
+                                             validate_crc));
         return Ok(MessageSet {
             raw_data: Cow::Owned(data),
             messages: ms.messages,
         });
     }
 
-    fn from_slice<'b>(raw_data: &'b [u8], req_offset: i64, validate_crc: bool)
-                      -> Result<MessageSet<'b>>
-    {
+    fn from_slice<'b>(raw_data: &'b [u8],
+                      req_offset: i64,
+                      validate_crc: bool)
+                      -> Result<MessageSet<'b>> {
         let mut r = ZReader::new(raw_data);
         let mut msgs = Vec::new();
         while !r.is_empty() {
@@ -392,7 +399,7 @@ impl<'a> MessageSet<'a> {
                                 msgs.push(Message {
                                     offset: offset,
                                     key: pmsg.key,
-                                    value: pmsg.value
+                                    value: pmsg.value,
                                 });
                             }
                         }
@@ -420,7 +427,9 @@ impl<'a> MessageSet<'a> {
         })
     }
 
-    fn next_message<'b>(r: &mut ZReader<'b>, validate_crc: bool) -> Result<(i64, ProtocolMessage<'b>)> {
+    fn next_message<'b>(r: &mut ZReader<'b>,
+                        validate_crc: bool)
+                        -> Result<(i64, ProtocolMessage<'b>)> {
         let offset = try!(r.read_i64());
         let msg_data = try!(r.read_bytes());
         Ok((offset, try!(ProtocolMessage::from_slice(msg_data, validate_crc))))
@@ -474,8 +483,7 @@ mod tests {
     use super::{FetchRequest, Response, Message};
     use error::{Error, KafkaCode};
 
-    static FETCH1_TXT: &'static str =
-        include_str!("../../test-data/fetch1.txt");
+    static FETCH1_TXT: &'static str = include_str!("../../test-data/fetch1.txt");
     static FETCH1_FETCH_RESPONSE_NOCOMPRESSION_K0821: &'static [u8] =
         include_bytes!("../../test-data/fetch1.mytopic.1p.nocompression.kafka.0821");
     static FETCH1_FETCH_RESPONSE_SNAPPY_K0821: &'static [u8] =
@@ -487,8 +495,7 @@ mod tests {
     static FETCH1_FETCH_RESPONSE_GZIP_K0821: &'static [u8] =
         include_bytes!("../../test-data/fetch1.mytopic.1p.gzip.kafka.0821");
 
-    static FETCH2_TXT: &'static str =
-        include_str!("../../test-data/fetch2.txt");
+    static FETCH2_TXT: &'static str = include_str!("../../test-data/fetch2.txt");
     static FETCH2_FETCH_RESPONSE_NOCOMPRESSION_K0900: &'static [u8] =
         include_bytes!("../../test-data/fetch2.mytopic.nocompression.kafka.0900");
     static FETCH2_FETCH_RESPONSE_NOCOMPRESSION_INVALID_CRC_K0900: &'static [u8] =
@@ -511,7 +518,10 @@ mod tests {
         all_msgs
     }
 
-    fn test_decode_new_fetch_response(msg_per_line: &str, response: Vec<u8>, requests: Option<&FetchRequest>, validate_crc: bool) {
+    fn test_decode_new_fetch_response(msg_per_line: &str,
+                                      response: Vec<u8>,
+                                      requests: Option<&FetchRequest>,
+                                      validate_crc: bool) {
         let resp = Response::from_vec(response, requests, validate_crc);
         let resp = resp.unwrap();
 
@@ -538,7 +548,7 @@ mod tests {
     fn skip_lines(mut lines: &str, mut n: usize) -> &str {
         while n > 0 {
             n -= 1;
-            lines = &lines[lines.find('\n').map(|i| i+1).unwrap_or(0) ..];
+            lines = &lines[lines.find('\n').map(|i| i + 1).unwrap_or(0)..];
         }
         lines
     }
@@ -548,15 +558,19 @@ mod tests {
         let mut req = FetchRequest::new(0, "test", -1, -1);
         req.add("my-topic", 0, 0, -1);
         req.add("foo-quux", 0, 100, -1);
-        test_decode_new_fetch_response(
-            FETCH1_TXT, FETCH1_FETCH_RESPONSE_NOCOMPRESSION_K0821.to_owned(), Some(&req), false);
+        test_decode_new_fetch_response(FETCH1_TXT,
+                                       FETCH1_FETCH_RESPONSE_NOCOMPRESSION_K0821.to_owned(),
+                                       Some(&req),
+                                       false);
 
         // ~ pretend we asked for messages as of offset five (while
         // the server delivered the zero-offset message as well)
         req = FetchRequest::new(0, "test", -1, -1);
         req.add("my-topic", 0, 5, -1);
-        test_decode_new_fetch_response(
-            skip_lines(FETCH1_TXT, 5), FETCH1_FETCH_RESPONSE_NOCOMPRESSION_K0821.to_owned(), Some(&req), false);
+        test_decode_new_fetch_response(skip_lines(FETCH1_TXT, 5),
+                                       FETCH1_FETCH_RESPONSE_NOCOMPRESSION_K0821.to_owned(),
+                                       Some(&req),
+                                       false);
     }
 
     // verify we don't crash but cleanly fail and report we don't
@@ -566,8 +580,12 @@ mod tests {
     fn test_unsupported_compression_snappy() {
         let mut req = FetchRequest::new(0, "test", -1, -1);
         req.add("my-topic", 0, 0, -1);
-        let r = Response::from_vec(FETCH1_FETCH_RESPONSE_SNAPPY_K0821.to_owned(), Some(&req), false);
-        assert!(match r { Err(Error::UnsupportedCompression) => true, _ => false });
+        let r =
+            Response::from_vec(FETCH1_FETCH_RESPONSE_SNAPPY_K0821.to_owned(), Some(&req), false);
+        assert!(match r {
+            Err(Error::UnsupportedCompression) => true,
+            _ => false,
+        });
     }
 
     #[cfg(feature = "snappy")]
@@ -575,15 +593,19 @@ mod tests {
     fn test_from_slice_snappy_k0821() {
         let mut req = FetchRequest::new(0, "test", -1, -1);
         req.add("my-topic", 0, 0, -1);
-        test_decode_new_fetch_response(
-            FETCH1_TXT, FETCH1_FETCH_RESPONSE_SNAPPY_K0821.to_owned(), Some(&req), false);
+        test_decode_new_fetch_response(FETCH1_TXT,
+                                       FETCH1_FETCH_RESPONSE_SNAPPY_K0821.to_owned(),
+                                       Some(&req),
+                                       false);
 
         // ~ pretend we asked for messages as of offset three (while
         // the server delivered the zero-offset message as well)
         req = FetchRequest::new(0, "test", -1, -1);
         req.add("my-topic", 0, 3, -1);
-        test_decode_new_fetch_response(
-            skip_lines(FETCH1_TXT, 3), FETCH1_FETCH_RESPONSE_SNAPPY_K0821.to_owned(), Some(&req), false);
+        test_decode_new_fetch_response(skip_lines(FETCH1_TXT, 3),
+                                       FETCH1_FETCH_RESPONSE_SNAPPY_K0821.to_owned(),
+                                       Some(&req),
+                                       false);
     }
 
     #[cfg(feature = "snappy")]
@@ -591,8 +613,10 @@ mod tests {
     fn test_from_slice_snappy_k0822() {
         let mut req = FetchRequest::new(0, "test", -1, -1);
         req.add("my-topic", 0, 0, -1);
-        test_decode_new_fetch_response(
-            FETCH1_TXT, FETCH1_FETCH_RESPONSE_SNAPPY_K0822.to_owned(), Some(&req), false);
+        test_decode_new_fetch_response(FETCH1_TXT,
+                                       FETCH1_FETCH_RESPONSE_SNAPPY_K0822.to_owned(),
+                                       Some(&req),
+                                       false);
     }
 
     #[cfg(feature = "gzip")]
@@ -600,34 +624,49 @@ mod tests {
     fn test_from_slice_gzip_k0821() {
         let mut req = FetchRequest::new(0, "test", -1, -1);
         req.add("my-topic", 0, 0, -1);
-        test_decode_new_fetch_response(
-            FETCH1_TXT, FETCH1_FETCH_RESPONSE_GZIP_K0821.to_owned(), Some(&req), false);
+        test_decode_new_fetch_response(FETCH1_TXT,
+                                       FETCH1_FETCH_RESPONSE_GZIP_K0821.to_owned(),
+                                       Some(&req),
+                                       false);
 
         // ~ pretend we asked for messages as of offset one (while the
         // server delivered the zero-offset message as well)
         req.add("my-topic", 0, 1, -1);
-        test_decode_new_fetch_response(
-            skip_lines(FETCH1_TXT, 1), FETCH1_FETCH_RESPONSE_GZIP_K0821.to_owned(), Some(&req), false);
+        test_decode_new_fetch_response(skip_lines(FETCH1_TXT, 1),
+                                       FETCH1_FETCH_RESPONSE_GZIP_K0821.to_owned(),
+                                       Some(&req),
+                                       false);
 
         // ~ pretend we asked for messages as of offset ten (while the
         // server delivered the zero-offset message as well)
         req.add("my-topic", 0, 10, -1);
-        test_decode_new_fetch_response(
-            skip_lines(FETCH1_TXT, 10), FETCH1_FETCH_RESPONSE_GZIP_K0821.to_owned(), Some(&req), false);
+        test_decode_new_fetch_response(skip_lines(FETCH1_TXT, 10),
+                                       FETCH1_FETCH_RESPONSE_GZIP_K0821.to_owned(),
+                                       Some(&req),
+                                       false);
     }
 
     #[test]
     fn test_crc_validation() {
-        test_decode_new_fetch_response(FETCH2_TXT, FETCH2_FETCH_RESPONSE_NOCOMPRESSION_K0900.to_owned(), None, true);
+        test_decode_new_fetch_response(FETCH2_TXT,
+                                       FETCH2_FETCH_RESPONSE_NOCOMPRESSION_K0900.to_owned(),
+                                       None,
+                                       true);
 
         // now test the same message but with an invalid crc checksum
         // (modified by hand)
         // 1) without checking the crc ... since only the crc field is
         // artificially falsified ... we expect the rest of the
         // message to be parsed correctly
-        test_decode_new_fetch_response(FETCH2_TXT, FETCH2_FETCH_RESPONSE_NOCOMPRESSION_INVALID_CRC_K0900.to_owned(), None, false);
+        test_decode_new_fetch_response(
+            FETCH2_TXT,
+            FETCH2_FETCH_RESPONSE_NOCOMPRESSION_INVALID_CRC_K0900.to_owned(),
+            None,
+            false);
         // 2) with checking the crc ... parsing should fail immediately
-        match Response::from_vec(FETCH2_FETCH_RESPONSE_NOCOMPRESSION_INVALID_CRC_K0900.to_owned(), None, true) {
+        match Response::from_vec(FETCH2_FETCH_RESPONSE_NOCOMPRESSION_INVALID_CRC_K0900.to_owned(),
+                                 None,
+                                 true) {
             Ok(_) => panic!("Expected error, but got successful response!"),
             Err(Error::Kafka(KafkaCode::CorruptMessage)) => {}
             Err(e) => panic!("Expected KafkaCode::CorruptMessage error, but got: {:?}", e),
@@ -655,48 +694,66 @@ mod tests {
 
         #[bench]
         fn bench_decode_new_fetch_response_nocompression_k0821(b: &mut Bencher) {
-            bench_decode_new_fetch_response(b, super::FETCH1_FETCH_RESPONSE_NOCOMPRESSION_K0821.to_owned(), false)
+            bench_decode_new_fetch_response(
+                b,
+                super::FETCH1_FETCH_RESPONSE_NOCOMPRESSION_K0821.to_owned(),
+                false)
         }
 
         #[cfg(feature = "snappy")]
         #[bench]
         fn bench_decode_new_fetch_response_snappy_k0821(b: &mut Bencher) {
-            bench_decode_new_fetch_response(b, super::FETCH1_FETCH_RESPONSE_SNAPPY_K0821.to_owned(), false)
+            bench_decode_new_fetch_response(b,
+                                            super::FETCH1_FETCH_RESPONSE_SNAPPY_K0821.to_owned(),
+                                            false)
         }
 
         #[cfg(feature = "snappy")]
         #[bench]
         fn bench_decode_new_fetch_response_snappy_k0822(b: &mut Bencher) {
-            bench_decode_new_fetch_response(b, super::FETCH1_FETCH_RESPONSE_SNAPPY_K0822.to_owned(), false)
+            bench_decode_new_fetch_response(b,
+                                            super::FETCH1_FETCH_RESPONSE_SNAPPY_K0822.to_owned(),
+                                            false)
         }
 
         #[cfg(feature = "gzip")]
         #[bench]
         fn bench_decode_new_fetch_response_gzip_k0821(b: &mut Bencher) {
-            bench_decode_new_fetch_response(b, super::FETCH1_FETCH_RESPONSE_GZIP_K0821.to_owned(), false)
+            bench_decode_new_fetch_response(b,
+                                            super::FETCH1_FETCH_RESPONSE_GZIP_K0821.to_owned(),
+                                            false)
         }
 
         #[bench]
         fn bench_decode_new_fetch_response_nocompression_k0821_validate_crc(b: &mut Bencher) {
-            bench_decode_new_fetch_response(b, super::FETCH1_FETCH_RESPONSE_NOCOMPRESSION_K0821.to_owned(), true)
+            bench_decode_new_fetch_response(
+                b,
+                super::FETCH1_FETCH_RESPONSE_NOCOMPRESSION_K0821.to_owned(),
+                true)
         }
 
         #[cfg(feature = "snappy")]
         #[bench]
         fn bench_decode_new_fetch_response_snappy_k0821_validate_crc(b: &mut Bencher) {
-            bench_decode_new_fetch_response(b, super::FETCH1_FETCH_RESPONSE_SNAPPY_K0821.to_owned(), true)
+            bench_decode_new_fetch_response(b,
+                                            super::FETCH1_FETCH_RESPONSE_SNAPPY_K0821.to_owned(),
+                                            true)
         }
 
         #[cfg(feature = "snappy")]
         #[bench]
         fn bench_decode_new_fetch_response_snappy_k0822_validate_crc(b: &mut Bencher) {
-            bench_decode_new_fetch_response(b, super::FETCH1_FETCH_RESPONSE_SNAPPY_K0822.to_owned(), true)
+            bench_decode_new_fetch_response(b,
+                                            super::FETCH1_FETCH_RESPONSE_SNAPPY_K0822.to_owned(),
+                                            true)
         }
 
         #[cfg(feature = "gzip")]
         #[bench]
         fn bench_decode_new_fetch_response_gzip_k0821_validate_crc(b: &mut Bencher) {
-            bench_decode_new_fetch_response(b, super::FETCH1_FETCH_RESPONSE_GZIP_K0821.to_owned(), true)
+            bench_decode_new_fetch_response(b,
+                                            super::FETCH1_FETCH_RESPONSE_GZIP_K0821.to_owned(),
+                                            true)
         }
     }
 }
diff --git a/src/protocol/metadata.rs b/src/protocol/metadata.rs
index 180f80831..2c9bfe259 100644
--- a/src/protocol/metadata.rs
+++ b/src/protocol/metadata.rs
@@ -9,26 +9,21 @@ use super::{API_KEY_METADATA, API_VERSION};
 #[derive(Debug)]
 pub struct MetadataRequest<'a, T: 'a> {
     pub header: HeaderRequest<'a>,
-    pub topics: &'a [T]
+    pub topics: &'a [T],
 }
 
 impl<'a, T: AsRef<str>> MetadataRequest<'a, T> {
-    pub fn new(correlation_id: i32, client_id: &'a str, topics: &'a [T])
-               -> MetadataRequest<'a, T>
-    {
+    pub fn new(correlation_id: i32, client_id: &'a str, topics: &'a [T]) -> MetadataRequest<'a, T> {
         MetadataRequest {
-            header: HeaderRequest::new(
-                API_KEY_METADATA, API_VERSION, correlation_id, client_id),
-            topics: topics
+            header: HeaderRequest::new(API_KEY_METADATA, API_VERSION, correlation_id, client_id),
+            topics: topics,
         }
     }
 }
 
 impl<'a, T: AsRef<str> + 'a> ToByte for MetadataRequest<'a, T> {
     fn encode<W: Write>(&self, buffer: &mut W) -> Result<()> {
-        try_multi!(
-            self.header.encode(buffer),
-            AsStrings(self.topics).encode(buffer))
+        try_multi!(self.header.encode(buffer), AsStrings(self.topics).encode(buffer))
     }
 }
 
@@ -38,21 +33,21 @@ impl<'a, T: AsRef<str> + 'a> ToByte for MetadataRequest<'a, T> {
 pub struct MetadataResponse {
     pub header: HeaderResponse,
     pub brokers: Vec<BrokerMetadata>,
-    pub topics: Vec<TopicMetadata>
+    pub topics: Vec<TopicMetadata>,
 }
 
 #[derive(Default, Debug)]
 pub struct BrokerMetadata {
     pub node_id: i32,
     pub host: String,
-    pub port: i32
+    pub port: i32,
 }
 
 #[derive(Default, Debug)]
 pub struct TopicMetadata {
     pub error: i16,
     pub topic: String,
-    pub partitions: Vec<PartitionMetadata>
+    pub partitions: Vec<PartitionMetadata>,
 }
 
 #[derive(Default, Debug)]
@@ -61,7 +56,7 @@ pub struct PartitionMetadata {
     pub id: i32,
     pub leader: i32,
     pub replicas: Vec<i32>,
-    pub isr: Vec<i32>
+    pub isr: Vec<i32>,
 }
 
 impl FromByte for MetadataResponse {
@@ -69,11 +64,9 @@ impl FromByte for MetadataResponse {
 
     #[allow(unused_must_use)]
     fn decode<T: Read>(&mut self, buffer: &mut T) -> Result<()> {
-        try_multi!(
-            self.header.decode(buffer),
-            self.brokers.decode(buffer),
-            self.topics.decode(buffer)
-        )
+        try_multi!(self.header.decode(buffer),
+                   self.brokers.decode(buffer),
+                   self.topics.decode(buffer))
     }
 }
 
@@ -82,11 +75,7 @@ impl FromByte for BrokerMetadata {
 
     #[allow(unused_must_use)]
     fn decode<T: Read>(&mut self, buffer: &mut T) -> Result<()> {
-        try_multi!(
-            self.node_id.decode(buffer),
-            self.host.decode(buffer),
-            self.port.decode(buffer)
-        )
+        try_multi!(self.node_id.decode(buffer), self.host.decode(buffer), self.port.decode(buffer))
     }
 }
 
@@ -95,11 +84,9 @@ impl FromByte for TopicMetadata {
 
     #[allow(unused_must_use)]
     fn decode<T: Read>(&mut self, buffer: &mut T) -> Result<()> {
-        try_multi!(
-            self.error.decode(buffer),
-            self.topic.decode(buffer),
-            self.partitions.decode(buffer)
-        )
+        try_multi!(self.error.decode(buffer),
+                   self.topic.decode(buffer),
+                   self.partitions.decode(buffer))
     }
 }
 
@@ -108,12 +95,10 @@ impl FromByte for PartitionMetadata {
 
     #[allow(unused_must_use)]
     fn decode<T: Read>(&mut self, buffer: &mut T) -> Result<()> {
-        try_multi!(
-            self.error.decode(buffer),
-            self.id.decode(buffer),
-            self.leader.decode(buffer),
-            self.replicas.decode(buffer),
-            self.isr.decode(buffer)
-        )
+        try_multi!(self.error.decode(buffer),
+                   self.id.decode(buffer),
+                   self.leader.decode(buffer),
+                   self.replicas.decode(buffer),
+                   self.isr.decode(buffer))
     }
 }
diff --git a/src/protocol/mod.rs b/src/protocol/mod.rs
index fe789779a..fa8dd0004 100644
--- a/src/protocol/mod.rs
+++ b/src/protocol/mod.rs
@@ -28,19 +28,19 @@ pub use self::fetch::FetchRequest;
 pub use self::produce::{ProduceRequest, ProduceResponse};
 pub use self::offset::{OffsetRequest, OffsetResponse};
 pub use self::metadata::{MetadataRequest, MetadataResponse};
-pub use self::consumer::{GroupCoordinatorRequest, GroupCoordinatorResponse,
-                         OffsetFetchVersion, OffsetFetchRequest, OffsetFetchResponse,
-                         OffsetCommitVersion, OffsetCommitRequest, OffsetCommitResponse};
+pub use self::consumer::{GroupCoordinatorRequest, GroupCoordinatorResponse, OffsetFetchVersion,
+                         OffsetFetchRequest, OffsetFetchResponse, OffsetCommitVersion,
+                         OffsetCommitRequest, OffsetCommitResponse};
 
 // --------------------------------------------------------------------
 
-const API_KEY_PRODUCE: i16   = 0;
-const API_KEY_FETCH: i16     = 1;
-const API_KEY_OFFSET: i16    = 2;
-const API_KEY_METADATA: i16  = 3;
+const API_KEY_PRODUCE: i16 = 0;
+const API_KEY_FETCH: i16 = 1;
+const API_KEY_OFFSET: i16 = 2;
+const API_KEY_METADATA: i16 = 3;
 // 4-7 reserved for non-public kafka api services
-const API_KEY_OFFSET_COMMIT: i16     = 8;
-const API_KEY_OFFSET_FETCH: i16      = 9;
+const API_KEY_OFFSET_COMMIT: i16 = 8;
+const API_KEY_OFFSET_FETCH: i16 = 9;
 const API_KEY_GROUP_COORDINATOR: i16 = 10;
 
 // the default version of Kafka API we are requesting
@@ -79,7 +79,11 @@ fn test_kafka_code_from_protocol() {
         }
     };
 
-    assert!(if let None = KafkaCode::from_protocol(0) { true } else { false });
+    assert!(if let None = KafkaCode::from_protocol(0) {
+        true
+    } else {
+        false
+    });
     assert_kafka_code!(KafkaCode::OffsetOutOfRange, KafkaCode::OffsetOutOfRange as i16);
     assert_kafka_code!(KafkaCode::IllegalGeneration, KafkaCode::IllegalGeneration as i16);
     assert_kafka_code!(KafkaCode::UnsupportedVersion, KafkaCode::UnsupportedVersion as i16);
@@ -109,9 +113,11 @@ pub struct HeaderRequest<'a> {
 }
 
 impl<'a> HeaderRequest<'a> {
-    fn new(api_key: i16, api_version: i16, correlation_id: i32, client_id: &'a str)
-           -> HeaderRequest
-    {
+    fn new(api_key: i16,
+           api_version: i16,
+           correlation_id: i32,
+           client_id: &'a str)
+           -> HeaderRequest {
         HeaderRequest {
             api_key: api_key,
             api_version: api_version,
@@ -123,11 +129,10 @@ impl<'a> HeaderRequest<'a> {
 
 impl<'a> ToByte for HeaderRequest<'a> {
     fn encode<W: Write>(&self, buffer: &mut W) -> Result<()> {
-        try_multi!(
-            self.api_key.encode(buffer),
-            self.api_version.encode(buffer),
-            self.correlation_id.encode(buffer),
-            self.client_id.encode(buffer))
+        try_multi!(self.api_key.encode(buffer),
+                   self.api_version.encode(buffer),
+                   self.correlation_id.encode(buffer),
+                   self.client_id.encode(buffer))
     }
 }
 
@@ -135,7 +140,7 @@ impl<'a> ToByte for HeaderRequest<'a> {
 
 #[derive(Default, Debug, Clone)]
 pub struct HeaderResponse {
-    pub correlation: i32
+    pub correlation: i32,
 }
 
 impl FromByte for HeaderResponse {
@@ -180,10 +185,8 @@ fn test_to_millis_i32() {
     fn assert_valid(d: Duration, expected_millis: i32) {
         let r = to_millis_i32(d);
         match r {
-            Ok(m) =>
-                assert_eq!(expected_millis, m),
-            Err(e) =>
-                panic!("Expected Ok({}) but got Err({:?})", expected_millis, e),
+            Ok(m) => assert_eq!(expected_millis, m),
+            Err(e) => panic!("Expected Ok({}) but got Err({:?})", expected_millis, e),
         }
     }
     assert_valid(Duration::from_millis(1_234), 1_234);
diff --git a/src/protocol/offset.rs b/src/protocol/offset.rs
index e7467ffcd..b34dc1b46 100644
--- a/src/protocol/offset.rs
+++ b/src/protocol/offset.rs
@@ -11,13 +11,13 @@ use super::{API_KEY_OFFSET, API_VERSION};
 pub struct OffsetRequest<'a> {
     pub header: HeaderRequest<'a>,
     pub replica: i32,
-    pub topic_partitions: Vec<TopicPartitionOffsetRequest<'a>>
+    pub topic_partitions: Vec<TopicPartitionOffsetRequest<'a>>,
 }
 
 #[derive(Default, Debug)]
 pub struct TopicPartitionOffsetRequest<'a> {
     pub topic: &'a str,
-    pub partitions: Vec<PartitionOffsetRequest>
+    pub partitions: Vec<PartitionOffsetRequest>,
 }
 
 #[derive(Default, Debug)]
@@ -30,10 +30,9 @@ pub struct PartitionOffsetRequest {
 impl<'a> OffsetRequest<'a> {
     pub fn new(correlation_id: i32, client_id: &'a str) -> OffsetRequest<'a> {
         OffsetRequest {
-            header: HeaderRequest::new(
-                API_KEY_OFFSET, API_VERSION, correlation_id, client_id),
+            header: HeaderRequest::new(API_KEY_OFFSET, API_VERSION, correlation_id, client_id),
             replica: -1,
-            topic_partitions: vec!()
+            topic_partitions: vec![],
         }
     }
 
@@ -54,7 +53,7 @@ impl<'a> TopicPartitionOffsetRequest<'a> {
     pub fn new(topic: &'a str) -> TopicPartitionOffsetRequest<'a> {
         TopicPartitionOffsetRequest {
             topic: topic,
-            partitions: vec!()
+            partitions: vec![],
         }
     }
 
@@ -65,7 +64,7 @@ impl<'a> TopicPartitionOffsetRequest<'a> {
 
 impl PartitionOffsetRequest {
     pub fn new(partition: i32, time: i64) -> PartitionOffsetRequest {
-        PartitionOffsetRequest{
+        PartitionOffsetRequest {
             partition: partition,
             max_offsets: 1,
             time: time,
@@ -74,31 +73,24 @@ impl PartitionOffsetRequest {
 }
 
 impl<'a> ToByte for OffsetRequest<'a> {
-    fn encode<T:Write>(&self, buffer: &mut T) -> Result<()> {
-        try_multi!(
-            self.header.encode(buffer),
-            self.replica.encode(buffer),
-            self.topic_partitions.encode(buffer)
-        )
+    fn encode<T: Write>(&self, buffer: &mut T) -> Result<()> {
+        try_multi!(self.header.encode(buffer),
+                   self.replica.encode(buffer),
+                   self.topic_partitions.encode(buffer))
     }
 }
 
 impl<'a> ToByte for TopicPartitionOffsetRequest<'a> {
     fn encode<T: Write>(&self, buffer: &mut T) -> Result<()> {
-        try_multi!(
-            self.topic.encode(buffer),
-            self.partitions.encode(buffer)
-        )
+        try_multi!(self.topic.encode(buffer), self.partitions.encode(buffer))
     }
 }
 
 impl ToByte for PartitionOffsetRequest {
     fn encode<T: Write>(&self, buffer: &mut T) -> Result<()> {
-        try_multi!(
-            self.partition.encode(buffer),
-            self.time.encode(buffer),
-            self.max_offsets.encode(buffer)
-        )
+        try_multi!(self.partition.encode(buffer),
+                   self.time.encode(buffer),
+                   self.max_offsets.encode(buffer))
     }
 }
 
@@ -120,7 +112,7 @@ pub struct TopicPartitionOffsetResponse {
 pub struct PartitionOffsetResponse {
     pub partition: i32,
     pub error: i16,
-    pub offset: Vec<i64>
+    pub offset: Vec<i64>,
 }
 
 impl PartitionOffsetResponse {
@@ -128,12 +120,14 @@ impl PartitionOffsetResponse {
         PartitionOffset {
             partition: self.partition,
             offset: match Error::from_protocol(self.error) {
-                None => Ok(match self.offset.first() {
-                    Some(offs) => *offs,
-                    None => -1,
-                }),
+                None => {
+                    Ok(match self.offset.first() {
+                        Some(offs) => *offs,
+                        None => -1,
+                    })
+                }
                 Some(e) => Err(e),
-            }
+            },
         }
     }
 }
@@ -143,10 +137,7 @@ impl FromByte for OffsetResponse {
 
     #[allow(unused_must_use)]
     fn decode<T: Read>(&mut self, buffer: &mut T) -> Result<()> {
-        try_multi!(
-            self.header.decode(buffer),
-            self.topic_partitions.decode(buffer)
-        )
+        try_multi!(self.header.decode(buffer), self.topic_partitions.decode(buffer))
     }
 }
 
@@ -155,10 +146,7 @@ impl FromByte for TopicPartitionOffsetResponse {
 
     #[allow(unused_must_use)]
     fn decode<T: Read>(&mut self, buffer: &mut T) -> Result<()> {
-        try_multi!(
-            self.topic.decode(buffer),
-            self.partitions.decode(buffer)
-        )
+        try_multi!(self.topic.decode(buffer), self.partitions.decode(buffer))
     }
 }
 
@@ -167,10 +155,8 @@ impl FromByte for PartitionOffsetResponse {
 
     #[allow(unused_must_use)]
     fn decode<T: Read>(&mut self, buffer: &mut T) -> Result<()> {
-        try_multi!(
-            self.partition.decode(buffer),
-            self.error.decode(buffer),
-            self.offset.decode(buffer)
-        )
+        try_multi!(self.partition.decode(buffer),
+                   self.error.decode(buffer),
+                   self.offset.decode(buffer))
     }
 }
diff --git a/src/protocol/produce.rs b/src/protocol/produce.rs
index c19c4e622..b774b7cae 100644
--- a/src/protocol/produce.rs
+++ b/src/protocol/produce.rs
@@ -46,24 +46,26 @@ pub struct MessageProduceRequest<'a> {
 }
 
 impl<'a, 'b> ProduceRequest<'a, 'b> {
-    pub fn new(required_acks: i16, timeout: i32,
-               correlation_id: i32, client_id: &'a str,
+    pub fn new(required_acks: i16,
+               timeout: i32,
+               correlation_id: i32,
+               client_id: &'a str,
                compression: Compression)
-               -> ProduceRequest<'a, 'b>
-    {
+               -> ProduceRequest<'a, 'b> {
         ProduceRequest {
-            header: HeaderRequest::new(
-                API_KEY_PRODUCE, API_VERSION, correlation_id, client_id),
+            header: HeaderRequest::new(API_KEY_PRODUCE, API_VERSION, correlation_id, client_id),
             required_acks: required_acks,
             timeout: timeout,
-            topic_partitions: vec!(),
-            compression: compression
+            topic_partitions: vec![],
+            compression: compression,
         }
     }
 
-    pub fn add(&mut self, topic: &'b str, partition: i32,
-               key: Option<&'b [u8]>, value: Option<&'b [u8]>)
-    {
+    pub fn add(&mut self,
+               topic: &'b str,
+               partition: i32,
+               key: Option<&'b [u8]>,
+               value: Option<&'b [u8]>) {
         for tp in &mut self.topic_partitions {
             if tp.topic == topic {
                 tp.add(partition, key, value);
@@ -80,12 +82,12 @@ impl<'a> TopicPartitionProduceRequest<'a> {
     pub fn new(topic: &'a str, compression: Compression) -> TopicPartitionProduceRequest<'a> {
         TopicPartitionProduceRequest {
             topic: topic,
-            partitions: vec!(),
-            compression: compression
+            partitions: vec![],
+            compression: compression,
         }
     }
 
-    pub fn add(&mut self, partition: i32, key: Option<&'a [u8]>, value: Option<&'a[u8]>) {
+    pub fn add(&mut self, partition: i32, key: Option<&'a [u8]>, value: Option<&'a [u8]>) {
         for pp in &mut self.partitions {
             if pp.partition == partition {
                 pp.add(key, value);
@@ -97,8 +99,14 @@ impl<'a> TopicPartitionProduceRequest<'a> {
 }
 
 impl<'a> PartitionProduceRequest<'a> {
-    pub fn new<'b>(partition: i32, key: Option<&'b [u8]>, value: Option<&'b [u8]>) -> PartitionProduceRequest<'b> {
-        let mut r = PartitionProduceRequest { partition: partition, messages: Vec::new() };
+    pub fn new<'b>(partition: i32,
+                   key: Option<&'b [u8]>,
+                   value: Option<&'b [u8]>)
+                   -> PartitionProduceRequest<'b> {
+        let mut r = PartitionProduceRequest {
+            partition: partition,
+            messages: Vec::new(),
+        };
         r.add(key, value);
         r
     }
@@ -134,8 +142,7 @@ impl<'a> PartitionProduceRequest<'a> {
     //
     // MessetSet => [Offset MessageSize Message]
     // MessageSets are not preceded by an int32 like other array elements in the protocol.
-    fn _encode<W: Write>(&self, out: &mut W, compression: Compression)
-                         -> Result<()> {
+    fn _encode<W: Write>(&self, out: &mut W, compression: Compression) -> Result<()> {
         try!(self.partition.encode(out));
 
         // ~ render the whole MessageSet first to a temporary buffer
@@ -165,9 +172,7 @@ impl<'a> PartitionProduceRequest<'a> {
 // ~ A helper method to render `cdata` into `out` as a compressed message.
 // ~ `out` is first cleared and then populated with the rendered message.
 #[cfg(any(feature = "snappy", feature = "gzip"))]
-fn render_compressed(out: &mut Vec<u8>, cdata: &[u8], compression: Compression)
-                     -> Result<()>
-{
+fn render_compressed(out: &mut Vec<u8>, cdata: &[u8], compression: Compression) -> Result<()> {
     out.clear();
     let cmsg = MessageProduceRequest::new(None, Some(cdata));
     cmsg._encode_to_buf(out, MESSAGE_MAGIC_BYTE, compression as i8)
@@ -175,7 +180,10 @@ fn render_compressed(out: &mut Vec<u8>, cdata: &[u8], compression: Compression)
 
 impl<'a> MessageProduceRequest<'a> {
     fn new<'b>(key: Option<&'b [u8]>, value: Option<&'b [u8]>) -> MessageProduceRequest<'b> {
-        MessageProduceRequest { key: key, value: value }
+        MessageProduceRequest {
+            key: key,
+            value: value,
+        }
     }
 
     // render a single message as: Offset MessageSize Message
@@ -208,11 +216,11 @@ impl<'a> MessageProduceRequest<'a> {
 
         // compute the crc and store it back in the reserved space
         crc = to_crc(&buffer[(crc_pos + 4)..]) as i32;
-        try!(crc.encode(&mut &mut buffer[crc_pos .. crc_pos + 4]));
+        try!(crc.encode(&mut &mut buffer[crc_pos..crc_pos + 4]));
 
         // compute the size and store it back in the reserved space
         size = (buffer.len() - crc_pos) as i32;
-        try!(size.encode(&mut &mut buffer[size_pos .. size_pos + 4]));
+        try!(size.encode(&mut &mut buffer[size_pos..size_pos + 4]));
 
         Ok(())
     }
@@ -232,7 +240,7 @@ impl<'a> ToByte for Option<&'a [u8]> {
 #[derive(Default, Debug, Clone)]
 pub struct ProduceResponse {
     pub header: HeaderResponse,
-    pub topic_partitions: Vec<TopicPartitionProduceResponse>
+    pub topic_partitions: Vec<TopicPartitionProduceResponse>,
 }
 
 #[derive(Default, Debug, Clone)]
@@ -245,7 +253,7 @@ pub struct TopicPartitionProduceResponse {
 pub struct PartitionProduceResponse {
     pub partition: i32,
     pub error: i16,
-    pub offset: i64
+    pub offset: i64,
 }
 
 impl ProduceResponse {
@@ -274,7 +282,7 @@ impl PartitionProduceResponse {
             offset: match Error::from_protocol(self.error) {
                 None => Ok(self.offset),
                 Some(e) => Err(e),
-            }
+            },
         }
     }
 }
@@ -284,10 +292,7 @@ impl FromByte for ProduceResponse {
 
     #[allow(unused_must_use)]
     fn decode<T: Read>(&mut self, buffer: &mut T) -> Result<()> {
-        try_multi!(
-            self.header.decode(buffer),
-            self.topic_partitions.decode(buffer)
-        )
+        try_multi!(self.header.decode(buffer), self.topic_partitions.decode(buffer))
     }
 }
 
@@ -296,10 +301,7 @@ impl FromByte for TopicPartitionProduceResponse {
 
     #[allow(unused_must_use)]
     fn decode<T: Read>(&mut self, buffer: &mut T) -> Result<()> {
-        try_multi!(
-            self.topic.decode(buffer),
-            self.partitions.decode(buffer)
-        )
+        try_multi!(self.topic.decode(buffer), self.partitions.decode(buffer))
     }
 }
 
@@ -308,11 +310,8 @@ impl FromByte for PartitionProduceResponse {
 
     #[allow(unused_must_use)]
     fn decode<T: Read>(&mut self, buffer: &mut T) -> Result<()> {
-        try_multi!(
-            self.partition.decode(buffer),
-            self.error.decode(buffer),
-            self.offset.decode(buffer)
-        )
+        try_multi!(self.partition.decode(buffer),
+                   self.error.decode(buffer),
+                   self.offset.decode(buffer))
     }
 }
-
diff --git a/src/protocol/zreader.rs b/src/protocol/zreader.rs
index 9829cffaa..1fee40b5e 100644
--- a/src/protocol/zreader.rs
+++ b/src/protocol/zreader.rs
@@ -140,7 +140,7 @@ fn test_read_i8() {
 fn test_read_i16() {
     let data = &[1, 2, 16, 1];
     let mut r = ZReader::new(data);
-    assert_eq!(258 , r.read_i16().unwrap());
+    assert_eq!(258, r.read_i16().unwrap());
     assert_eq!(4097, r.read_i16().unwrap());
     assert!(r.read_i16().is_err());
 
@@ -184,10 +184,8 @@ fn test_read_i64() {
 
 #[test]
 fn test_read_str() {
-    let data = &[0u8, 5, b'h', b'e', b'l', b'l', b'o',
-                 0u8, 7, b',', b' ', b'w', b'o', b'r', b'l', b'd',
-                 255, // a "null" string
-                 28]; // some byte
+    let data = &[0u8, 5, b'h', b'e', b'l', b'l', b'o', 0u8, 7, b',', b' ', b'w', b'o', b'r', b'l',
+                 b'd', 255 /* a "null" string */, 28]; // some byte
 
     let mut r = ZReader::new(data);
     assert_eq!("hello", r.read_str().unwrap());
@@ -202,8 +200,7 @@ fn test_read_str() {
 /// returned slice/string.
 #[test]
 fn test_mutability_lifetimes() {
-    let data = &[0, 2, b'h', b'i',
-                 0, 2, b'h', b'o'];
+    let data = &[0, 2, b'h', b'i', 0, 2, b'h', b'o'];
 
     let mut r = ZReader::new(data);
     let x = r.read_str().unwrap();


Test patch: