Add events listener mechanism

First attempt at fixing #443

Author: Keruspe

examples/connection.rs

@@ -1,3 +1,4 @@
+use futures_lite::stream::StreamExt;
 use lapin::{
     BasicProperties, Connection, ConnectionProperties, message::DeliveryResult, options::*,
     publisher_confirm::Confirmation, types::FieldTable,
@@ -18,11 +19,21 @@ fn main() {
             .await
             .expect("connection error");
 
+        let mut events_listener = conn.events_listener();
+
+        async_global_executor::spawn(async move {
+            while let Some(event) = events_listener.next().await {
+                info!(?event, "GOT EVENT");
+            }
+        })
+        .detach();
+
         info!("CONNECTED");
 
         {
             //send channel
             let channel_a = conn.create_channel().await.expect("create_channel");
+
             //receive channel
             let channel_b = conn.create_channel().await.expect("create_channel");
             info!(state=?conn.status().state());

src/channel.rs

@@ -12,6 +12,7 @@ use crate::{
     consumer::Consumer,
     consumers::Consumers,
     error_handler::ErrorHandler,
+    events::EventsSender,
     frames::{ExpectedReply, Frames},
     internal_rpc::InternalRPCHandle,
     message::{BasicGetMessage, BasicReturnMessage, Delivery},
@@ -54,9 +55,10 @@ pub struct Channel {
     internal_rpc: InternalRPCHandle,
     frames: Frames,
     error_handler: ErrorHandler,
+    events_sender: EventsSender,
     executor: Arc<dyn FullExecutor + Send + Sync>,
     channel_closer: Option<Arc<ChannelCloser>>,
-    connection_closer: Option<Arc<ConnectionCloser>>,
+    _connection_closer: Option<Arc<ConnectionCloser>>,
     recovery_config: RecoveryConfig,
 }
 
@@ -94,6 +96,7 @@ impl Channel {
         executor: Arc<dyn FullExecutor + Send + Sync>,
         connection_closer: Option<Arc<ConnectionCloser>>,
         recovery_config: RecoveryConfig,
+        events_sender: EventsSender,
     ) -> Channel {
         let returned_messages = ReturnedMessages::default();
         let status = ChannelStatus::new(channel_id, internal_rpc.clone());
@@ -120,9 +123,10 @@ impl Channel {
             internal_rpc,
             frames,
             error_handler: ErrorHandler::default(),
+            events_sender,
             executor,
             channel_closer,
-            connection_closer,
+            _connection_closer: connection_closer,
             recovery_config,
         }
     }
@@ -193,25 +197,9 @@ impl Channel {
     }
 
     pub(crate) fn clone_internal(&self) -> Self {
-        Self {
-            id: self.id,
-            configuration: self.configuration.clone(),
-            status: self.status.clone(),
-            connection_status: self.connection_status.clone(),
-            local_registry: self.local_registry.clone(),
-            acknowledgements: self.acknowledgements.clone(),
-            consumers: self.consumers.clone(),
-            basic_get_delivery: self.basic_get_delivery.clone(),
-            returned_messages: self.returned_messages.clone(),
-            waker: self.waker.clone(),
-            internal_rpc: self.internal_rpc.clone(),
-            frames: self.frames.clone(),
-            error_handler: self.error_handler.clone(),
-            executor: self.executor.clone(),
-            channel_closer: None,
-            connection_closer: self.connection_closer.clone(),
-            recovery_config: self.recovery_config.clone(),
-        }
+        let mut this = self.clone();
+        this.channel_closer = None;
+        this
     }
 
     fn wake(&self) {
@@ -658,6 +646,7 @@ impl Channel {
         if !recover {
             self.set_closed(err);
             if let Some(error) = error {
+                self.events_sender.error(error.clone());
                 self.error_handler.on_error(error);
             }
         }
@@ -883,6 +872,7 @@ impl Channel {
         {
             self.connection_status.set_state(ConnectionState::Connected);
             resolver.resolve(*connection);
+            self.events_sender.connected();
             Ok(())
         } else {
             error!(?state, ?step, "Invalid state");
@@ -919,6 +909,7 @@ impl Channel {
 
     fn on_connection_blocked_received(&self, _method: protocol::connection::Blocked) -> Result<()> {
         self.connection_status.block();
+        self.events_sender.connection_blocked();
         Ok(())
     }
 
@@ -927,6 +918,7 @@ impl Channel {
         _method: protocol::connection::Unblocked,
     ) -> Result<()> {
         self.connection_status.unblock();
+        self.events_sender.connection_unblocked();
         self.wake();
         Ok(())
     }

src/channels.rs

@@ -3,6 +3,7 @@ use crate::{
     ConnectionState, ConnectionStatus, Error, ErrorKind, Promise, Result,
     connection_closer::ConnectionCloser,
     error_handler::ErrorHandler,
+    events::{Events, EventsSender},
     frames::Frames,
     heartbeat::Heartbeat,
     id_sequence::IdSequence,
@@ -36,6 +37,7 @@ pub(crate) struct Channels {
     heartbeat: Heartbeat,
     connection_killswitch: KillSwitch,
     error_handler: ErrorHandler,
+    events: Events,
     uri: AMQPUri,
     options: ConnectionProperties,
     recovery_config: RecoveryConfig,
@@ -52,6 +54,7 @@ impl Channels {
         executor: Arc<dyn FullExecutor + Send + Sync>,
         uri: AMQPUri,
         options: ConnectionProperties,
+        events: Events,
     ) -> Self {
         let recovery_config = options.recovery_config.clone().unwrap_or_default();
         let mut inner = Inner::new(configuration.clone(), waker, recovery_config.clone());
@@ -61,6 +64,7 @@ impl Channels {
             internal_rpc.clone(),
             frames.clone(),
             executor.clone(),
+            events.sender(),
             None,
         );
         channel0.set_state(ChannelState::Connected);
@@ -76,6 +80,7 @@ impl Channels {
             heartbeat,
             connection_killswitch: KillSwitch::default(),
             error_handler: ErrorHandler::default(),
+            events,
             uri,
             options,
             recovery_config,
@@ -88,6 +93,7 @@ impl Channels {
             self.internal_rpc.clone(),
             self.frames.clone(),
             self.executor.clone(),
+            self.events.sender(),
             connection_closer,
         )
     }
@@ -190,6 +196,7 @@ impl Channels {
         }
 
         self.frames.drop_pending(error.clone());
+        self.events.sender().error(error.clone());
         self.error_handler.on_error(error.clone());
         for (id, channel) in self.lock_inner().channels.iter() {
             self.frames.clear_expected_replies(*id, error.clone());
@@ -352,6 +359,7 @@ impl Channels {
             self.connection_status.clone(),
             self.clone(),
             self.internal_rpc.clone(),
+            self.events.clone(),
         )
         .start(self.uri.clone(), self.options.clone())
         .await?;
@@ -431,6 +439,7 @@ impl Inner {
         internal_rpc: InternalRPCHandle,
         frames: Frames,
         executor: Arc<dyn FullExecutor + Send + Sync>,
+        events_sender: EventsSender,
         connection_closer: Option<Arc<ConnectionCloser>>,
     ) -> Channel {
         debug!(%id, "create channel");
@@ -444,6 +453,7 @@ impl Inner {
             executor,
             connection_closer,
             self.recovery_config.clone(),
+            events_sender,
         )
     }
 
@@ -453,6 +463,7 @@ impl Inner {
         internal_rpc: InternalRPCHandle,
         frames: Frames,
         executor: Arc<dyn FullExecutor + Send + Sync>,
+        events_sender: EventsSender,
         connection_closer: Arc<ConnectionCloser>,
     ) -> Result<Channel> {
         debug!("create channel");
@@ -474,6 +485,7 @@ impl Inner {
                     internal_rpc,
                     frames,
                     executor,
+                    events_sender,
                     Some(connection_closer),
                 );
                 self.channels.insert(id, channel.clone_internal());

src/connection.rs

@@ -1,10 +1,11 @@
 use crate::{
-    ConnectionProperties, Error, ErrorKind, Promise, Result,
+    ConnectionProperties, Error, ErrorKind, Event, Promise, Result,
     channel::Channel,
     channels::Channels,
     configuration::Configuration,
     connection_closer::ConnectionCloser,
     connection_status::ConnectionStatus,
+    events::Events,
     frames::Frames,
     heartbeat::Heartbeat,
     internal_rpc::{InternalRPC, InternalRPCHandle},
@@ -17,6 +18,7 @@ use crate::{
 };
 use amq_protocol::frame::{AMQPFrame, ProtocolVersion};
 use async_trait::async_trait;
+use futures_core::Stream;
 use std::{fmt, io, sync::Arc};
 use tracing::{Level, level_enabled, trace};
 
@@ -35,6 +37,7 @@ pub struct Connection {
     configuration: Configuration,
     status: ConnectionStatus,
     channels: Channels,
+    events: Events,
     io_loop: ThreadHandle,
     closer: Arc<ConnectionCloser>,
 }
@@ -45,12 +48,14 @@ impl Connection {
         status: ConnectionStatus,
         channels: Channels,
         internal_rpc: InternalRPCHandle,
+        events: Events,
     ) -> Self {
         let closer = Arc::new(ConnectionCloser::new(status.clone(), internal_rpc));
         Self {
             configuration,
             status,
             channels,
+            events,
             io_loop: ThreadHandle::default(),
             closer,
         }
@@ -61,8 +66,9 @@ impl Connection {
         status: ConnectionStatus,
         channels: Channels,
         internal_rpc: InternalRPCHandle,
+        events: Events,
     ) -> Self {
-        let conn = Self::new(configuration, status, channels, internal_rpc);
+        let conn = Self::new(configuration, status, channels, internal_rpc, events);
         conn.closer.noop();
         conn
     }
@@ -120,6 +126,11 @@ impl Connection {
         channel.clone().channel_open(channel).await
     }
 
+    /// Get a Stream of connection Events
+    pub fn events_listener(&self) -> impl Stream<Item = Event> + Send + 'static {
+        self.events.listener()
+    }
+
     /// Block current thread while the connection is still active.
     /// This is useful when you only have a consumer and nothing else keeping your application
     /// "alive".
@@ -190,6 +201,7 @@ impl Connection {
         let socket_state = SocketState::default();
         let internal_rpc = InternalRPC::new(executor.clone(), socket_state.handle());
         let heartbeat = Heartbeat::new(status.clone(), executor.clone(), reactor.clone());
+        let events = Events::new();
         let channels = Channels::new(
             configuration.clone(),
             status.clone(),
@@ -200,8 +212,15 @@ impl Connection {
             executor,
             uri.clone(),
             options.clone(),
+            events.clone(),
+        );
+        let conn = Connection::new(
+            configuration,
+            status,
+            channels,
+            internal_rpc.handle(),
+            events,
         );
-        let conn = Connection::new(configuration, status, channels, internal_rpc.handle());
         let io_loop = IoLoop::new(
             conn.status.clone(),
             conn.configuration.clone(),
@@ -328,6 +347,7 @@ mod tests {
         let socket_state = SocketState::default();
         let internal_rpc = InternalRPC::new(executor.clone(), socket_state.handle());
         let heartbeat = Heartbeat::new(status.clone(), executor.clone(), reactor);
+        let events = Events::new();
         let channels = Channels::new(
             configuration.clone(),
             status.clone(),
@@ -338,8 +358,15 @@ mod tests {
             executor,
             uri.clone(),
             ConnectionProperties::default(),
+            events.clone(),
+        );
+        let conn = Connection::new(
+            configuration,
+            status,
+            channels,
+            internal_rpc.handle(),
+            events,
         );
-        let conn = Connection::new(configuration, status, channels, internal_rpc.handle());
         conn.status.set_state(ConnectionState::Connected);
         conn
     }