Merge remote-tracking branch 'origin/main'

# Conflicts:
#	Cargo.lock
#	amqp_transport/src/connection.rs

amqp_transport/Cargo.toml

@@ -9,10 +9,12 @@ edition = "2021"
 amqp_core = { path = "../amqp_core" }
 amqp_messaging = { path = "../amqp_messaging" }
 anyhow = "1.0.53"
+bytes = "1.1.0"
 nom = "7.1.0"
 once_cell = "1.9.0"
 rand = "0.8.4"
 regex = "1.5.4"
+smallvec = { version = "1.8.0", features = ["union"] }
 thiserror = "1.0.30"
 tokio = { version = "1.16.1", features = ["full", "tracing"] }
 tracing = "0.1.30"

amqp_transport/src/connection.rs

@@ -1,19 +1,25 @@
-use crate::error::{ConException, ProtocolError, Result};
-use crate::frame::{Frame, FrameType};
-use crate::{frame, methods, sasl};
-use amqp_core::methods::{FieldValue, Method, Table};
-use amqp_core::GlobalData;
-use anyhow::Context;
+use std::cmp::Ordering;
 use std::collections::HashMap;
 use std::net::SocketAddr;
 use std::pin::Pin;
 use std::time::Duration;
+
+use anyhow::Context;
+use bytes::Bytes;
+use smallvec::SmallVec;
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
 use tokio::net::TcpStream;
 use tokio::time;
 use tracing::{debug, error, info, warn};
 use uuid::Uuid;
 
+use amqp_core::methods::{FieldValue, Method, Table};
+use amqp_core::GlobalData;
+
+use crate::error::{ConException, ProtocolError, Result};
+use crate::frame::{ContentHeader, Frame, FrameType};
+use crate::{frame, methods, sasl};
+
 fn ensure_conn(condition: bool) -> Result<()> {
     if condition {
         Ok(())
@@ -47,6 +53,12 @@ pub struct Connection {
 
 const DEFAULT_TIMEOUT: Duration = Duration::from_secs(30);
 
+enum WaitForBodyStatus {
+    Method(Method),
+    Header(Method, ContentHeader, SmallVec<[Bytes; 1]>),
+    None,
+}
+
 impl Connection {
     pub fn new(
         id: Uuid,
@@ -95,7 +107,7 @@ impl Connection {
             &Frame {
                 kind: FrameType::Method,
                 channel,
-                payload,
+                payload: payload.into(),
             },
             &mut self.stream,
         )
@@ -196,15 +208,46 @@ impl Connection {
     }
 
     async fn main_loop(&mut self) -> Result<()> {
+        // todo: find out how header/body frames can interleave between channels
+        let mut wait_for_body = WaitForBodyStatus::None;
+
         loop {
             debug!("Waiting for next frame");
             let frame = frame::read_frame(&mut self.stream, self.max_frame_size).await?;
             self.reset_timeout();
 
             match frame.kind {
-                FrameType::Method => self.dispatch_method(frame).await?,
+                FrameType::Method => wait_for_body = self.dispatch_method(frame).await?,
                 FrameType::Heartbeat => {}
-                _ => warn!(frame_type = ?frame.kind, "TODO"),
+                FrameType::Header => match wait_for_body {
+                    WaitForBodyStatus::None => warn!(channel = %frame.channel, "unexpected header"),
+                    WaitForBodyStatus::Method(method) => {
+                        wait_for_body =
+                            WaitForBodyStatus::Header(method, ContentHeader::new(), SmallVec::new())
+                    }
+                    WaitForBodyStatus::Header(_, _, _) => {
+                        warn!(channel = %frame.channel, "already got header")
+                    }
+                },
+                FrameType::Body => match &mut wait_for_body {
+                    WaitForBodyStatus::None => warn!(channel = %frame.channel, "unexpected body"),
+                    WaitForBodyStatus::Method(_) => {
+                        warn!(channel = %frame.channel, "unexpected body")
+                    }
+                    WaitForBodyStatus::Header(_, header, vec) => {
+                        vec.push(frame.payload);
+                        match vec
+                            .iter()
+                            .map(Bytes::len)
+                            .sum::<usize>()
+                            .cmp(&usize::try_from(header.body_size).unwrap())
+                        {
+                            Ordering::Equal => todo!("process body"),
+                            Ordering::Greater => todo!("too much data!"),
+                            Ordering::Less => {} // wait for next body
+                        }
+                    }
+                },
             }
         }
     }
@@ -219,6 +262,7 @@ impl Connection {
             }
             Method::ChannelOpen { .. } => self.channel_open(frame.channel).await?,
             Method::ChannelClose { .. } => self.channel_close(frame.channel, method).await?,
+            Method::BasicPublish { .. } => return Ok(WaitForBodyStatus::Method(method)),
             _ => {
                 let channel_handle = self
                     .channels
@@ -235,7 +279,7 @@ impl Connection {
             }
         }
 
-        Ok(())
+        Ok(WaitForBodyStatus::None)
     }
 
     async fn channel_open(&mut self, num: u16) -> Result<()> {

amqp_transport/src/frame.rs

@@ -1,5 +1,8 @@
 use crate::error::{ConException, ProtocolError, Result};
+use amqp_core::methods::FieldValue;
 use anyhow::Context;
+use bytes::Bytes;
+use smallvec::SmallVec;
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
 use tracing::trace;
 
@@ -18,7 +21,7 @@ pub struct Frame {
     pub kind: FrameType,
     pub channel: u16,
     /// Includes the whole payload, also including the metadata from each type.
-    pub payload: Vec<u8>,
+    pub payload: Bytes,
 }
 
 #[derive(Debug, Copy, Clone, PartialEq, Eq)]
@@ -30,6 +33,21 @@ pub enum FrameType {
     Heartbeat = 8,
 }
 
+#[derive(Debug, Clone, PartialEq)]
+pub struct ContentHeader {
+    pub class_id: u16,
+    pub weight: u16,
+    pub body_size: u64,
+    pub property_flags: SmallVec<[u16; 1]>,
+    pub property_fields: Vec<FieldValue>,
+}
+
+impl ContentHeader {
+    pub fn new() -> Self {
+        todo!()
+    }
+}
+
 pub async fn write_frame<W>(frame: &Frame, mut w: W) -> Result<()>
 where
     W: AsyncWriteExt + Unpin,
@@ -72,7 +90,7 @@ where
     let frame = Frame {
         kind,
         channel,
-        payload,
+        payload: payload.into(),
     };
 
     trace!(?frame, "Received frame");
@@ -99,6 +117,7 @@ fn parse_frame_type(kind: u8, channel: u16) -> Result<FrameType> {
 #[cfg(test)]
 mod tests {
     use crate::frame::{Frame, FrameType};
+    use bytes::Bytes;
 
     #[tokio::test]
     async fn read_small_body() {
@@ -127,7 +146,7 @@ mod tests {
             Frame {
                 kind: FrameType::Method,
                 channel: 0,
-                payload: vec![1, 2, 3],
+                payload: Bytes::from_static(&[1, 2, 3]),
             }
         );
     }

amqp_transport/src/sasl.rs

@@ -1,6 +1,6 @@
 //! (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))
+//! Currently only supports PLAIN (see [RFC 4616](https://datatracker.ietf.org/doc/html/rfc4616))
 
 use crate::error::{ConException, Result};
 

amqp_transport/src/tests.rs

@@ -22,7 +22,7 @@ async fn write_start_ok_frame() {
     let frame = frame::Frame {
         kind: FrameType::Method,
         channel: 0,
-        payload,
+        payload: payload.into(),
     };
 
     let mut output = Vec::new();