send message!

amqp_transport/src/connection.rs

@@ -2,6 +2,7 @@ use std::cmp::Ordering;
 use std::collections::HashMap;
 use std::net::SocketAddr;
 use std::pin::Pin;
+use std::sync::Arc;
 use std::time::Duration;
 
 use anyhow::Context;
@@ -13,6 +14,7 @@ use tokio::time;
 use tracing::{debug, error, info, warn};
 use uuid::Uuid;
 
+use amqp_core::message::{RawMessage, RoutingInformation};
 use amqp_core::methods::{FieldValue, Method, Table};
 use amqp_core::GlobalData;
 
@@ -33,6 +35,8 @@ const CHANNEL_MAX: u16 = 0;
 const FRAME_SIZE_MAX: u32 = 0;
 const HEARTBEAT_DELAY: u16 = 0;
 
+const BASIC_CLASS_ID: u16 = 60;
+
 pub struct Channel {
     /// A handle to the global channel representation. Used to remove the channel when it's dropped
     handle: amqp_core::ChannelHandle,
@@ -254,15 +258,12 @@ impl Connection {
             }
             Method::ChannelOpen { .. } => self.channel_open(frame.channel).await?,
             Method::ChannelClose { .. } => self.channel_close(frame.channel, method).await?,
-            Method::BasicPublish { .. } => {
-                const BASIC_CLASS_ID: u16 = 60;
-                match self.channels.get_mut(&frame.channel) {
-                    Some(channel) => {
-                        channel.status = ChannelStatus::NeedHeader(BASIC_CLASS_ID, Box::new(method))
-                    }
-                    None => return Err(ConException::Todo.into_trans()),
+            Method::BasicPublish { .. } => match self.channels.get_mut(&frame.channel) {
+                Some(channel) => {
+                    channel.status = ChannelStatus::NeedHeader(BASIC_CLASS_ID, Box::new(method))
                 }
-            }
+                None => return Err(ConException::Todo.into_trans()),
+            },
             _ => {
                 let channel_handle = self
                     .channels
@@ -305,33 +306,84 @@ impl Connection {
     }
 
     fn dispatch_body(&mut self, frame: Frame) -> Result<()> {
-        self.channels
+        let channel = self
+            .channels
             .get_mut(&frame.channel)
-            .ok_or_else(|| ConException::Todo.into_trans())
-            .and_then(|channel| match channel.status.take() {
-                ChannelStatus::Default => {
-                    warn!(channel = %frame.channel, "unexpected body");
-                    Err(ConException::UnexpectedFrame.into_trans())
-                }
-                ChannelStatus::NeedHeader(_, _) => {
-                    warn!(channel = %frame.channel, "unexpected body");
-                    Err(ConException::UnexpectedFrame.into_trans())
-                }
-                ChannelStatus::NeedsBody(_, header, mut 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
+            .ok_or_else(|| ConException::Todo.into_trans())?;
+
+        match channel.status.take() {
+            ChannelStatus::Default => {
+                warn!(channel = %frame.channel, "unexpected body");
+                Err(ConException::UnexpectedFrame.into_trans())
+            }
+            ChannelStatus::NeedHeader(_, _) => {
+                warn!(channel = %frame.channel, "unexpected body");
+                Err(ConException::UnexpectedFrame.into_trans())
+            }
+            ChannelStatus::NeedsBody(method, header, mut vec) => {
+                vec.push(frame.payload);
+                match vec
+                    .iter()
+                    .map(Bytes::len)
+                    .sum::<usize>()
+                    .cmp(&usize::try_from(header.body_size).unwrap())
+                {
+                    Ordering::Equal => {
+                        self.process_method_with_body(*method, *header, vec, frame.channel)
                     }
-                    Ok(())
+                    Ordering::Greater => Err(ConException::Todo.into_trans()),
+                    Ordering::Less => Ok(()), // wait for next body
                 }
-            })
+            }
+        }
+    }
+
+    fn process_method_with_body(
+        &mut self,
+        method: Method,
+        header: ContentHeader,
+        payloads: SmallVec<[Bytes; 1]>,
+        channel: ChannelId,
+    ) -> Result<()> {
+        // The only method with content that is sent to the server is Basic.Publish.
+        ensure_conn(header.class_id == BASIC_CLASS_ID)?;
+
+        if let Method::BasicPublish {
+            exchange,
+            routing_key,
+            mandatory,
+            immediate,
+            ..
+        } = method
+        {
+            let message = RawMessage {
+                id: Uuid::from_bytes(rand::random()),
+                properties: header.property_fields,
+                routing: RoutingInformation {
+                    exchange,
+                    routing_key,
+                    mandatory,
+                    immediate,
+                },
+                content: payloads,
+            };
+            let message = Arc::new(message);
+
+            let channel = self
+                .channels
+                .get(&channel)
+                .ok_or_else(|| ConException::Todo.into_trans())?;
+
+            // Spawn the handler for the publish. The connection task goes back to handling
+            // just the connection.
+            tokio::spawn(amqp_messaging::methods::handle_basic_publish(
+                channel.handle.clone(),
+                message,
+            ));
+            Ok(())
+        } else {
+            Err(ConException::Todo.into_trans())
+        }
     }
 
     async fn channel_open(&mut self, channel_id: ChannelId) -> Result<()> {

amqp_transport/src/frame.rs

@@ -56,42 +56,66 @@ pub enum FrameType {
     Heartbeat = 8,
 }
 
-#[derive(Debug, Clone, PartialEq)]
-pub struct BasicProperties {
-    content_type: Option<methods::Shortstr>,
-    content_encoding: Option<methods::Shortstr>,
-    headers: Option<methods::Table>,
-    delivery_mode: Option<methods::Octet>,
-    priority: Option<methods::Octet>,
-    correlation_id: Option<methods::Shortstr>,
-    reply_to: Option<methods::Shortstr>,
-    expiration: Option<methods::Shortstr>,
-    message_id: Option<methods::Shortstr>,
-    timestamp: Option<methods::Timestamp>,
-    r#type: Option<methods::Shortstr>,
-    user_id: Option<methods::Shortstr>,
-    app_id: Option<methods::Shortstr>,
-    reserved: Option<methods::Shortstr>,
-}
-
 #[derive(Debug, Clone, PartialEq)]
 pub struct ContentHeader {
     pub class_id: u16,
     pub weight: u16,
     pub body_size: u64,
-    pub property_fields: BasicProperties,
+    pub property_fields: methods::Table,
 }
 
 mod content_header_parse {
     use crate::error::TransError;
-    use crate::frame::{BasicProperties, ContentHeader};
+    use crate::frame::ContentHeader;
+    use crate::methods::parse_helper::{octet, shortstr, table, timestamp};
+    use amqp_core::methods;
+    use amqp_core::methods::FieldValue::{FieldTable, ShortShortUInt, ShortString, Timestamp};
     use nom::number::complete::{u16, u64};
     use nom::number::Endianness::Big;
 
     type IResult<'a, T> = nom::IResult<&'a [u8], T, TransError>;
 
-    pub fn basic_properties(_property_flags: u16, _input: &[u8]) -> IResult<'_, BasicProperties> {
-        todo!()
+    pub fn basic_properties(flags: u16, input: &[u8]) -> IResult<'_, methods::Table> {
+        macro_rules! parse_property {
+            (if $flags:ident >> $n:literal, $parser:ident($input:ident)?, $map:ident.insert($name:expr, $ctor:path)) => {
+                if (($flags >> $n) & 1) == 1 {
+                    let (input, value) = $parser($input)?;
+                    $map.insert(String::from($name), $ctor(value));
+                    input
+                } else {
+                    $input
+                }
+            };
+        }
+
+        let mut map = methods::Table::new();
+
+        let input = parse_property!(if flags >> 15, shortstr(input)?, map.insert("content-type", ShortString));
+        let input = parse_property!(if flags >> 14, shortstr(input)?, map.insert("content-encoding", ShortString));
+        let input =
+            parse_property!(if flags >> 13, table(input)?, map.insert("headers", FieldTable));
+        let input = parse_property!(if flags >> 12, octet(input)?, map.insert("delivery-mode", ShortShortUInt));
+        let input =
+            parse_property!(if flags >> 11, octet(input)?, map.insert("priority", ShortShortUInt));
+        let input = parse_property!(if flags >> 10, shortstr(input)?, map.insert("correlation-id", ShortString));
+        let input =
+            parse_property!(if flags >> 9, shortstr(input)?, map.insert("reply-to", ShortString));
+        let input =
+            parse_property!(if flags >> 8, shortstr(input)?, map.insert("expiration", ShortString));
+        let input =
+            parse_property!(if flags >> 7, shortstr(input)?, map.insert("message-id", ShortString));
+        let input =
+            parse_property!(if flags >> 6, timestamp(input)?, map.insert("timestamp", Timestamp));
+        let input =
+            parse_property!(if flags >> 5, shortstr(input)?, map.insert("type", ShortString));
+        let input =
+            parse_property!(if flags >> 4, shortstr(input)?, map.insert("user-id", ShortString));
+        let input =
+            parse_property!(if flags >> 3, shortstr(input)?, map.insert("app-id", ShortString));
+        let input =
+            parse_property!(if flags >> 2, shortstr(input)?, map.insert("reserved", ShortString));
+
+        Ok((input, map))
     }
 
     pub fn header(input: &[u8]) -> IResult<'_, Box<ContentHeader>> {
@@ -101,6 +125,7 @@ mod content_header_parse {
 
         // I do not quite understand this here. Apparently, there can be more than 15 flags?
         // But the Basic class only specifies 15, so idk. Don't care about this for now
+        // Todo: But probably later.
         let (input, property_flags) = u16(Big)(input)?;
         let (input, property_fields) = basic_properties(property_flags, input)?;
 

amqp_transport/src/methods/mod.rs

@@ -4,7 +4,7 @@ use rand::Rng;
 use std::collections::HashMap;
 
 mod generated;
-mod parse_helper;
+pub mod parse_helper;
 #[cfg(test)]
 mod tests;
 mod write_helper;