more queue stuff

Cargo.lock

@@ -87,9 +87,9 @@ dependencies = [
 
 [[package]]
 name = "anyhow"
-version = "1.0.54"
+version = "1.0.55"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7a99269dff3bc004caa411f38845c20303f1e393ca2bd6581576fa3a7f59577d"
+checksum = "159bb86af3a200e19a068f4224eae4c8bb2d0fa054c7e5d1cacd5cef95e684cd"
 
 [[package]]
 name = "async-trait"

amqp_core/src/error.rs

@@ -25,8 +25,8 @@ pub enum ConException {
     ChannelError,
     #[error("505 Unexpected Frame")]
     UnexpectedFrame,
-    #[error("540 Not implemented")]
-    NotImplemented,
+    #[error("540 Not implemented. '{0}'")]
+    NotImplemented(&'static str),
     #[error("xxx Not decided yet")]
     Todo,
 }

amqp_core/src/lib.rs

@@ -8,7 +8,7 @@ pub mod methods;
 pub mod queue;
 
 use crate::connection::{ChannelHandle, ConnectionHandle};
-use crate::queue::{Queue, QueueId};
+use crate::queue::{Queue, QueueName};
 use connection::{ChannelId, ConnectionId};
 use parking_lot::Mutex;
 use std::collections::HashMap;
@@ -44,7 +44,7 @@ impl GlobalData {
 pub struct GlobalDataInner {
     pub connections: HashMap<ConnectionId, ConnectionHandle>,
     pub channels: HashMap<ChannelId, ChannelHandle>,
-    pub queues: HashMap<QueueId, Queue>,
+    pub queues: HashMap<QueueName, Queue>,
     /// Todo: This is just for testing and will be removed later!
     pub default_exchange: HashMap<String, Queue>,
 }

amqp_core/src/macros.rs

@@ -48,12 +48,23 @@ macro_rules! newtype {
                 &self.0
             }
         }
+
+        impl<T> std::convert::From<T> for $name
+        where
+            $ty: From<T>,
+        {
+            fn from(other: T) -> Self {
+                Self(other.into())
+            }
+        }
     };
 }
 
 #[macro_export]
 macro_rules! amqp_todo {
     () => {
-        return Err(::amqp_core::error::ConException::NotImplemented.into())
+        return Err(
+            ::amqp_core::error::ConException::NotImplemented(concat!(file!(), ":", line!())).into(),
+        )
     };
 }

amqp_core/src/queue.rs

@@ -1,5 +1,5 @@
 use crate::message::Message;
-use crate::{newtype_id, ChannelId};
+use crate::{newtype, newtype_id, ChannelId};
 use parking_lot::Mutex;
 use std::sync::atomic::AtomicUsize;
 use std::sync::Arc;
@@ -8,10 +8,16 @@ pub type Queue = Arc<RawQueue>;
 
 newtype_id!(pub QueueId);
 
