add exchanges to internal model

Cargo.lock

@@ -442,8 +442,8 @@ dependencies = [
  "haesli_datastructure",
  "parking_lot",
  "rand",
- "smallvec",
  "thiserror",
+ "tinyvec",
  "tokio",
  "uuid",
 ]
@@ -493,8 +493,8 @@ dependencies = [
  "once_cell",
  "rand",
  "regex",
- "smallvec",
  "thiserror",
+ "tinyvec",
  "tokio",
  "tracing",
 ]
@@ -1291,6 +1291,21 @@ dependencies = [
  "serde_json",
 ]
 
+[[package]]
+name = "tinyvec"
+version = "1.5.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2c1c1d5a42b6245520c249549ec267180beaffcc0615401ac8e31853d4b6d8d2"
+dependencies = [
+ "tinyvec_macros",
+]
+
+[[package]]
+name = "tinyvec_macros"
+version = "0.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cda74da7e1a664f795bb1f8a87ec406fb89a02522cf6e50620d016add6dbbf5c"
+
 [[package]]
 name = "tokio"
 version = "1.17.0"

haesli_core/Cargo.toml

@@ -10,7 +10,7 @@ haesli_datastructure = { path = "../haesli_datastructure" }
 bytes = "1.1.0"
 parking_lot = "0.12.0"
 rand = "0.8.5"
-smallvec = { version = "1.8.0", features = ["union"] }
+tinyvec = { version = "1.5.1", features = ["alloc", "rustc_1_55"] }
 thiserror = "1.0.30"
 tokio = { version = "1.17.0", features = ["sync"] }
 uuid = "0.8.2"

haesli_core/src/connection.rs

@@ -7,7 +7,7 @@ use std::{
 
 use bytes::Bytes;
 use parking_lot::Mutex;
-use smallvec::SmallVec;
+use tinyvec::TinyVec;
 use tokio::sync::mpsc;
 
 use crate::{
@@ -67,7 +67,7 @@ pub struct ConnectionInner {
 pub enum ConnectionEvent {
     Shutdown,
     Method(ChannelNum, Box<Method>),
-    MethodContent(ChannelNum, Box<Method>, ContentHeader, SmallVec<[Bytes; 1]>),
+    MethodContent(ChannelNum, Box<Method>, ContentHeader, TinyVec<[Bytes; 1]>),
 }
 
 pub type ConEventSender = mpsc::Sender<ConnectionEvent>;

haesli_core/src/exchange.rs

@@ -1,13 +1,88 @@
+use std::{borrow::Borrow, collections::HashMap, sync::Arc};
+
+use crate::{newtype, Queue};
+
+#[derive(Debug)]
 pub enum ExchangeType {
     /// Routes a message to a queue if the routing-keys are equal
-    Direct,
+    Direct { bindings: HashMap<String, Queue> },
     /// Always routes the message to a queue
-    Fanout,
+    Fanout { bindings: Vec<Queue> },
     /// Routes a message to a queue if the routing key matches the pattern
-    Topic,
+    Topic { bindings: Vec<(String, Queue)> },
     /// Is bound with a table of headers and values, and matches if the message headers
     /// match up with the binding headers
+    ///
+    /// Unsupported for now.
     Headers,
     /// The message is sent to the server system service with the name of the routing-key
+    ///
+    /// Unsupported for now.
     System,
 }
+
+newtype!(
+    /// The name of a queue. A newtype wrapper around `Arc<str>`, which guarantees cheap clones.
+    #[derive(Debug, Clone, PartialEq, Eq, Hash)]
+    pub ExchangeName: Arc<str>
+);
+
+impl Borrow<str> for ExchangeName {
+    fn borrow(&self) -> &str {
+        Borrow::borrow(&self.0)
+    }
+}
+
+#[derive(Debug)]
+pub struct Exchange {
+    pub name: ExchangeName,
+    pub kind: ExchangeType,
+}
+
+#[derive(Debug)]
+pub struct Binding {}
+
+pub fn default_exchanges() -> HashMap<ExchangeName, Exchange> {
+    // 3.1.3 - The spec requires a few default exchanges to exist
+
+    let empty_name = ExchangeName::new("".to_owned().into());
+    let empty = Exchange {
+        name: empty_name.clone(),
+        kind: ExchangeType::Direct {
+            bindings: HashMap::new(),
+        },
+    };
+
+    let direct_name = ExchangeName::new("amqp.direct".to_owned().into());
+    let direct = Exchange {
+        name: direct_name.clone(),
+        kind: ExchangeType::Direct {
+            bindings: HashMap::new(),
+        },
+    };
+
+    let fanout_name = ExchangeName::new("amqp.fanout".to_owned().into());
+    let fanout = Exchange {
+        name: fanout_name.clone(),
+        kind: ExchangeType::Fanout {
+            bindings: Vec::new(),
+        },
+    };
+
+    let topic_name = ExchangeName::new("amqp.topic".to_owned().into());
+    let topic = Exchange {
+        name: topic_name.clone(),
+        kind: ExchangeType::Topic {
+            bindings: Vec::new(),
+        },
+    };
+
+    // we don't implement headers (yet), so don't provide the default exchange for it
+
+    HashMap::from([
+        (empty_name, empty),
+        (direct_name, direct),
+        (fanout_name, fanout),
+        (topic_name, topic),
+    ])
+}

haesli_core/src/lib.rs

@@ -21,10 +21,12 @@ use uuid::Uuid;
 
 use crate::{
     connection::{Channel, Connection},
+    exchange::{Exchange, ExchangeName},
     queue::{Queue, QueueName},
 };
 
 #[derive(Clone)]
+// todo: what if this was downstream?
 pub struct GlobalData {
     inner: Arc<Mutex<GlobalDataInner>>,
 }
@@ -42,6 +44,7 @@ impl Default for GlobalData {
                 connections: HashMap::new(),
                 channels: HashMap::new(),
                 queues: HashMap::new(),
+                exchanges: exchange::default_exchanges(),
                 default_exchange: HashMap::new(),
             })),
         }
