cargo fmt

benches/bank.rs

@@ -37,8 +37,7 @@ fn bench_process_transaction(bencher: &mut Bencher) {
 
             // Finally, return the transaction to the benchmark.
             tx
-        })
+        }).collect();
-        .collect();
 
     bencher.iter(|| {
         // Since benchmarker runs this multiple times, we need to clear the signatures.

benches/banking_stage.rs

@@ -116,8 +116,7 @@ fn bench_banking_stage_multi_accounts(bencher: &mut Bencher) {
                 i as i64,
                 mint.last_id(),
             )
-        })
+        }).collect();
-        .collect();
 
     let (verified_sender, verified_receiver) = channel();
     let (signal_sender, signal_receiver) = channel();
@@ -131,8 +130,7 @@ fn bench_banking_stage_multi_accounts(bencher: &mut Bencher) {
                 mint_total / num_src_accounts as i64,
                 mint.last_id(),
             )
-        })
+        }).collect();
-        .collect();
 
     bencher.iter(move || {
         let bank = Arc::new(Bank::new(&mint));
@@ -143,8 +141,7 @@ fn bench_banking_stage_multi_accounts(bencher: &mut Bencher) {
                 .map(|x| {
                     let len = (*x).read().unwrap().packets.len();
                     (x, iter::repeat(1).take(len).collect())
-                })
+                }).collect();
-                .collect();
 
         verified_sender.send(verified_setup).unwrap();
         BankingStage::process_packets(&bank, &verified_receiver, &signal_sender, &packet_recycler)
@@ -157,8 +154,7 @@ fn bench_banking_stage_multi_accounts(bencher: &mut Bencher) {
             .map(|x| {
                 let len = (*x).read().unwrap().packets.len();
                 (x, iter::repeat(1).take(len).collect())
-            })
+            }).collect();
-            .collect();
 
         verified_sender.send(verified).unwrap();
         BankingStage::process_packets(&bank, &verified_receiver, &signal_sender, &packet_recycler)
@@ -187,8 +183,7 @@ fn bench_banking_stage_single_from(bencher: &mut Bencher) {
                 i as i64,
                 mint.last_id(),
             )
-        })
+        }).collect();
-        .collect();
 
     let (verified_sender, verified_receiver) = channel();
     let (signal_sender, signal_receiver) = channel();
@@ -201,8 +196,7 @@ fn bench_banking_stage_single_from(bencher: &mut Bencher) {
             .map(|x| {
                 let len = (*x).read().unwrap().packets.len();
                 (x, iter::repeat(1).take(len).collect())
-            })
+            }).collect();
-            .collect();
         verified_sender.send(verified).unwrap();
         BankingStage::process_packets(&bank, &verified_receiver, &signal_sender, &packet_recycler)
             .unwrap();

src/bank.rs

@@ -449,8 +449,7 @@ impl Bank {
             .map(|(acc, tx)| match acc {
                 Err(e) => Err(e.clone()),
                 Ok(ref mut accounts) => Self::execute_transaction(tx, accounts),
-            })
+            }).collect();
-            .collect();
         let execution_elapsed = now.elapsed();
         let now = Instant::now();
         Self::store_accounts(&res, &loaded_accounts, &mut accounts);
@@ -1010,8 +1009,7 @@ mod tests {
                 let last_id = hash(&serialize(&i).unwrap()); // Unique hash
                 bank.register_entry_id(&last_id);
                 last_id
-            })
+            }).collect();
-            .collect();
         assert_eq!(bank.count_valid_ids(&[]).len(), 0);
         assert_eq!(bank.count_valid_ids(&[mint.last_id()]).len(), 0);
         for (i, id) in bank.count_valid_ids(&ids).iter().enumerate() {

src/banking_stage.rs

@@ -52,8 +52,7 @@ impl BankingStage {
                         _ => error!("{:?}", e),
                     }
                 }
-            })
+            }).unwrap();
-            .unwrap();
         (BankingStage { thread_hdl }, signal_receiver)
     }
 
