apply clippy suggestions, remove unwraps

bin/src/command/requests.rs

@@ -388,12 +388,12 @@ impl GatheringTask for LoadStaticConfigTask {
     ) {
         let mut messages = vec![];
         for (worker_id, response) in self.gatherer.responses {
-            match ResponseStatus::try_from(response.status).unwrap() {
-                ResponseStatus::Ok => {}
-                ResponseStatus::Failure => {
+            match ResponseStatus::try_from(response.status) {
+                Ok(ResponseStatus::Failure) => {
                     messages.push(format!("worker {worker_id}: {}", response.message))
                 }
-                ResponseStatus::Processing => {}
+                Ok(ResponseStatus::Ok) | Ok(ResponseStatus::Processing) => {}
+                Err(e) => warn!("error decoding response status: {}", e),
             }
         }
 
@@ -468,12 +468,11 @@ impl GatheringTask for WorkerTask {
         let mut messages = vec![];
 
         for (worker_id, response) in self.gatherer.responses {
-            match ResponseStatus::try_from(response.status).unwrap() {
-                ResponseStatus::Ok => messages.push(format!("{worker_id}: OK")),
-                ResponseStatus::Failure => {
+            match ResponseStatus::try_from(response.status) {
+                Ok(ResponseStatus::Ok) => messages.push(format!("{worker_id}: OK")),
+                Ok(ResponseStatus::Failure) | Ok(ResponseStatus::Processing) | Err(_) => {
                     messages.push(format!("{worker_id}: {}", response.message))
                 }
-                ResponseStatus::Processing => {}
             }
         }
 
@@ -750,10 +749,14 @@ impl GatheringTask for StatusTask {
         _timed_out: bool,
     ) {
         for (worker_id, response) in self.gatherer.responses {
-            let new_run_state = match ResponseStatus::try_from(response.status).unwrap() {
-                ResponseStatus::Ok => RunState::Running,
-                ResponseStatus::Processing => continue,
-                ResponseStatus::Failure => RunState::NotAnswering,
+            let new_run_state = match ResponseStatus::try_from(response.status) {
+                Ok(ResponseStatus::Ok) => RunState::Running,
+                Ok(ResponseStatus::Processing) => continue,
+                Ok(ResponseStatus::Failure) => RunState::NotAnswering,
+                Err(e) => {
+                    warn!("error decoding response status: {}", e);
+                    continue;
+                }
             };
 
             self.worker_infos

bin/src/command/server.rs

@@ -153,13 +153,14 @@ impl Gatherer for DefaultGatherer {
         worker_id: WorkerId,
         message: WorkerResponse,
     ) {
-        match ResponseStatus::try_from(message.status).unwrap() {
-            ResponseStatus::Ok => self.ok += 1,
-            ResponseStatus::Failure => self.errors += 1,
-            ResponseStatus::Processing => client.return_processing(format!(
+        match ResponseStatus::try_from(message.status) {
+            Ok(ResponseStatus::Ok) => self.ok += 1,
+            Ok(ResponseStatus::Failure) => self.errors += 1,
+            Ok(ResponseStatus::Processing) => client.return_processing(format!(
                 "Worker {} is processing {}. {}",
                 worker_id, message.id, message.message
             )),
+            Err(e) => warn!("error decoding response status: {}", e),
         }
         self.responses.push((worker_id, message));
     }

bin/src/command/upgrade.rs

@@ -236,9 +236,8 @@ impl Gatherer for UpgradeWorkerTask {
         worker_id: WorkerId,
         message: WorkerResponse,
     ) {
-        // TODO: there's gotta be a better way than unwrapping here
-        match ResponseStatus::try_from(message.status).unwrap() {
-            ResponseStatus::Ok => {
+        match ResponseStatus::try_from(message.status) {
+            Ok(ResponseStatus::Ok) => {
                 self.ok += 1;
                 match self.progress {
                     UpgradeWorkerProgress::RequestingListenSockets { .. } => {}
@@ -250,11 +249,12 @@ impl Gatherer for UpgradeWorkerTask {
                     }
                 }
             }
-            ResponseStatus::Failure => self.errors += 1,
-            ResponseStatus::Processing => client.return_processing(format!(
+            Ok(ResponseStatus::Failure) => self.errors += 1,
+            Ok(ResponseStatus::Processing) => client.return_processing(format!(
                 "Worker {} is processing {}. {}",
                 worker_id, message.id, message.message
             )),
+            Err(e) => warn!("error decoding response status: {}", e),
         }
         self.responses.push((worker_id, message));
     }

bin/src/ctl/mod.rs

@@ -73,8 +73,6 @@ pub fn ctl(args: cli::Args) -> Result<(), CtlError> {
         setup_logging_with_config(&config, "CTL");
     }
 
-    debug!("started CLI with config: {:?}", config);
-
     // If the command is `config check` then exit because if we are here, the configuration is valid
     if let SubCmd::Config {
         cmd: ConfigCmd::Check,

bin/src/worker.rs

@@ -125,11 +125,8 @@ pub fn begin_worker_process(
     );
     info!("worker {} starting...", id);
 
-    // let initial_state = read_requests_from_file(&mut configuration_state_file)
-
     let initial_state = read_initial_state_from_file(&mut configuration_state_file)
         .map_err(WorkerError::ReadRequestsFromFile)?;
-    // .expect("could not parse configuration state data");
 
     worker_to_main_channel
         .nonblocking()
@@ -201,10 +198,8 @@ pub fn fork_main_into_worker(
     })?;
 
     state
-        // .write_requests_to_file(&mut state_file)
         .write_initial_state_to_file(&mut state_file)
         .map_err(WorkerError::WriteStateFile)?;
-    // .expect("Could not write state to file");
 
     state_file.rewind().map_err(WorkerError::Rewind)?;
 
@@ -234,7 +229,7 @@ pub fn fork_main_into_worker(
         }
     })?;
 
-    let worker_config = ServerConfig::from_config(config);
+    let worker_config = ServerConfig::from(config);
 
     let mut main_to_worker_channel: Channel<ServerConfig, WorkerResponse> = Channel::new(
         main_to_worker,

command/src/channel.rs

@@ -361,7 +361,7 @@ impl<Tx: Debug + ProstMessage + Default, Rx: Debug + ProstMessage + Default> Cha
 
             if buffer.len() >= message_len {
                 let message = Rx::decode(&buffer[delimiter_size()..message_len])
-                    .map_err(|decode_error| ChannelError::InvalidProtobufMessage(decode_error))?;
+                    .map_err(ChannelError::InvalidProtobufMessage)?;
                 self.front_buf.consume(message_len);
                 return Ok(Some(message));
             }

command/src/config.rs

@@ -1747,16 +1747,21 @@ fn display_toml_error(file: &str, error: &toml::de::Error) {
 }
 
 impl ServerConfig {
-    /// reduce the config to the bare minimum needed by a worker
-    pub fn from_config(config: &Config) -> ServerConfig {
-        let metrics = config.metrics.clone().and_then(|m| {
-            Some(ServerMetricsConfig {
-                address: m.address.to_string(),
-                tagged_metrics: m.tagged_metrics,
-                prefix: m.prefix,
-            })
+    /// size of the slab for the Session manager
+    pub fn slab_capacity(&self) -> u64 {
+        10 + 2 * self.max_connections
+    }
+}
+
+/// reduce the config to the bare minimum needed by a worker
+impl From<&Config> for ServerConfig {
+    fn from(config: &Config) -> Self {
+        let metrics = config.metrics.clone().map(|m| ServerMetricsConfig {
+            address: m.address.to_string(),
+            tagged_metrics: m.tagged_metrics,
+            prefix: m.prefix,
         });
-        ServerConfig {
+        Self {
             max_connections: config.max_connections as u64,
             front_timeout: config.front_timeout,
             back_timeout: config.back_timeout,
@@ -1774,11 +1779,6 @@ impl ServerConfig {
             metrics,
         }
     }
-
-    /// size of the slab for the Session manager
-    pub fn slab_capacity(&self) -> u64 {
-        10 + 2 * self.max_connections
-    }
 }
 
 #[cfg(test)]

command/src/request.rs

@@ -7,7 +7,7 @@ use std::{
     str::FromStr,
 };
 
-use prost::Message;
+use prost::{DecodeError, Message};
 
 use crate::{
     proto::{
@@ -26,9 +26,9 @@ pub enum RequestError {
     #[error("invalid value {value} for field '{name}'")]
     InvalidValue { name: String, value: i32 },
     #[error("Could not read requests from file: {0}")]
-    FileError(std::io::Error),
-    #[error("Could not parse requests: {0}")]
-    ParseError(String),
+    ReadFile(std::io::Error),
+    #[error("Could not decode requests: {0}")]
+    Decode(DecodeError),
 }
 
 impl Request {
@@ -135,14 +135,16 @@ impl fmt::Display for WorkerRequest {
 
 pub fn read_initial_state_from_file(file: &mut File) -> Result<InitialState, RequestError> {
     let mut buf_reader = BufReader::new(file);
+    read_initial_state(&mut buf_reader)
+}
+
+pub fn read_initial_state<R: Read>(reader: &mut R) -> Result<InitialState, RequestError> {
     let mut buffer = Vec::new();
-    buf_reader
+    reader
         .read_to_end(&mut buffer)
-        .map_err(|e| RequestError::FileError(e))?;
+        .map_err(RequestError::ReadFile)?;
 
-    let initial_state =
-        InitialState::decode(&buffer[..]).map_err(|e| RequestError::ParseError(e.to_string()))?;
-    Ok(initial_state)
+    InitialState::decode(&buffer[..]).map_err(RequestError::Decode)
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]

command/src/scm_socket.rs

@@ -116,7 +116,7 @@ impl ScmSocket {
         debug!("{} received :{:?}", self.fd, (size, file_descriptor_length));
 
         let listeners_count = ListenersCount::decode_length_delimited(&buf[..size])
-            .map_err(|error| ScmSocketError::DecodeError(error))?;
+            .map_err(ScmSocketError::DecodeError)?;
 
         let mut http_addresses = parse_addresses(&listeners_count.http)?;
         let mut tls_addresses = parse_addresses(&listeners_count.tls)?;

command/src/state.rs

@@ -1383,11 +1383,9 @@ impl ConfigState {
 
     // create requests needed for a worker to recreate the state
     pub fn produce_initial_state(&self) -> InitialState {
-        let mut counter = 0usize;
         let mut worker_requests = Vec::new();
-        for request in self.generate_requests() {
+        for (counter, request) in self.generate_requests().into_iter().enumerate() {
             worker_requests.push(WorkerRequest::new(format!("SAVE-{counter}"), request));
-            counter += 1;
         }
         InitialState {
             requests: worker_requests,

e2e/src/sozu/worker.rs

@@ -53,7 +53,7 @@ impl Worker {
         let config = ConfigBuilder::new(file_config, "")
             .into_config()
             .expect("could not create Config");
-        ServerConfig::from_config(&config)
+        ServerConfig::from(&config)
     }
 
     pub fn empty_config() -> (ServerConfig, Listeners, ConfigState) {