add dashboard

amqp_transport/Cargo.toml

@@ -6,6 +6,7 @@ edition = "2021"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
+amqp_core = { path = "../amqp_core" }
 anyhow = { version = "1.0.53", features = ["backtrace"] }
 nom = "7.1.0"
 once_cell = "1.9.0"
@@ -14,7 +15,6 @@ regex = "1.5.4"
 thiserror = "1.0.30"
 tokio = { version = "1.16.1", features = ["full"] }
 tracing = "0.1.30"
-tracing-subscriber = "0.3.8"
 uuid = "0.8.2"
 
 [features]

amqp_transport/src/connection.rs

@@ -26,15 +26,17 @@ pub struct Connection {
     max_frame_size: usize,
     heartbeat_delay: u16,
     channel_max: u16,
+    connection_handle: amqp_core::ConnectionHandle,
 }
 
 impl Connection {
-    pub fn new(stream: TcpStream) -> Self {
+    pub fn new(stream: TcpStream, connection_handle: amqp_core::ConnectionHandle) -> Self {
         Self {
             stream,
             max_frame_size: FRAME_SIZE_MIN_MAX,
             heartbeat_delay: HEARTBEAT_DELAY,
             channel_max: CHANNEL_MAX,
+            connection_handle,
         }
     }
 
@@ -43,6 +45,9 @@ impl Connection {
             Ok(()) => {}
             Err(err) => error!(%err, "Error during processing of connection"),
         }
+
+        let connection_handle = self.connection_handle.lock();
+        connection_handle.close();
     }
 
     pub async fn process_connection(&mut self) -> Result<()> {

amqp_transport/src/lib.rs

@@ -9,25 +9,34 @@ mod sasl;
 mod tests;
 
 use crate::connection::Connection;
+use amqp_core::GlobalData;
 use anyhow::Result;
 use tokio::net;
 use tracing::{info, info_span, Instrument};
 use uuid::Uuid;
 
-pub async fn do_thing_i_guess() -> Result<()> {
+pub async fn do_thing_i_guess(global_data: GlobalData) -> Result<()> {
     info!("Binding TCP listener...");
     let listener = net::TcpListener::bind(("127.0.0.1", 5672)).await?;
     info!(addr = ?listener.local_addr()?, "Successfully bound TCP listener");
 
     loop {
-        let (stream, _) = listener.accept().await?;
+        let (stream, peer_addr) = listener.accept().await?;
 
         let id = Uuid::from_bytes(rand::random());
 
         info!(local_addr = ?stream.local_addr(), %id, "Accepted new connection");
         let span = info_span!("client-connection", %id);
 
-        let connection = Connection::new(stream);
+        let connection_handle =
+            amqp_core::Connection::new_handle(id, peer_addr, global_data.clone());
+
+        let mut global_data = global_data.lock();
+        global_data
+            .connections
+            .insert(id, connection_handle.clone());
+
+        let connection = Connection::new(stream, connection_handle);
 
         tokio::spawn(connection.start_connection_processing().instrument(span));
     }