Fix various items turned up by linting

All of the changes here were recommended by clippy as improvements to
the code style, and to reduce future issues related to code clarity.

src/flowgger/input/tls/mod.rs

@@ -70,7 +70,7 @@ pub fn config_parse(config: &Config) -> (TlsConfig, String, u64) {
             x.as_str().expect("input.listen must be an ip:port string")
         })
         .to_owned();
-    let threads = get_default_threads(&config);
+    let threads = get_default_threads(config);
     let cert = config
         .lookup("input.tls_cert")
         .map_or(DEFAULT_CERT, |x| {
@@ -113,7 +113,7 @@ pub fn config_parse(config: &Config) -> (TlsConfig, String, u64) {
             x.as_bool()
                 .expect("input.tls_verify_peer must be a boolean")
         });
-    let ca_file: Option<PathBuf> = config.lookup("input.tls_ca_file").map_or(None, |x| {
+    let ca_file: Option<PathBuf> = config.lookup("input.tls_ca_file").and_then(|x| {
         Some(PathBuf::from(
             x.as_str()
                 .expect("input.tls_ca_file must be a path to a file"),
@@ -142,26 +142,28 @@ pub fn config_parse(config: &Config) -> (TlsConfig, String, u64) {
                 .expect(r#"input.framing must be a string set to "line", "nul" or "syslen""#)
         })
         .to_owned();
-    let mut acceptor_builder = match tls_modern {
-        false => SslAcceptorBuilder::mozilla_intermediate_raw(SslMethod::tls()),
-        true => SslAcceptorBuilder::mozilla_modern_raw(SslMethod::tls()),
-    }.unwrap();
+        let mut acceptor_builder = (if tls_modern {
+            SslAcceptorBuilder::mozilla_modern_raw(SslMethod::tls())
+        }
+        else {
+            SslAcceptorBuilder::mozilla_intermediate_raw(SslMethod::tls())
+        }).unwrap();
     {
         let mut ctx = acceptor_builder.builder_mut();
         if let Some(ca_file) = ca_file {
             ctx.set_ca_file(&ca_file)
                 .expect("Unable to read the trusted CA file");
         }
-        if verify_peer == false {
+        if !verify_peer {
             ctx.set_verify(SSL_VERIFY_NONE);
         } else {
             ctx.set_verify_depth(TLS_VERIFY_DEPTH);
             ctx.set_verify(SSL_VERIFY_PEER | SSL_VERIFY_FAIL_IF_NO_PEER_CERT);
         }
         let mut opts =
             SSL_OP_CIPHER_SERVER_PREFERENCE | SSL_OP_NO_SESSION_RESUMPTION_ON_RENEGOTIATION;
-        if compression == false {
-            opts = opts | SSL_OP_NO_COMPRESSION;
+        if !compression {
+            opts |= SSL_OP_NO_COMPRESSION;
         }
         ctx.set_options(opts);
         set_fs(&mut ctx);

src/flowgger/input/tls/tls_input.rs

@@ -17,7 +17,7 @@ pub struct TlsInput {
 
 impl TlsInput {
     pub fn new(config: &Config) -> TlsInput {
-        let (tls_config, listen, timeout) = config_parse(&config);
+        let (tls_config, listen, timeout) = config_parse(config);
         TlsInput {
             listen: listen,
             tls_config: tls_config,
@@ -35,17 +35,14 @@ impl Input for TlsInput {
     ) {
         let listener = TcpListener::bind(&self.listen as &str).unwrap();
         for client in listener.incoming() {
-            match client {
-                Ok(client) => {
-                    let _ = client.set_read_timeout(self.timeout);
-                    let tx = tx.clone();
-                    let (decoder, encoder) = (decoder.clone_boxed(), encoder.clone_boxed());
-                    let tls_config = self.tls_config.clone();
-                    thread::spawn(move || {
-                        handle_client(client, tx, decoder, encoder, tls_config);
-                    });
-                }
-                Err(_) => {}
+            if let Ok(client) = client {
+                let _ = client.set_read_timeout(self.timeout);
+                let tx = tx.clone();
+                let (decoder, encoder) = (decoder.clone_boxed(), encoder.clone_boxed());
+                let tls_config = self.tls_config.clone();
+                thread::spawn(move || {
+                    handle_client(client, tx, decoder, encoder, tls_config);
+                });
             }
         }
     }

src/flowgger/input/udp_input.rs

@@ -9,7 +9,7 @@ use std::str;
 use std::sync::mpsc::SyncSender;
 
 const DEFAULT_LISTEN: &'static str = "0.0.0.0:514";
-const MAX_UDP_PACKET_SIZE: usize = 65527;
+const MAX_UDP_PACKET_SIZE: usize = 65_527;
 const MAX_COMPRESSION_RATIO: usize = 5;
 
 pub struct UdpInput {
@@ -47,7 +47,7 @@ impl Input for UdpInput {
                 Err(_) => continue,
             };
             let line = &buf[..length];
-            if let Err(e) = handle_record_maybe_compressed(&line, &tx, &decoder, &encoder) {
+            if let Err(e) = handle_record_maybe_compressed(line, &tx, &decoder, &encoder) {
                 let _ = writeln!(stderr(), "{}", e);
             }
         }
@@ -66,15 +66,15 @@ fn handle_record_maybe_compressed(
         let mut decompressed = Vec::with_capacity(MAX_UDP_PACKET_SIZE * MAX_COMPRESSION_RATIO);
         match line.zlib_decode().read_to_end(&mut decompressed) {
             Ok(_) => handle_record(&decompressed, tx, decoder, encoder),
-            Err(_) => return Err("Corrupted compressed (zlib) record"),
+            Err(_) => Err("Corrupted compressed (zlib) record"),
         }
     } else if line.len() >= 24 && (line[0] == 0x1f && line[1] == 0x8b && line[2] == 0x08) {
         let mut decompressed = Vec::with_capacity(MAX_UDP_PACKET_SIZE * MAX_COMPRESSION_RATIO);
         match line.gz_decode()
             .and_then(|mut x| x.read_to_end(&mut decompressed))
         {
             Ok(_) => handle_record(&decompressed, tx, decoder, encoder),
-            Err(_) => return Err("Corrupted compressed (gzip) record"),
+            Err(_) => Err("Corrupted compressed (gzip) record"),
         }
     } else {
         handle_record(line, tx, decoder, encoder)

src/flowgger/mod.rs

@@ -56,20 +56,20 @@ fn get_input_tcpco(_config: &Config) -> ! {
 
 fn get_input(input_type: &str, config: &Config) -> Box<Input> {
     match input_type {
-        "redis" => Box::new(RedisInput::new(&config)) as Box<Input>,
-        "stdin" => Box::new(StdinInput::new(&config)) as Box<Input>,
-        "tcp" | "syslog-tcp" => Box::new(TcpInput::new(&config)) as Box<Input>,
-        "tcp_co" | "tcpco" | "syslog-tcp_co" | "syslog-tcpco" => get_input_tcpco(&config),
-        "tls" | "syslog-tls" => Box::new(TlsInput::new(&config)) as Box<Input>,
-        "tls_co" | "tlsco" | "syslog-tls_co" | "syslog-tlsco" => get_input_tlsco(&config),
-        "udp" => Box::new(UdpInput::new(&config)) as Box<Input>,
+        "redis" => Box::new(RedisInput::new(config)) as Box<Input>,
+        "stdin" => Box::new(StdinInput::new(config)) as Box<Input>,
+        "tcp" | "syslog-tcp" => Box::new(TcpInput::new(config)) as Box<Input>,
+        "tcp_co" | "tcpco" | "syslog-tcp_co" | "syslog-tcpco" => get_input_tcpco(config),
+        "tls" | "syslog-tls" => Box::new(TlsInput::new(config)) as Box<Input>,
+        "tls_co" | "tlsco" | "syslog-tls_co" | "syslog-tlsco" => get_input_tlsco(config),
+        "udp" => Box::new(UdpInput::new(config)) as Box<Input>,
         _ => panic!("Invalid input type: {}", input_type),
     }
 }
 
 #[cfg(feature = "kafka")]
 fn get_output_kafka(config: &Config) -> Box<Output> {
-    Box::new(KafkaOutput::new(&config)) as Box<Output>
+    Box::new(KafkaOutput::new(config)) as Box<Output>
 }
 
 #[cfg(not(feature = "kafka"))]
@@ -79,9 +79,9 @@ fn get_output_kafka(_config: &Config) -> ! {
 
 fn get_output(output_type: &str, config: &Config) -> Box<Output> {
     match output_type {
-        "stdout" | "debug" => Box::new(DebugOutput::new(&config)) as Box<Output>,
-        "kafka" => get_output_kafka(&config),
-        "tls" | "syslog-tls" => Box::new(TlsOutput::new(&config)) as Box<Output>,
+        "stdout" | "debug" => Box::new(DebugOutput::new(config)) as Box<Output>,
+        "kafka" => get_output_kafka(config),
+        "tls" | "syslog-tls" => Box::new(TlsOutput::new(config)) as Box<Output>,
         _ => panic!("Invalid output type: {}", output_type),
     }
 }

src/flowgger/output/kafka_output.rs

@@ -13,7 +13,7 @@ const KAFKA_DEFAULT_ACKS: i16 = 0;
 const KAFKA_DEFAULT_COALESCE: usize = 1;
 const KAFKA_DEFAULT_COMPRESSION: &'static str = "none";
 const KAFKA_DEFAULT_THREADS: u32 = 1;
-const KAFKA_DEFAULT_TIMEOUT: u64 = 60000;
+const KAFKA_DEFAULT_TIMEOUT: u64 = 60_000;
 
 pub struct KafkaOutput {
     config: KafkaConfig,
@@ -201,7 +201,7 @@ impl Output for KafkaOutput {
             let _ = writeln!(stderr(), "Output framing is ignored with the Kafka output");
         }
         for _ in 0..self.threads {
-            let arx = arx.clone();
+            let arx = Arc::clone(&arx);
             let config = self.config.clone();
             thread::spawn(move || {
                 let mut worker = KafkaWorker::new(arx, config);

src/flowgger/output/tls_output.rs

@@ -116,7 +116,7 @@ impl TlsWorker {
                     _ => return Err(e),
                 },
             };
-            if self.tls_config.async == false {
+            if !self.tls_config.async {
                 writer.flush()?;
             }
         }
@@ -125,7 +125,7 @@ impl TlsWorker {
     fn run(self) {
         let tls_config = &self.tls_config;
         let mut rng = rand::thread_rng();
-        let mut recovery_delay = tls_config.recovery_delay_init as f64;
+        let mut recovery_delay = f64::from(tls_config.recovery_delay_init);
         let mut last_recovery;
         loop {
             last_recovery = chrono::offset::Utc::now();
@@ -162,10 +162,10 @@ impl TlsWorker {
             }
             let now = chrono::offset::Utc::now();
             if now.signed_duration_since(last_recovery)
-                > chrono::Duration::milliseconds(tls_config.recovery_probe_time as i64)
+                > chrono::Duration::milliseconds(i64::from(tls_config.recovery_probe_time))
             {
-                recovery_delay = tls_config.recovery_delay_init as f64;
-            } else if recovery_delay < tls_config.recovery_delay_max as f64 {
+                recovery_delay = f64::from(tls_config.recovery_delay_init);
+            } else if recovery_delay < f64::from(tls_config.recovery_delay_max) {
                 let between = Range::new(0.0, recovery_delay);
                 let mut rng = rand::thread_rng();
                 recovery_delay += between.ind_sample(&mut rng);
@@ -189,7 +189,7 @@ fn new_tcp(connect_chosen: &str) -> Result<TcpStream, io::Error> {
 
 impl TlsOutput {
     pub fn new(config: &Config) -> TlsOutput {
-        let (tls_config, threads) = config_parse(&config);
+        let (tls_config, threads) = config_parse(config);
         TlsOutput {
             config: tls_config,
             threads: threads,
@@ -200,7 +200,7 @@ impl TlsOutput {
 impl Output for TlsOutput {
     fn start(&self, arx: Arc<Mutex<Receiver<Vec<u8>>>>, merger: Option<Box<Merger>>) {
         for _ in 0..self.threads {
-            let arx = arx.clone();
+            let arx = Arc::clone(&arx);
             let config = self.config.clone();
             let merger = match merger {
                 Some(ref merger) => Some(merger.clone_boxed()) as Option<Box<Merger + Send>>,
@@ -245,13 +245,13 @@ fn config_parse(config: &Config) -> (TlsConfig, u32) {
                 .to_owned()
         })
         .collect();
-    let cert: Option<PathBuf> = config.lookup("output.tls_cert").map_or(None, |x| {
+    let cert: Option<PathBuf> = config.lookup("output.tls_cert").and_then(|x| {
         Some(PathBuf::from(
             x.as_str()
                 .expect("output.tls_cert must be a path to a .pem file"),
         ))
     });
-    let key: Option<PathBuf> = config.lookup("output.tls_key").map_or(None, |x| {
+    let key: Option<PathBuf> = config.lookup("output.tls_key").and_then(|x| {
         Some(PathBuf::from(
             x.as_str()
                 .expect("output.tls_key must be a path to a .pem file"),
@@ -270,7 +270,7 @@ fn config_parse(config: &Config) -> (TlsConfig, u32) {
             x.as_bool()
                 .expect("output.tls_verify_peer must be a boolean")
         });
-    let ca_file: Option<PathBuf> = config.lookup("output.tls_ca_file").map_or(None, |x| {
+    let ca_file: Option<PathBuf> = config.lookup("output.tls_ca_file").and_then(|x| {
         Some(PathBuf::from(
             x.as_str()
                 .expect("output.tls_ca_file must be a path to a file"),
@@ -319,7 +319,7 @@ fn config_parse(config: &Config) -> (TlsConfig, u32) {
     let mut connector_builder = SslConnectorBuilder::new(SslMethod::tls()).unwrap();
     {
         let mut ctx = connector_builder.builder_mut();
-        if verify_peer == false {
+        if !verify_peer {
             ctx.set_verify(SSL_VERIFY_NONE);
         } else {
             ctx.set_verify_depth(TLS_VERIFY_DEPTH);
@@ -331,8 +331,8 @@ fn config_parse(config: &Config) -> (TlsConfig, u32) {
         }
         let mut opts =
             SSL_OP_CIPHER_SERVER_PREFERENCE | SSL_OP_NO_SESSION_RESUMPTION_ON_RENEGOTIATION;
-        if compression == false {
-            opts = opts | SSL_OP_NO_COMPRESSION;
+        if !compression {
+            opts |= SSL_OP_NO_COMPRESSION;
         }
         ctx.set_options(opts);
         set_fs(&mut ctx);

src/flowgger/record_capnp.rs

@@ -59,7 +59,7 @@ pub mod record {
     }
 
     impl<'a> Reader<'a> {
-        pub fn borrow<'b>(&'b self) -> Reader<'b> {
+        pub fn borrow(&self) -> Reader {
             Reader { ..*self }
         }
 
@@ -213,10 +213,10 @@ pub mod record {
         pub fn as_reader(self) -> Reader<'a> {
             ::capnp::traits::FromStructReader::new(self.builder.as_reader())
         }
-        pub fn borrow<'b>(&'b mut self) -> Builder<'b> {
+        pub fn borrow(&mut self) -> Builder {
             Builder { ..*self }
         }
-        pub fn borrow_as_reader<'b>(&'b self) -> Reader<'b> {
+        pub fn borrow_as_reader(&self) -> Reader {
             ::capnp::traits::FromStructReader::new(self.builder.as_reader())
         }
 
@@ -499,7 +499,7 @@ pub mod pair {
     }
 
     impl<'a> Reader<'a> {
-        pub fn borrow<'b>(&'b self) -> Reader<'b> {
+        pub fn borrow(&self) -> Reader {
             Reader { ..*self }
         }
 
@@ -577,10 +577,10 @@ pub mod pair {
         pub fn as_reader(self) -> Reader<'a> {
             ::capnp::traits::FromStructReader::new(self.builder.as_reader())
         }
-        pub fn borrow<'b>(&'b mut self) -> Builder<'b> {
+        pub fn borrow(&mut self) -> Builder {
             Builder { ..*self }
         }
-        pub fn borrow_as_reader<'b>(&'b self) -> Reader<'b> {
+        pub fn borrow_as_reader(&self) -> Reader {
             ::capnp::traits::FromStructReader::new(self.builder.as_reader())
         }
 
@@ -613,9 +613,9 @@ pub mod pair {
             self.builder.set_data_field::<u16>(0, 0);
             self.builder.get_pointer_field(1).clear();
             self.builder.set_bool_field(16, false);
-            self.builder.set_data_field::<f64>(1, 0u8 as f64);
-            self.builder.set_data_field::<i64>(1, 0u8 as i64);
-            self.builder.set_data_field::<u64>(1, 0u8 as u64);
+            self.builder.set_data_field::<f64>(1, 0f64);
+            self.builder.set_data_field::<i64>(1, 0i64);
+            self.builder.set_data_field::<u64>(1, 0u64);
             ::capnp::traits::FromStructBuilder::new(self.builder)
         }
     }
@@ -702,7 +702,7 @@ pub mod pair {
         }
 
         impl<'a> Reader<'a> {
-            pub fn borrow<'b>(&'b self) -> Reader<'b> {
+            pub fn borrow(&self) -> Reader {
                 Reader { ..*self }
             }
 
@@ -801,10 +801,10 @@ pub mod pair {
             pub fn as_reader(self) -> Reader<'a> {
                 ::capnp::traits::FromStructReader::new(self.builder.as_reader())
             }
-            pub fn borrow<'b>(&'b mut self) -> Builder<'b> {
+            pub fn borrow(&mut self) -> Builder {
                 Builder { ..*self }
             }
-            pub fn borrow_as_reader<'b>(&'b self) -> Reader<'b> {
+            pub fn borrow_as_reader(&self) -> Reader {
                 ::capnp::traits::FromStructReader::new(self.builder.as_reader())
             }
 

src/flowgger/splitter/nul_splitter.rs

@@ -40,7 +40,7 @@ impl<T: Read> Splitter<T> for NulSplitter {
             };
             if let Err(e) = handle_line(line, &tx, &decoder, &encoder) {
                 let line = line.trim();
-                if line.len() > 0 {
+                if !line.is_empty() {
                     let _ = writeln!(stderr(), "{}: [{}]", e, line.trim());
                 }
             }

src/flowgger/utils/mod.rs

@@ -10,14 +10,14 @@ impl PreciseTimestamp {
     pub fn now() -> Self {
         let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap();
         PreciseTimestamp {
-            ts: now.as_secs() as f64 + now.subsec_nanos() as f64 / 1e9,
+            ts: now.as_secs() as f64 + f64::from(now.subsec_nanos()) / 1e9,
         }
     }
 
     #[inline]
     pub fn from_datetime(tsd: DateTime<FixedOffset>) -> Self {
         PreciseTimestamp {
-            ts: tsd.timestamp() as f64 + tsd.naive_utc().nanosecond() as f64 / 1e9,
+            ts: tsd.timestamp() as f64 + f64::from(tsd.naive_utc().nanosecond()) / 1e9,
         }
     }