@@ -66,8 +65,7 @@ impl BankingStage {
                 deserialize(&x.data[0..x.meta.size])
                     .map(|req| (req, x.meta.addr()))
                     .ok()
-            })
+            }).collect()
-            .collect()
     }
 
     /// Process the incoming packets and send output `Signal` messages to `signal_sender`.
@@ -105,8 +103,7 @@ impl BankingStage {
                     } else {
                         None
                     },
-                })
+                }).collect();
-                .collect();
 
             debug!("process_transactions");
             let results = bank.process_transactions(transactions);

src/bin/bench-streamer.rs

@@ -68,8 +68,7 @@ fn main() -> Result<()> {
                 .value_name("NUM")
                 .takes_value(true)
                 .help("Use NUM receive sockets"),
-        )
+        ).get_matches();
-        .get_matches();
 
     if let Some(n) = matches.value_of("num-recv-sockets") {
         num_sockets = max(num_sockets, n.to_string().parse().expect("integer"));

src/bin/bench-tps.rs

@@ -134,8 +134,7 @@ fn send_barrier_transaction(barrier_client: &mut ThinClient, last_id: &mut Hash,
                     .add_tag(
                         "op",
                         influxdb::Value::String("send_barrier_transaction".to_string()),
-                    )
+                    ).add_field("poll_count", influxdb::Value::Integer(poll_count))
-                    .add_field("poll_count", influxdb::Value::Integer(poll_count))
                     .add_field("duration", influxdb::Value::Integer(duration_ms as i64))
                     .to_owned(),
             );
@@ -146,8 +145,7 @@ fn send_barrier_transaction(barrier_client: &mut ThinClient, last_id: &mut Hash,
                     &id.pubkey(),
                     &Duration::from_millis(100),
                     &Duration::from_secs(10),
-                )
+                ).expect("Failed to get balance");
-                .expect("Failed to get balance");
             if balance != 1 {
                 panic!("Expected an account balance of 1 (balance: {}", balance);
             }
@@ -194,8 +192,7 @@ fn generate_txs(
             } else {
                 Transaction::new(keypair, id.pubkey(), 1, *last_id)
             }
-        })
+        }).collect();
-        .collect();
 
     let duration = signing_start.elapsed();
     let ns = duration.as_secs() * 1_000_000_000 + u64::from(duration.subsec_nanos());
@@ -213,8 +210,7 @@ fn generate_txs(
             .add_field(
                 "duration",
                 influxdb::Value::Integer(duration_as_ms(&duration) as i64),
-            )
+            ).to_owned(),
-            .to_owned(),
     );
 
     let sz = transactions.len() / threads;
@@ -266,8 +262,7 @@ fn do_tx_transfers(
                     .add_field(
                         "duration",
                         influxdb::Value::Integer(duration_as_ms(&transfer_start.elapsed()) as i64),
-                    )
+                    ).add_field("count", influxdb::Value::Integer(tx_len as i64))
-                    .add_field("count", influxdb::Value::Integer(tx_len as i64))
                     .to_owned(),
             );
         }
@@ -591,10 +586,8 @@ fn main() {
                 .name("solana-client-sample".to_string())
                 .spawn(move || {
                     sample_tx_count(&exit_signal, &maxes, first_tx_count, &v, sample_period);
-                })
+                }).unwrap()
-                .unwrap()
+        }).collect();
-        })
-        .collect();
 
     let shared_txs: Arc<RwLock<VecDeque<Vec<Transaction>>>> =
         Arc::new(RwLock::new(VecDeque::new()));
@@ -619,10 +612,8 @@ fn main() {
                         &shared_tx_active_thread_count,
                         &total_tx_sent_count,
                     );
-                })
+                }).unwrap()
-                .unwrap()
+        }).collect();
-        })
-        .collect();
 
     // generate and send transactions for the specified duration
     let start = Instant::now();

src/bin/drone.rs