+newtype!(
+    /// The name of a queue. A newtype wrapper around `Arc<str>`, which guarantees cheap clones.
+    #[derive(Debug, Clone, PartialEq, Eq, Hash)]
+    pub QueueName: Arc<str>
+);
+
 #[derive(Debug)]
 pub struct RawQueue {
     pub id: QueueId,
-    pub name: String,
+    pub name: QueueName,
     pub messages: Mutex<Vec<Message>>, // use a concurrent linked list???
     pub durable: bool,
     pub exclusive: Option<ChannelId>,

amqp_dashboard/src/lib.rs

@@ -51,6 +51,7 @@ async fn get_style_css() -> Response {
 #[derive(Serialize)]
 struct Data {
     connections: Vec<Connection>,
+    queues: Vec<Queue>,
 }
 
 #[derive(Serialize)]
@@ -66,6 +67,13 @@ struct Channel {
     number: u16,
 }
 
+#[derive(Serialize)]
+struct Queue {
+    id: String,
+    name: String,
+    durable: bool,
+}
+
 async fn get_data(global_data: GlobalData) -> impl IntoResponse {
     let global_data = global_data.lock();
 
@@ -92,7 +100,20 @@ async fn get_data(global_data: GlobalData) -> impl IntoResponse {
         })
         .collect();
 
-    let data = Data { connections };
+    let queues = global_data
+        .queues
+        .values()
+        .map(|queue| Queue {
+            id: queue.id.to_string(),
+            name: queue.name.to_string(),
+            durable: queue.durable,
+        })
+        .collect();
+
+    let data = Data {
+        connections,
+        queues,
+    };
 
     Json(data)
 }

amqp_messaging/src/methods/consume.rs

@@ -1,10 +1,11 @@
+use amqp_core::amqp_todo;
 use amqp_core::connection::ChannelHandle;
 use amqp_core::error::ProtocolError;
-use amqp_core::methods::BasicConsume;
+use amqp_core::methods::{BasicConsume, Method};
 
 pub async fn consume(
     _channel_handle: ChannelHandle,
     _basic_consume: BasicConsume,
-) -> Result<(), ProtocolError> {
-    Ok(())
+) -> Result<Method, ProtocolError> {
+    amqp_todo!()
 }

amqp_messaging/src/methods/mod.rs

@@ -18,17 +18,19 @@ pub async fn handle_basic_publish(_channel_handle: ChannelHandle, message: Messa
 pub async fn handle_method(
     channel_handle: ChannelHandle,
     method: Method,
-) -> Result<(), ProtocolError> {
+) -> Result<Method, ProtocolError> {
     info!(?method, "Handling method");
 
-    match method {
+    let response = match method {
         Method::ExchangeDeclare(_) => amqp_todo!(),
         Method::ExchangeDeclareOk(_) => amqp_todo!(),
         Method::ExchangeDelete(_) => amqp_todo!(),
         Method::ExchangeDeleteOk(_) => amqp_todo!(),
-        Method::QueueDeclare(queue_declare) => queue::declare(channel_handle, queue_declare).await,
+        Method::QueueDeclare(queue_declare) => {
+            queue::declare(channel_handle, queue_declare).await?
+        }
         Method::QueueDeclareOk { .. } => amqp_todo!(),
-        Method::QueueBind(queue_bind) => queue::bind(channel_handle, queue_bind).await,
+        Method::QueueBind(queue_bind) => queue::bind(channel_handle, queue_bind).await?,
         Method::QueueBindOk(_) => amqp_todo!(),
         Method::QueueUnbind { .. } => amqp_todo!(),
         Method::QueueUnbindOk(_) => amqp_todo!(),
@@ -38,7 +40,7 @@ pub async fn handle_method(
         Method::QueueDeleteOk { .. } => amqp_todo!(),
         Method::BasicQos { .. } => amqp_todo!(),
         Method::BasicQosOk(_) => amqp_todo!(),
-        Method::BasicConsume(consume) => consume::consume(channel_handle, consume).await,
+        Method::BasicConsume(consume) => consume::consume(channel_handle, consume).await?,
         Method::BasicConsumeOk { .. } => amqp_todo!(),
         Method::BasicCancel { .. } => amqp_todo!(),
         Method::BasicCancelOk { .. } => amqp_todo!(),
@@ -62,5 +64,7 @@ pub async fn handle_method(
             unreachable!("Basic.Publish is handled somewhere else because it has a body")
         }
         _ => unreachable!("Method handled by transport layer"),
-    }
+    };
+
+    Ok(response)
 }

amqp_messaging/src/methods/queue.rs

@@ -1,7 +1,7 @@
 use amqp_core::connection::ChannelHandle;
 use amqp_core::error::{ConException, ProtocolError};
-use amqp_core::methods::{QueueBind, QueueDeclare};
-use amqp_core::queue::{QueueDeletion, QueueId, RawQueue};
+use amqp_core::methods::{Method, QueueBind, QueueDeclare, QueueDeclareOk};
+use amqp_core::queue::{QueueDeletion, QueueId, QueueName, RawQueue};
 use amqp_core::{amqp_todo, GlobalData};
 use parking_lot::Mutex;
 use std::sync::atomic::AtomicUsize;
@@ -10,7 +10,7 @@ use std::sync::Arc;
 pub async fn declare(
     channel_handle: ChannelHandle,
     queue_declare: QueueDeclare,
-) -> Result<(), ProtocolError> {
+) -> Result<Method, ProtocolError> {
     let QueueDeclare {
         queue: queue_name,
         passive,
@@ -22,12 +22,15 @@ pub async fn declare(
         ..
     } = queue_declare;
 
+    let queue_name = QueueName::new(queue_name.into());
+
     if !arguments.is_empty() {
         return Err(ConException::Todo.into());
     }
 
-    let (global_data, id) = {
+    let global_data = {
         let channel = channel_handle.lock();
+        let global_data = channel.global_data.clone();
 
         if passive || no_wait {
             amqp_todo!();
@@ -47,31 +50,46 @@ pub async fn declare(
             },
         });
 
-        let global_data = channel.global_data.clone();
-
         {
             let mut global_data_lock = global_data.lock();
-            global_data_lock.queues.insert(id, queue);
+            global_data_lock.queues.insert(queue_name.clone(), queue);
         }
 
-        (global_data, id)
+        global_data
     };
 
-    bind_queue(global_data, id, (), queue_name).await
+    bind_queue(global_data, (), queue_name.clone().into_inner()).await?;
+
+    Ok(Method::QueueDeclareOk(QueueDeclareOk {
+        queue: queue_name.to_string(),
+        message_count: 0,
+        consumer_count: 0,
+    }))
 }
 
 pub async fn bind(
     _channel_handle: ChannelHandle,
     _queue_bind: QueueBind,
-) -> Result<(), ProtocolError> {
+) -> Result<Method, ProtocolError> {
     amqp_todo!();
 }
 
 async fn bind_queue(
-    _global_data: GlobalData,
-    _queue: QueueId,
+    global_data: GlobalData,
     _exchange: (),
-    _routing_key: String,
+    routing_key: Arc<str>,
 ) -> Result<(), ProtocolError> {
-    amqp_todo!();
+    let mut global_data = global_data.lock();
+
+    // todo: don't
+    let queue = global_data
+        .queues
+        .get(&QueueName::new(routing_key.clone()))
+        .unwrap()
+        .clone();
+    global_data
+        .default_exchange
+        .insert(routing_key.to_string(), queue);
+
+    Ok(())
 }

amqp_transport/src/connection.rs

@@ -223,7 +223,6 @@ impl Connection {
 
     async fn main_loop(&mut self) -> Result<()> {
         loop {
-            debug!("Waiting for next frame");
             let frame = frame::read_frame(&mut self.stream, self.max_frame_size).await?;
             self.reset_timeout();
 
@@ -277,9 +276,11 @@ impl Connection {
                     .clone();
 
                 // call into amqp_messaging to handle the method
-                // amqp_messaging then branches and spawns a new task for longer running things,
-                // so the connection task will only be "blocked" for a short amount of time
+                // it returns the response method that we are supposed to send
+                // maybe this might become an `Option` in the future
+                let return_method =
                     amqp_messaging::methods::handle_method(channel_handle, method).await?;
+                self.send_method(frame.channel, return_method).await?;
             }
         }
         Ok(())

amqp_transport/src/error.rs

@@ -1,10 +1,8 @@
-#![allow(dead_code)]
-
 use std::io::Error;
 
 pub use amqp_core::error::{ConException, ProtocolError};
 
-pub type StdResult<T, E> = std::result::Result<T, E>;
+type StdResult<T, E> = std::result::Result<T, E>;
 
 pub type Result<T> = StdResult<T, TransError>;
 

test-js/src/declare-queue.js

@@ -0,0 +1,18 @@
+import { assert, connectAmqp } from './utils/utils.js';
+
+const queueName = 'test-queue-124';
+
+const connection = await connectAmqp();
+
+const channel = await connection.createChannel();
+
+const reply = await channel.assertQueue(queueName, { durable: true });
+
+assert(reply.messageCount === 0, 'Message found in queue');
+assert(reply.consumerCount === 0, 'Consumer listening on queue');
+assert(reply.queue === queueName, 'Wrong queue name returned');
+
+console.log(`created queue '${queueName}'`);
+
+await channel.close();
+await connection.close();

test-js/src/utils/utils.js

@@ -16,3 +16,9 @@ export const connectAmqp = async () => {
     {}
   );
 };
+
+export const assert = (cond, msg) => {
+  if (!cond) {
+    throw new Error(`Assertion failed: ${msg}`);
+  }
+};

xtask/src/codegen/parser.rs

@@ -21,8 +21,8 @@ impl Codegen {
             self.output,
             "pub mod parse {{
 use amqp_core::methods::*;
-use crate::methods::parse_helper::*;
 use crate::error::TransError;
+use crate::methods::parse_helper::*;
 use nom::{{branch::alt, bytes::complete::tag}};
 use regex::Regex;
 use once_cell::sync::Lazy;

xtask/src/codegen/write.rs

@@ -7,8 +7,8 @@ impl Codegen {
             self.output,
             "pub mod write {{
 use amqp_core::methods::*;
-use crate::methods::write_helper::*;
 use crate::error::TransError;
+use crate::methods::write_helper::*;
 use std::io::Write;
 
 pub fn write_method<W: Write>(method: Method, mut writer: W) -> Result<(), TransError> {{