@@ -59,6 +62,7 @@ pub struct GlobalDataInner {
     pub connections: HashMap<ConnectionId, Connection>,
     pub channels: HashMap<ChannelId, Channel>,
     pub queues: HashMap<QueueName, Queue>,
+    pub exchanges: HashMap<ExchangeName, Exchange>,
     /// Todo: This is just for testing and will be removed later!
     pub default_exchange: HashMap<String, Queue>,
 }

haesli_core/src/macros.rs

@@ -63,7 +63,7 @@ macro_rules! newtype {
         where
             $ty: std::fmt::Display,
         {
-            fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+            fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
                 std::fmt::Display::fmt(&self.0, f)
             }
         }

haesli_core/src/message.rs

@@ -1,7 +1,7 @@
 use std::sync::Arc;
 
 use bytes::Bytes;
-use smallvec::SmallVec;
+use tinyvec::TinyVec;
 
 use crate::{connection::ContentHeader, newtype_id};
 
@@ -14,7 +14,7 @@ pub struct MessageInner {
     pub id: MessageId,
     pub header: ContentHeader,
     pub routing: RoutingInformation,
-    pub content: SmallVec<[Bytes; 1]>,
+    pub content: TinyVec<[Bytes; 1]>,
 }
 
 #[derive(Debug)]

haesli_core/src/queue.rs

@@ -1,7 +1,7 @@
 use std::{
     borrow::Borrow,
     collections::HashMap,
-    fmt::{Debug, Formatter},
+    fmt::Debug,
     sync::{atomic::AtomicUsize, Arc},
 };
 

haesli_messaging/src/lib.rs

@@ -5,5 +5,6 @@ use haesli_core::error::ProtocolError;
 
 pub mod methods;
 mod queue_worker;
+mod routing;
 
 type Result<T> = std::result::Result<T, ProtocolError>;

haesli_messaging/src/methods/mod.rs

@@ -2,15 +2,14 @@ mod consume;
 mod publish;
 mod queue;
 