@@ -48,8 +48,7 @@ fn main() -> Result<(), Box<error::Error>> {
                 .takes_value(true)
                 .required(true)
                 .help("rendezvous with the network at this gossip entry point"),
-        )
+        ).arg(
-        .arg(
             Arg::with_name("keypair")
                 .short("k")
                 .long("keypair")
@@ -57,22 +56,19 @@ fn main() -> Result<(), Box<error::Error>> {
                 .takes_value(true)
                 .required(true)
                 .help("File to read the client's keypair from"),
-        )
+        ).arg(
-        .arg(
             Arg::with_name("slice")
                 .long("slice")
                 .value_name("SECONDS")
                 .takes_value(true)
                 .help("Time slice over which to limit requests to drone"),
-        )
+        ).arg(
-        .arg(
             Arg::with_name("cap")
                 .long("cap")
                 .value_name("NUMBER")
                 .takes_value(true)
                 .help("Request limit for time slice"),
-        )
+        ).get_matches();
-        .get_matches();
 
     let network = matches
         .value_of("network")
@@ -159,8 +155,7 @@ fn main() -> Result<(), Box<error::Error>> {
                         io::ErrorKind::Other,
                         format!("Drone response: {:?}", err),
                     ))
-                }))
+                })).then(|_| Ok(()));
-                .then(|_| Ok(()));
             tokio::spawn(server)
         });
     tokio::run(done);

src/bin/fullnode-config.rs

@@ -21,31 +21,27 @@ fn main() {
                 .long("local")
                 .takes_value(false)
                 .help("detect network address from local machine configuration"),
-        )
+        ).arg(
-        .arg(
             Arg::with_name("keypair")
                 .short("k")
                 .long("keypair")
                 .value_name("PATH")
                 .takes_value(true)
                 .help("/path/to/id.json"),
-        )
+        ).arg(
-        .arg(
             Arg::with_name("public")
                 .short("p")
                 .long("public")
                 .takes_value(false)
                 .help("detect public network address using public servers"),
-        )
+        ).arg(
-        .arg(
             Arg::with_name("bind")
                 .short("b")
                 .long("bind")
                 .value_name("PORT")
                 .takes_value(true)
                 .help("bind to port or address"),
-        )
+        ).get_matches();
-        .get_matches();
 
     let bind_addr: SocketAddr = {
         let mut bind_addr = parse_port_or_addr(matches.value_of("bind"), FULLNODE_PORT_RANGE.0);

src/bin/fullnode.rs

@@ -36,16 +36,14 @@ fn main() -> () {
                 .value_name("FILE")
                 .takes_value(true)
                 .help("run with the identity found in FILE"),
-        )
+        ).arg(
-        .arg(
             Arg::with_name("network")
                 .short("n")
                 .long("network")
                 .value_name("HOST:PORT")
                 .takes_value(true)
                 .help("connect/rendezvous with the network at this gossip entry point"),
-        )
+        ).arg(
-        .arg(
             Arg::with_name("ledger")
                 .short("l")
                 .long("ledger")
@@ -53,8 +51,7 @@ fn main() -> () {
                 .takes_value(true)
                 .required(true)
                 .help("use DIR as persistent ledger location"),
-        )
+        ).get_matches();
-        .get_matches();
 
     let (keypair, ncp) = if let Some(i) = matches.value_of("identity") {
         let path = i.to_string();

src/bin/genesis.rs

@@ -25,8 +25,7 @@ fn main() -> Result<(), Box<error::Error>> {
                 .takes_value(true)
                 .required(true)
                 .help("Number of tokens with which to initialize mint"),
-        )
+        ).arg(
-        .arg(
             Arg::with_name("ledger")
                 .short("l")
                 .long("ledger")
@@ -34,8 +33,7 @@ fn main() -> Result<(), Box<error::Error>> {
                 .takes_value(true)
                 .required(true)
                 .help("use DIR as persistent ledger location"),
-        )
+        ).get_matches();
-        .get_matches();
 
     let tokens = value_t_or_exit!(matches, "tokens", i64);
     let ledger_path = matches.value_of("ledger").unwrap();

src/bin/keygen.rs

@@ -22,8 +22,7 @@ fn main() -> Result<(), Box<error::Error>> {
                 .value_name("PATH")
                 .takes_value(true)
                 .help("path to generated file"),
-        )
+        ).get_matches();
-        .get_matches();
 
     let rnd = SystemRandom::new();
     let pkcs8_bytes = Ed25519KeyPair::generate_pkcs8(&rnd)?;

src/bin/wallet.rs

@@ -69,23 +69,20 @@ fn main() -> Result<(), Box<error::Error>> {
                 .value_name("PATH")
                 .takes_value(true)
                 .help("/path/to/leader.json"),
-        )
+        ).arg(
-        .arg(
             Arg::with_name("keypair")
                 .short("k")
                 .long("keypair")
                 .value_name("PATH")
                 .takes_value(true)
                 .help("/path/to/id.json"),
-        )
+        ).arg(
-        .arg(
             Arg::with_name("timeout")
                 .long("timeout")
                 .value_name("SECONDS")
                 .takes_value(true)
                 .help("Max SECONDS to wait to get necessary gossip from the network"),
-        )
+        ).subcommand(
-        .subcommand(
             SubCommand::with_name("airdrop")
                 .about("Request a batch of tokens")
                 .arg(
@@ -96,8 +93,7 @@ fn main() -> Result<(), Box<error::Error>> {
                         .required(true)
                         .help("The number of tokens to request"),
                 ),
-        )
+        ).subcommand(
-        .subcommand(
             SubCommand::with_name("pay")
                 .about("Send a payment")
                 .arg(
@@ -107,16 +103,14 @@ fn main() -> Result<(), Box<error::Error>> {
                         .takes_value(true)
                         .required(true)
                         .help("The number of tokens to send"),
-                )
+                ).arg(
-                .arg(
                     Arg::with_name("to")
                         .long("to")
                         .value_name("PUBKEY")
                         .takes_value(true)
                         .help("The pubkey of recipient"),
                 ),
-        )
+        ).subcommand(
-        .subcommand(
             SubCommand::with_name("confirm")
                 .about("Confirm your payment by signature")
                 .arg(
@@ -126,8 +120,7 @@ fn main() -> Result<(), Box<error::Error>> {
                         .required(true)
                         .help("The transaction signature to confirm"),
                 ),
-        )
+        ).subcommand(SubCommand::with_name("balance").about("Get your balance"))
-        .subcommand(SubCommand::with_name("balance").about("Get your balance"))
         .subcommand(SubCommand::with_name("address").about("Get your public key"))
         .get_matches();
 

src/blob_fetch_stage.rs

@@ -32,8 +32,7 @@ impl BlobFetchStage {
             .into_iter()
             .map(|socket| {
                 streamer::blob_receiver(socket, exit.clone(), recycler.clone(), sender.clone())
-            })
+            }).collect();
-            .collect();
 
         (BlobFetchStage { exit, thread_hdls }, receiver)
     }

src/broadcast_stage.rs

@@ -183,8 +183,7 @@ impl BroadcastStage {
             .name("solana-broadcaster".to_string())
             .spawn(move || {
                 Self::run(&sock, &crdt, &window, entry_height, &recycler, &receiver);
-            })
+            }).unwrap();
-            .unwrap();
 
         BroadcastStage { thread_hdl }
     }

