add channel open support

amqp_transport/src/connection.rs

@@ -1,12 +1,18 @@
+use crate::classes::Class;
 use crate::error::{ConException, ProtocolError, Result};
 use crate::frame::{Frame, FrameType};
 use crate::{classes, frame, sasl};
+use amqp_core::GlobalData;
 use anyhow::Context;
 use std::collections::HashMap;
 use std::net::SocketAddr;
+use std::pin::Pin;
+use std::time::Duration;
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
 use tokio::net::TcpStream;
-use tracing::{debug, error, info};
+use tokio::time;
+use tracing::{debug, error, info, warn};
+use uuid::Uuid;
 
 fn ensure_conn(condition: bool) -> Result<()> {
     if condition {
@@ -21,22 +27,42 @@ const CHANNEL_MAX: u16 = 0;
 const FRAME_SIZE_MAX: u32 = 0;
 const HEARTBEAT_DELAY: u16 = 0;
 
+pub struct Channel {
+    num: u16,
+    channel_handle: amqp_core::ChannelHandle,
+}
+
 pub struct Connection {
+    id: Uuid,
     stream: TcpStream,
     max_frame_size: usize,
     heartbeat_delay: u16,
     channel_max: u16,
+    next_timeout: Pin<Box<time::Sleep>>,
+    channels: HashMap<u16, Channel>,
     connection_handle: amqp_core::ConnectionHandle,
+    global_data: GlobalData,
 }
 
+const DEFAULT_TIMEOUT: Duration = Duration::from_secs(30);
+
 impl Connection {
-    pub fn new(stream: TcpStream, connection_handle: amqp_core::ConnectionHandle) -> Self {
+    pub fn new(
+        id: Uuid,
+        stream: TcpStream,
+        connection_handle: amqp_core::ConnectionHandle,
+        global_data: GlobalData,
+    ) -> Self {
         Self {
+            id,
             stream,
             max_frame_size: FRAME_SIZE_MIN_MAX,
             heartbeat_delay: HEARTBEAT_DELAY,
             channel_max: CHANNEL_MAX,
+            next_timeout: Box::pin(time::sleep(DEFAULT_TIMEOUT)),
             connection_handle,
+            channels: HashMap::new(),
+            global_data,
         }
     }
 
@@ -58,10 +84,7 @@ impl Connection {
 
         info!("Connection is ready for usage!");
 
-        loop {
-            let method = self.recv_method().await?;
-            debug!(?method, "Received method");
-        }
+        self.main_loop().await
     }
 
     async fn send_method(&mut self, channel: u16, method: classes::Class) -> Result<()> {
@@ -146,6 +169,7 @@ impl Connection {
             self.channel_max = channel_max;
             self.max_frame_size = usize::try_from(frame_max).unwrap();
             self.heartbeat_delay = heartbeat;
+            self.reset_timeout();
         }
 
         Ok(())
@@ -170,6 +194,103 @@ impl Connection {
         Ok(())
     }
 
+    async fn main_loop(&mut self) -> Result<()> {
+        loop {
+            tokio::select! {
+                frame = frame::read_frame(&mut self.stream, self.max_frame_size) => {
+                    debug!(?frame);
+                    let frame = frame?;
+                    self.reset_timeout();
+
+                    match frame.kind {
+                        FrameType::Method => self.dispatch_method(frame).await?,
+                        FrameType::Heartbeat => {}
+                        _ => warn!(frame_type = ?frame.kind, "TODO"),
+                    }
+                }
+                _ = &mut self.next_timeout => {
+                    if self.heartbeat_delay != 0 {
+                        return Err(ProtocolError::CloseNow.into());
+                    }
+                }
+            }
+        }
+    }
+
+    async fn dispatch_method(&mut self, frame: Frame) -> Result<()> {
+        let method = classes::parse_method(&frame.payload)?;
+        debug!(?method, "Received method");
+
+        match method {
+            classes::Class::Connection(classes::Connection::Close { .. }) => {
+                // todo: handle closing
+            }
+            classes::Class::Channel(classes::Channel::Open { .. }) => {
+                self.channel_open(frame.channel).await?
+            }
+
+            _ => {
+                // we don't handle this here, forward it to *somewhere*
+            }
+        }
+
+        Ok(())
+    }
+
+    async fn channel_open(&mut self, num: u16) -> Result<()> {
+        let id = Uuid::from_bytes(rand::random());
+        let channel_handle = amqp_core::Channel::new_handle(
+            id,
+            num,
+            self.connection_handle.clone(),
+            self.global_data.clone(),
+        );
+
+        let channel = Channel {
+            num,
+            channel_handle: channel_handle.clone(),
+        };
+
+        let prev = self.channels.insert(num, channel);
+        if let Some(prev) = prev {
+            self.channels.insert(num, prev); // restore previous state
+            return Err(ConException::ChannelError.into_trans());
+        }
+
+        {
+            let mut global_data = self.global_data.lock();
+            global_data.channels.insert(id, channel_handle.clone());
+            global_data
+                .connections
+                .get_mut(&self.id)
+                .unwrap()
+                .lock()
+                .channels
+                .insert(num, channel_handle);
+        }
+
+        info!(%num, "Opened new channel");
+
+        self.send_method(
+            num,
+            Class::Channel(classes::Channel::OpenOk {
+                reserved_1: Vec::new(),
+            }),
+        )
+        .await?;
+
+        time::sleep(Duration::from_secs(1000)).await; // for debugging the dashboard
+
+        Ok(())
+    }
+
+    fn reset_timeout(&mut self) {
+        if self.heartbeat_delay != 0 {
+            let next = Duration::from_secs(u64::from(self.heartbeat_delay));
+            self.next_timeout = Box::pin(time::sleep(next));
+        }
+    }
+
     async fn negotiate_version(&mut self) -> Result<()> {
         const HEADER_SIZE: usize = 8;
         const SUPPORTED_PROTOCOL_VERSION: &[u8] = &[0, 9, 1];

amqp_transport/src/lib.rs

@@ -31,12 +31,12 @@ pub async fn do_thing_i_guess(global_data: GlobalData) -> Result<()> {
         let connection_handle =
             amqp_core::Connection::new_handle(id, peer_addr, global_data.clone());
 
-        let mut global_data = global_data.lock();
-        global_data
+        let mut global_data_guard = global_data.lock();
+        global_data_guard
             .connections
             .insert(id, connection_handle.clone());
 
-        let connection = Connection::new(stream, connection_handle);
+        let connection = Connection::new(id, stream, connection_handle, global_data.clone());
 
         tokio::spawn(connection.start_connection_processing().instrument(span));
     }

amqp_transport/src/sasl.rs

@@ -1,4 +1,4 @@
-//! Partial implementation of the SASL Authentication (see [RFC 4422](https://datatracker.ietf.org/doc/html/rfc4422))
+//! (Very) partial implementation of SASL Authentication (see [RFC 4422](https://datatracker.ietf.org/doc/html/rfc4422))
 //!
 //! Currently only supports PLAN (see [RFC 4616](https://datatracker.ietf.org/doc/html/rfc4616))