-use haesli_core::{amqp_todo, connection::Channel, message::Message, methods::Method};
+use haesli_core::{amqp_todo, connection::Channel, methods::Method};
+pub use publish::publish;
 use tracing::info;
 
 use crate::Result;
 
-pub fn handle_basic_publish(channel_handle: Channel, message: Message) -> Result<()> {
+/// This is the entrypoint of methods not handled by the connection itself.
-    publish::publish(channel_handle, message)
+/// Note that Basic.Publish is *not* sent here, but to [`handle_basic_publish`](crate::handle_basic_publish)
-}
-
 pub async fn handle_method(channel_handle: Channel, method: Method) -> Result<Method> {
     info!(?method, "Handling method");
 

haesli_messaging/src/methods/publish.rs

@@ -7,7 +7,7 @@ use haesli_core::{
 };
 use tracing::{debug, error};
 
-use crate::Result;
+use crate::{routing, Result};
 
 pub fn publish(channel_handle: Channel, message: Message) -> Result<()> {
     debug!(?message, "Publishing message");
@@ -22,9 +22,14 @@ pub fn publish(channel_handle: Channel, message: Message) -> Result<()> {
 
     let global_data = global_data.lock();
 
-    let queue = global_data
+    let exchange = &message.routing.exchange;
-        .queues
+
-        .get(routing.routing_key.as_str())
+    let exchange = global_data
+        .exchanges
+        .get(exchange.as_str())
+        .ok_or(ChannelException::NotFound)?;
+
+    let queue = routing::route_message(exchange, &message.routing.routing_key)
         .ok_or(ChannelException::NotFound)?;
 
     queue

haesli_messaging/src/methods/queue.rs

@@ -11,7 +11,7 @@ use parking_lot::Mutex;
 use tokio::sync::mpsc;
 use tracing::debug;
 
-use crate::{queue_worker::QueueTask, Result};
+use crate::{queue_worker::QueueTask, routing, Result};
 
 pub fn declare(channel: Channel, queue_declare: QueueDeclare) -> Result<Method> {
     let QueueDeclare {
@@ -75,7 +75,7 @@ pub fn declare(channel: Channel, queue_declare: QueueDeclare) -> Result<Method>
             .or_insert_with(|| queue.clone());
     }
 
-    bind_queue(global_data.clone(), (), queue_name.clone().into_inner())?;
+    bind_queue(global_data.clone(), (), queue_name.to_string())?;
 
     let queue_task = QueueTask::new(global_data, event_recv, queue);
 
@@ -92,18 +92,22 @@ pub async fn bind(_channel_handle: Channel, _queue_bind: QueueBind) -> Result<Me
     amqp_todo!();
 }
 
-fn bind_queue(global_data: GlobalData, _exchange: (), routing_key: Arc<str>) -> Result<()> {
+fn bind_queue(global_data: GlobalData, _exchange: (), routing_key: String) -> Result<()> {
     let mut global_data = global_data.lock();
 
     // todo: don't
     let queue = global_data
         .queues
-        .get(&QueueName::new(routing_key.clone()))
+        .get(&QueueName::new(routing_key.clone().into()))
         .unwrap()
         .clone();
-    global_data
+
-        .default_exchange
+    let exchange = global_data
-        .insert(routing_key.to_string(), queue);
+        .exchanges
+        .get_mut("")
+        .expect("default empty exchange");
+
+    routing::bind(exchange, routing_key, queue);
 
     Ok(())
 }

haesli_messaging/src/routing.rs

@@ -0,0 +1,30 @@
+use haesli_core::{
+    exchange::{Exchange, ExchangeType},
+    queue::Queue,
+};
+
+pub fn bind(exchange: &mut Exchange, routing_key: String, queue: Queue) {
+    match &mut exchange.kind {
+        ExchangeType::Direct { bindings } => {
+            bindings.insert(routing_key, queue);
+        }
+        ExchangeType::Fanout { bindings } => bindings.push(queue),
+        ExchangeType::Topic { bindings } => bindings.push((routing_key, queue)),
+        ExchangeType::Headers => {} // unsupported
+        ExchangeType::System => {}  // unsupported
+    }
+}
+
+/// Route a message to a queue. Returns the queue to send it to, or `None` if it can't be matched
+pub fn route_message(exchange: &Exchange, routing_key: &str) -> Option<Queue> {
+    match &exchange.kind {
+        ExchangeType::Direct { bindings } => {
+            // 3.1.3.1 - routing-key = routing-key
+            bindings.get(routing_key).cloned()
+        }
+        ExchangeType::Fanout { .. } => None,
+        ExchangeType::Topic { .. } => None,
+        ExchangeType::Headers => None, // unsupported
+        ExchangeType::System => None,  // unsupported
+    }
+}

haesli_transport/Cargo.toml

@@ -14,7 +14,7 @@ nom = "7.1.0"
 once_cell = "1.9.0"
 rand = "0.8.4"
 regex = "1.5.4"
-smallvec = { version = "1.8.0", features = ["union"] }
+tinyvec = { version = "1.5.1", features = ["alloc", "rustc_1_55"] }
 thiserror = "1.0.30"
 tokio = { version = "1.16.1", features = ["full"] }
 tracing = "0.1.30"

haesli_transport/src/connection.rs

@@ -17,7 +17,7 @@ use haesli_core::{
     },
     GlobalData,
 };
-use smallvec::SmallVec;
+use tinyvec::TinyVec;
 use tokio::{
     io::{AsyncReadExt, AsyncWriteExt},
     net::TcpStream,
@@ -75,7 +75,7 @@ const DEFAULT_TIMEOUT: Duration = Duration::from_secs(30);
 enum ChannelStatus {
     Default,
     NeedHeader(u16, Box<Method>),
-    NeedsBody(Box<Method>, ContentHeader, SmallVec<[Bytes; 1]>),
+    NeedsBody(Box<Method>, ContentHeader, TinyVec<[Bytes; 1]>),
 }
 
 impl ChannelStatus {
@@ -166,7 +166,7 @@ impl TransportConnection {
         channel: ChannelNum,
         method: &Method,
         header: ContentHeader,
-        body: &SmallVec<[Bytes; 1]>,
+        body: &TinyVec<[Bytes; 1]>,
     ) -> Result<()> {
         self.send_method(channel, method).await?;
 
@@ -177,11 +177,7 @@ impl TransportConnection {
         self.send_bodies(channel, body).await
     }
 
-    async fn send_bodies(
+    async fn send_bodies(&mut self, channel: ChannelNum, body: &TinyVec<[Bytes; 1]>) -> Result<()> {
-        &mut self,
-        channel: ChannelNum,
-        body: &SmallVec<[Bytes; 1]>,
-    ) -> Result<()> {
         // this is inefficient if it's a huge message sent by a client with big frames to one with
         // small frames
         // we assume that this won't happen that that the first branch will be taken in most cases,
@@ -442,7 +438,7 @@ impl TransportConnection {
                     let header = parse_content_header(&frame.payload)?;
                     ensure_conn(header.class_id == class_id)?;
 
-                    channel.status = ChannelStatus::NeedsBody(method, header, SmallVec::new());
+                    channel.status = ChannelStatus::NeedsBody(method, header, TinyVec::new());
                     Ok(())
                 }
                 ChannelStatus::NeedsBody(_, _, _) => {
@@ -489,7 +485,7 @@ impl TransportConnection {
         &mut self,
         method: Method,
         header: ContentHeader,
-        payloads: SmallVec<[Bytes; 1]>,
+        payloads: TinyVec<[Bytes; 1]>,
         channel: ChannelNum,
     ) -> Result<()> {
         // The only method with content that is sent to the server is Basic.Publish.
@@ -518,7 +514,7 @@ impl TransportConnection {
 
             let channel = self.channels.get(&channel).ok_or(ConException::Todo)?;
 
-            haesli_messaging::methods::handle_basic_publish(channel.global_chan.clone(), message)?;
+            haesli_messaging::methods::publish(channel.global_chan.clone(), message)?;
             Ok(())
         } else {
             Err(ConException::Todo.into())