src/counter.rs

@@ -86,8 +86,7 @@ impl Counter {
                     .add_field(
                         "count",
                         influxdb::Value::Integer(counts as i64 - lastlog as i64),
-                    )
+                    ).to_owned(),
-                    .to_owned(),
             );
             self.lastlog
                 .compare_and_swap(lastlog, counts, Ordering::Relaxed);

src/crdt.rs

@@ -277,8 +277,7 @@ impl Crdt {
                     node.contact_info.rpu.to_string(),
                     node.contact_info.tpu.to_string()
                 )
-            })
+            }).collect();
-            .collect();
 
         format!(
             " NodeInfo.contact_info     | Node identifier\n\
@@ -402,8 +401,7 @@ impl Crdt {
                     trace!("{} purge skipped {} {} {}", self.id, k, now - v, limit);
                     None
                 }
-            })
+            }).collect();
-            .collect();
 
         inc_new_counter_info!("crdt-purge-count", dead_ids.len());
 
@@ -450,8 +448,7 @@ impl Crdt {
                     trace!("{}:broadcast node {} {}", me.id, v.id, v.contact_info.tvu);
                     true
                 }
-            })
+            }).cloned()
-            .cloned()
             .collect();
         cloned_table
     }
@@ -552,8 +549,7 @@ impl Crdt {
                     v.contact_info.tvu
                 );
                 e
