Better handling of Lapin connection issues and error delivered to the…

… waiter correctly

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
     name = "rabbitmq-consumer"
-    version = "1.0.2"
+    version = "1.0.3"
     authors = ["Dario Cancelliere <dario.cancelliere@facile.it>"]
     edition = "2018"
 

src/client/consumer/connection.rs

@@ -28,6 +28,12 @@ impl Connection {
     }
 
     pub async fn get_connection(&mut self) -> Result<Arc<LapinConnection>, ConnectionError> {
+        if let Ok(ref lapin) = self.lapin {
+            if lapin.status().errored() {
+                self.lapin = Err(ConnectionError::NotConnected);
+            }
+        }
+
         if self.lapin.is_err() {
             logger::log(&format!(
                 "Connecting to RabbitMQ at {}:{}...",

src/client/consumer/mod.rs

@@ -126,13 +126,7 @@ impl Consumer {
 
                 res
             }
-            Err(e) => {
-                for hook in &self.hooks {
-                    hook.write().await.on_error(&format!("{:?}", e));
-                }
-
-                Err(ConsumerError::ConnectionError(e))
-            }
+            Err(e) => Err(ConsumerError::ConnectionError(e)),
         }
     }
 

src/client/executor/mod.rs

@@ -4,7 +4,7 @@ mod waiter;
 use async_std::sync::{Arc, RwLock};
 
 use crate::client::consumer::{Consumer, ConsumerError, ConsumerResult};
-use crate::client::executor::events::EventsHandler;
+use crate::client::executor::events::{Events, EventsHandler};
 use crate::client::executor::waiter::Waiter;
 use crate::config::Config;
 
@@ -41,7 +41,9 @@ impl Executor {
             Ok(ConsumerResult::GenericOk) => ExecutorResult::Exit,
             Ok(ConsumerResult::Killed) => ExecutorResult::Killed,
             Err(e) => {
-                let waiter = self.waiter.read().await;
+                let mut waiter = self.waiter.write().await;
+                waiter.on_error(&format!("{:?}", e));
+
                 if waiter.is_to_close() {
                     return ExecutorResult::Error(e);
                 }