-            })
+            }).collect();
-            .collect();
 
         trace!("broadcast results {}", errs.len());
         for e in errs {
@@ -607,8 +603,7 @@ impl Crdt {
                 } else {
                     true
                 }
-            })
+            }).collect();
-            .collect();
         trace!("retransmit orders {}", orders.len());
         let errs: Vec<_> = orders
             .par_iter()
@@ -623,8 +618,7 @@ impl Crdt {
                 //TODO profile this, may need multiple sockets for par_iter
                 assert!(rblob.meta.size <= BLOB_SIZE);
                 s.send_to(&rblob.data[..rblob.meta.size], &v.contact_info.tvu)
-            })
+            }).collect();
-            .collect();
         for e in errs {
             if let Err(e) = &e {
                 inc_new_counter_info!("crdt-retransmit-send_to_error", 1, 1);
@@ -666,8 +660,7 @@ impl Crdt {
                 r.id != Pubkey::default()
                     && (Self::is_valid_address(&r.contact_info.tpu)
                         || Self::is_valid_address(&r.contact_info.tvu))
-            })
+            }).map(|x| x.ledger_state.last_id)
-            .map(|x| x.ledger_state.last_id)
             .collect()
     }
 
@@ -702,8 +695,7 @@ impl Crdt {
                 v.id != self.id
                     && !v.contact_info.ncp.ip().is_unspecified()
                     && !v.contact_info.ncp.ip().is_multicast()
-            })
+            }).collect();
-            .collect();
 
         let choose_peer_strategy = ChooseWeightedPeerStrategy::new(
             &self.remote,
@@ -867,8 +859,7 @@ impl Crdt {
                     let time_left = GOSSIP_SLEEP_MILLIS - elapsed;
                     sleep(Duration::from_millis(time_left));
                 }
-            })
+            }).unwrap()
-            .unwrap()
     }
     fn run_window_request(
         from: &NodeInfo,
@@ -1191,8 +1182,7 @@ impl Crdt {
                         me.table.len()
                     );
                 }
-            })
+            }).unwrap()
-            .unwrap()
     }
 
     fn is_valid_ip(addr: IpAddr) -> bool {

src/drone.rs

@@ -143,12 +143,10 @@ impl Drone {
                     .add_field(
                         "request_amount",
                         influxdb::Value::Integer(request_amount as i64),
-                    )
+                    ).add_field(
-                    .add_field(
                         "request_current",
                         influxdb::Value::Integer(self.request_current as i64),
-                    )
+                    ).to_owned(),
-                    .to_owned(),
             );
             client.retry_transfer_signed(&tx, 10)
         } else {
@@ -215,8 +213,7 @@ pub fn run_local_drone(mint_keypair: Keypair, network: SocketAddr, sender: Sende
                             io::ErrorKind::Other,
                             format!("Drone response: {:?}", err),
                         ))
-                    }))
+                    })).then(|_| Ok(()));
-                    .then(|_| Ok(()));
                 tokio::spawn(server)
             });
         tokio::run(done);

src/entry.rs

@@ -112,7 +112,8 @@ impl Entry {
             id: Hash::default(),
             transactions,
             has_more: false,
-        }).unwrap() <= BLOB_DATA_SIZE as u64
+        }).unwrap()
+            <= BLOB_DATA_SIZE as u64
     }
 
     /// Creates the next Tick Entry `num_hashes` after `start_hash`.

src/erasure.rs

@@ -327,8 +327,7 @@ pub fn generate_coding(
             .map(|(i, l)| {
                 trace!("{} i: {} data: {}", id, i, l.data[0]);
                 &l.data[..max_data_size]
-            })
+            }).collect();
-            .collect();
 
         let mut coding_locks: Vec<_> = coding_blobs
             .iter()
@@ -341,8 +340,7 @@ pub fn generate_coding(
             .map(|(i, l)| {
                 trace!("{} i: {} coding: {}", id, i, l.data[0],);
                 &mut l.data_mut()[..max_data_size]
-            })
+            }).collect();
-            .collect();
 
         generate_coding_blocks(coding_ptrs.as_mut_slice(), &data_ptrs)?;
         debug!(

src/fetch_stage.rs

@@ -33,8 +33,7 @@ impl FetchStage {
             .into_iter()
             .map(|socket| {
                 streamer::receiver(socket, exit.clone(), recycler.clone(), sender.clone())
-            })
+            }).collect();
-            .collect();
 
         (FetchStage { exit, thread_hdls }, receiver)
     }

src/fullnode.rs

@@ -384,8 +384,7 @@ mod tests {
                 let bank = Bank::new(&alice);
                 let entry = tn.info.clone();
                 Fullnode::new_with_bank(keypair, bank, 0, &[], tn, Some(&entry), None, false)
-            })
+            }).collect();
-            .collect();
         //each validator can exit in parallel to speed many sequential calls to `join`
         vals.iter().for_each(|v| v.exit());
         //while join is called sequentially, the above exit call notified all the

src/ledger.rs

@@ -598,8 +598,7 @@ mod tests {
                     )],
                     false,
                 )
-            })
+            }).collect()
-            .collect()
     }
 
     fn make_test_entries() -> Vec<Entry> {

src/metrics.rs

@@ -342,12 +342,10 @@ mod test {
             .add_field(
                 "random_bool",
                 influxdb::Value::Boolean(random::<u8>() < 128),
-            )
+            ).add_field(
-            .add_field(
                 "random_int",
                 influxdb::Value::Integer(random::<u8>() as i64),
-            )
+            ).to_owned();
-            .to_owned();
         agent.submit(point);
     }
 

src/record_stage.rs

@@ -39,8 +39,7 @@ impl RecordStage {
             .spawn(move || {
                 let mut recorder = Recorder::new(start_hash);
                 let _ = Self::process_signals(&mut recorder, &signal_receiver, &entry_sender);
-            })
+            }).unwrap();
-            .unwrap();
 
         (RecordStage { thread_hdl }, entry_receiver)
     }
@@ -72,8 +71,7 @@ impl RecordStage {
                     }
                     recorder.hash();
                 }
-            })
+            }).unwrap();
-            .unwrap();
 
         (RecordStage { thread_hdl }, entry_receiver)
     }

src/replicate_stage.rs

@@ -111,8 +111,7 @@ impl ReplicateStage {
                         _ => error!("{:?}", e),
                     }
                 }
-            })
+            }).unwrap();
-            .unwrap();
 
         let thread_hdls = vec![t_responder, t_replicate];
 

src/request_stage.rs

@@ -28,8 +28,7 @@ impl RequestStage {
                 deserialize(&x.data[0..x.meta.size])
                     .map(|req| (req, x.meta.addr()))
                     .ok()
-            })
+            }).collect()
-            .collect()
     }
 
     pub fn process_request_packets(
@@ -103,8 +102,7 @@ impl RequestStage {
                         _ => error!("{:?}", e),
                     }
                 }
-            })
+            }).unwrap();
-            .unwrap();
         (
             RequestStage {
                 thread_hdl,

src/retransmit_stage.rs

@@ -69,8 +69,7 @@ fn retransmitter(
                 }
             }
             trace!("exiting retransmitter");
-        })
+        }).unwrap()
-        .unwrap()
 }
 
 pub struct RetransmitStage {

src/sigverify.rs

@@ -96,8 +96,7 @@ pub fn ed25519_verify_cpu(batches: &[SharedPackets]) -> Vec<Vec<u8>> {
                 .par_iter()
                 .map(verify_packet)
                 .collect()
-        })
+        }).collect();
-        .collect();
     inc_new_counter_info!("ed25519_verify_cpu", count);
     rv
 }
@@ -115,8 +114,7 @@ pub fn ed25519_verify_disabled(batches: &[SharedPackets]) -> Vec<Vec<u8>> {
                 .par_iter()
                 .map(verify_packet_disabled)
                 .collect()
-        })
+        }).collect();
-        .collect();
     inc_new_counter_info!("ed25519_verify_disabled", count);
     rv
 }

src/sigverify_stage.rs

@@ -89,8 +89,7 @@ impl SigVerifyStage {
                 .add_field(
                     "total_time_ms",
                     influxdb::Value::Integer(total_time_ms as i64),
-                )
+                ).to_owned(),
-                .to_owned(),
         );
 
         Ok(())

src/streamer.rs

@@ -58,8 +58,7 @@ pub fn receiver(
         .spawn(move || {
             let _ = recv_loop(&sock, &exit, &recycler, &packet_sender);
             ()
-        })
+        }).unwrap()
-        .unwrap()
 }
 
 fn recv_send(sock: &UdpSocket, recycler: &BlobRecycler, r: &BlobReceiver) -> Result<()> {
@@ -104,8 +103,7 @@ pub fn responder(
                     _ => warn!("{} responder error: {:?}", name, e),
                 }
             }
-        })
+        }).unwrap()
-        .unwrap()
 }
 
 //TODO, we would need to stick block authentication before we create the
@@ -137,8 +135,7 @@ pub fn blob_receiver(
                 break;
             }
             let _ = recv_blobs(&recycler, &sock, &s);
-        })
+        }).unwrap()
-        .unwrap()
 }
 
 #[cfg(test)]

src/thin_client.rs

@@ -163,8 +163,7 @@ impl ThinClient {
                 .add_field(
                     "duration_ms",
                     influxdb::Value::Integer(timing::duration_as_ms(&now.elapsed()) as i64),
-                )
+                ).to_owned(),
-                .to_owned(),
         );
         result
     }
@@ -285,8 +284,7 @@ impl ThinClient {
                 .add_field(
                     "duration_ms",
                     influxdb::Value::Integer(timing::duration_as_ms(elapsed) as i64),
-                )
+                ).to_owned(),
-                .to_owned(),
         );
     }
 
@@ -359,8 +357,7 @@ impl ThinClient {
                 .add_field(
                     "duration_ms",
                     influxdb::Value::Integer(timing::duration_as_ms(&now.elapsed()) as i64),
-                )
+                ).to_owned(),
-                .to_owned(),
         );
         self.signature_status
     }

src/vote_stage.rs

@@ -83,8 +83,7 @@ fn get_last_id_to_vote_on(
             .add_field(
                 "valid_peers",
                 influxdb::Value::Integer(valid_ids.len() as i64),
-            )
+            ).to_owned(),
-            .to_owned(),
     );
 
     if valid_ids.len() > super_majority_index {
@@ -103,8 +102,7 @@ fn get_last_id_to_vote_on(
                 .add_field(
                     "duration_ms",
                     influxdb::Value::Integer((now - *last_valid_validator_timestamp) as i64),
-                )
+                ).to_owned(),
-                .to_owned(),
         );
     }
 
@@ -406,8 +404,7 @@ pub mod tests {
                 // sleep to get a different timestamp in the bank
                 sleep(Duration::from_millis(1));
                 last_id
-            })
+            }).collect();
-            .collect();
 
         // see that we fail to have 2/3rds consensus
         assert!(

src/wallet.rs

@@ -279,8 +279,7 @@ mod tests {
                             .required(true)
                             .help("The number of tokens to request"),
                     ),
-            )
+            ).subcommand(
-            .subcommand(
                 SubCommand::with_name("pay")
                     .about("Send a payment")
                     .arg(
@@ -290,16 +289,14 @@ mod tests {
                             .takes_value(true)
                             .required(true)
                             .help("The number of tokens to send"),
-                    )
+                    ).arg(
-                    .arg(
                         Arg::with_name("to")
                             .long("to")
                             .value_name("PUBKEY")
                             .takes_value(true)
                             .help("The pubkey of recipient"),
                     ),
-            )
+            ).subcommand(
-            .subcommand(
                 SubCommand::with_name("confirm")
                     .about("Confirm your payment by signature")
                     .arg(
@@ -309,8 +306,7 @@ mod tests {
                             .required(true)
                             .help("The transaction signature to confirm"),
                     ),
-            )
+            ).subcommand(SubCommand::with_name("balance").about("Get your balance"))
-            .subcommand(SubCommand::with_name("balance").about("Get your balance"))
             .subcommand(SubCommand::with_name("address").about("Get your public key"));
 
         let test_airdrop = test_commands

src/window.rs

@@ -90,8 +90,7 @@ impl WindowUtil for Window {
                 }
                 self[i].clear_data(recycler);
                 Some(pix)
-            })
+            }).collect()
-            .collect()
     }
 
     fn repair(
@@ -140,8 +139,7 @@ impl WindowUtil for Window {
                 } else {
                     " "
                 }
-            })
+            }).collect();
-            .collect();
 
         let buf: Vec<_> = self
             .iter()
@@ -157,8 +155,7 @@ impl WindowUtil for Window {
                     // data.is_none()
                     "c"
                 }
-            })
+            }).collect();
-            .collect();
         format!(
             "\n{}: WINDOW ({}): {}\n{}: WINDOW ({}): {}",
             id,

src/window_service.rs

@@ -61,7 +61,8 @@ fn add_block_to_retransmit_queue(
         leader_id
     );
     if p.get_id()
-        .expect("get_id in fn add_block_to_retransmit_queue") == leader_id
+        .expect("get_id in fn add_block_to_retransmit_queue")
+        == leader_id
     {
         //TODO
         //need to copy the retransmitted blob
@@ -293,8 +294,7 @@ pub fn window_service(
                     });
                 }
             }
-        })
+        }).unwrap()
-        .unwrap()
 }
 
 #[cfg(test)]
@@ -561,8 +561,7 @@ mod test {
                         let rv = repair_backoff(&mut last, &mut times, 1) as usize;
                         assert_eq!(times, x + 2);
                         rv
-                    })
+                    }).sum();
-                    .sum();
                 assert_eq!(times, 128);
                 assert_eq!(last, 1);
                 repair_backoff(&mut last, &mut times, 1);
@@ -571,8 +570,7 @@ mod test {
                 assert_eq!(times, 2);
                 assert_eq!(last, 2);
                 total
-            })
+            }).sum();
-            .sum();
         let avg = res / num_tests;
         assert!(avg >= 3);
         assert!(avg <= 5);

src/write_stage.rs

@@ -126,8 +126,7 @@ impl WriteStage {
                         error!("{:?}", e);
                     }
                 }
-            })
+            }).unwrap();
-            .unwrap();
 
         let thread_hdls = vec![t_responder, thread_hdl];
         (WriteStage { thread_hdls }, blob_receiver)

tests/data_replicator.rs

@@ -176,8 +176,7 @@ pub fn crdt_retransmit() -> result::Result<()> {
             s.set_read_timeout(Some(Duration::new(1, 0))).unwrap();
             let res = s.recv_from(&mut b.data);
             res.is_err() //true if failed to receive the retransmit packet
-        })
+        }).collect();
-        .collect();
     //true if failed receive the retransmit packet, r2, and r3 should succeed
     //r1 was the sender, so it should fail to receive the packet
     assert_eq!(res, [true, false, false]);

tests/multinode.rs

@@ -593,10 +593,8 @@ fn test_multi_node_dynamic_network() {
                     assert_eq!(bal, Some(500));
                     info!("sent balance to[{}/{}] {}", n, num_nodes, keypair.pubkey());
                     keypair
-                })
+                }).unwrap()
-                .unwrap()
+        }).collect();
-        })
-        .collect();
 
     info!("Waiting for keypairs to be created");
     let keypairs: Vec<_> = t1.into_iter().map(|t| t.join().unwrap()).collect();
@@ -622,10 +620,8 @@ fn test_multi_node_dynamic_network() {
                         true,
                     );
                     (rd, val)
-                })
+                }).unwrap()
-                .unwrap()
+        }).collect();
-        })
-        .collect();
 
     let mut validators: Vec<_> = t2.into_iter().map(|t| t.join().unwrap()).collect();
 
@@ -645,8 +641,7 @@ fn test_multi_node_dynamic_network() {
                 &alice_arc.read().unwrap().keypair(),
                 bob_pubkey,
                 &last_id,
-            )
+            ).unwrap();
-            .unwrap();
 
         expected_balance += 500;