big future

amqp_transport/src/classes/mod.rs

@@ -1,4 +1,3 @@
-use crate::classes::generated::Class;
 use crate::error::{ConException, ProtocolError, TransError};
 use std::collections::HashMap;
 
@@ -35,7 +34,7 @@ pub enum FieldValue {
 pub use generated::*;
 
 /// Parses the payload of a method frame into the class/method
-pub fn parse_method(payload: &[u8]) -> Result<Class, TransError> {
+pub fn parse_method(payload: &[u8]) -> Result<generated::Class, TransError> {
     let nom_result = generated::parse::parse_method(payload);
 
     match nom_result {

amqp_transport/src/classes/write_helper.rs

@@ -1,33 +1,32 @@
-use crate::classes::generated::{
-    Bit, Long, Longlong, Longstr, Octet, Short, Shortstr, Table, Timestamp,
-};
-use crate::classes::{FieldValue, TableFieldName};
-use crate::error::TransError;
+use crate::classes::FieldValue;
+use crate::classes::{Bit, Long, Longlong, Longstr, Octet, Short, Shortstr, Table, Timestamp};
+use crate::error::Result;
 use anyhow::Context;
-use std::io;
-use std::io::Write;
+use std::future::Future;
+use std::pin::Pin;
+use tokio::io::AsyncWriteExt;
 
-pub fn octet<W: Write>(value: Octet, writer: &mut W) -> Result<(), TransError> {
-    writer.write_all(&[value])?;
+pub async fn octet<W: AsyncWriteExt + Unpin>(value: Octet, writer: &mut W) -> Result<()> {
+    writer.write_u8(value).await?;
     Ok(())
 }
 
-pub fn short<W: Write>(value: Short, writer: &mut W) -> Result<(), TransError> {
-    writer.write_all(&value.to_be_bytes())?;
+pub async fn short<W: AsyncWriteExt + Unpin>(value: Short, writer: &mut W) -> Result<()> {
+    writer.write_u16(value).await?;
     Ok(())
 }
 
-pub fn long<W: Write>(value: Long, writer: &mut W) -> Result<(), TransError> {
-    writer.write_all(&value.to_be_bytes())?;
+pub async fn long<W: AsyncWriteExt + Unpin>(value: Long, writer: &mut W) -> Result<()> {
+    writer.write_u32(value).await?;
     Ok(())
 }
 
-pub fn longlong<W: Write>(value: Longlong, writer: &mut W) -> Result<(), TransError> {
-    writer.write_all(&value.to_be_bytes())?;
+pub async fn longlong<W: AsyncWriteExt + Unpin>(value: Longlong, writer: &mut W) -> Result<()> {
+    writer.write_u64(value).await?;
     Ok(())
 }
 
-pub fn bit<W: Write>(value: &[Bit], writer: &mut W) -> Result<(), TransError> {
+pub async fn bit<W: AsyncWriteExt + Unpin>(value: &[Bit], writer: &mut W) -> Result<()> {
     // accumulate bits into bytes, starting from the least significant bit in each byte
 
     // how many bits have already been packed into `current_buf`
@@ -36,7 +35,7 @@ pub fn bit<W: Write>(value: &[Bit], writer: &mut W) -> Result<(), TransError> {
 
     for &bit in value {
         if already_filled >= 8 {
-            writer.write_all(&[current_buf])?;
+            writer.write_u8(current_buf).await?;
             current_buf = 0;
             already_filled = 0;
         }
@@ -47,146 +46,151 @@ pub fn bit<W: Write>(value: &[Bit], writer: &mut W) -> Result<(), TransError> {
     }
 
     if already_filled > 0 {
-        writer.write_all(&[current_buf])?;
+        writer.write_u8(current_buf).await?;
     }
 
     Ok(())
 }
 
-pub fn shortstr<W: Write>(value: Shortstr, writer: &mut W) -> Result<(), TransError> {
+pub async fn shortstr<W: AsyncWriteExt + Unpin>(value: Shortstr, writer: &mut W) -> Result<()> {
     let len = u8::try_from(value.len()).context("shortstr too long")?;
-    writer.write_all(&[len])?;
-    writer.write_all(value.as_bytes())?;
+    writer.write_u8(len).await?;
+    writer.write_all(value.as_bytes()).await?;
 
     Ok(())
 }
 
-pub fn longstr<W: Write>(value: Longstr, writer: &mut W) -> Result<(), TransError> {
+pub async fn longstr<W: AsyncWriteExt + Unpin>(value: Longstr, writer: &mut W) -> Result<()> {
     let len = u32::try_from(value.len()).context("longstr too long")?;
-    writer.write_all(&len.to_be_bytes())?;
-    writer.write_all(value.as_slice())?;
+    writer.write_u32(len).await?;
+    writer.write_all(value.as_slice()).await?;
 
     Ok(())
 }
 
-pub fn timestamp<W: Write>(value: Timestamp, writer: &mut W) -> Result<(), TransError> {
-    writer.write_all(&value.to_be_bytes())?;
+pub async fn timestamp<W: AsyncWriteExt + Unpin>(value: Timestamp, writer: &mut W) -> Result<()> {
+    writer.write_u64(value).await?;
     Ok(())
 }
 
-pub fn table<W: Write>(table: Table, writer: &mut W) -> Result<(), TransError> {
+pub async fn table<W: AsyncWriteExt + Unpin>(table: Table, writer: &mut W) -> Result<()> {
     let len = u32::try_from(table.len()).context("table too big")?;
-    writer.write_all(&len.to_be_bytes())?;
+    writer.write_u32(len).await?;
 
     for (field_name, value) in table {
-        shortstr(field_name, writer)?;
-        field_value(value, writer)?;
+        shortstr(field_name, writer).await?;
+        field_value(value, writer).await?;
     }
 
     Ok(())
 }
 
-fn field_value<W: Write>(value: FieldValue, writer: &mut W) -> Result<(), TransError> {
-    match value {
-        FieldValue::Boolean(bool) => {
-            writer.write_all(&[b't', u8::from(bool)])?;
-        }
-        FieldValue::ShortShortInt(int) => {
-            writer.write_all(b"b")?;
-            writer.write_all(&int.to_be_bytes())?;
-        }
-        FieldValue::ShortShortUInt(int) => {
-            writer.write_all(&[b'B', int])?;
-        }
-        FieldValue::ShortInt(int) => {
-            writer.write_all(b"U")?;
-            writer.write_all(&int.to_be_bytes())?;
-        }
-        FieldValue::ShortUInt(int) => {
-            writer.write_all(b"u")?;
-            writer.write_all(&int.to_be_bytes())?;
-        }
-        FieldValue::LongInt(int) => {
-            writer.write_all(b"I")?;
-            writer.write_all(&int.to_be_bytes())?;
-        }
-        FieldValue::LongUInt(int) => {
-            writer.write_all(b"i")?;
-            writer.write_all(&int.to_be_bytes())?;
-        }
-        FieldValue::LongLongInt(int) => {
-            writer.write_all(b"L")?;
-            writer.write_all(&int.to_be_bytes())?;
-        }
-        FieldValue::LongLongUInt(int) => {
-            writer.write_all(b"l")?;
-            writer.write_all(&int.to_be_bytes())?;
-        }
-        FieldValue::Float(float) => {
-            writer.write_all(b"f")?;
-            writer.write_all(&float.to_be_bytes())?;
-        }
-        FieldValue::Double(float) => {
-            writer.write_all(b"d")?;
-            writer.write_all(&float.to_be_bytes())?;
-        }
-        FieldValue::DecimalValue(scale, long) => {
-            writer.write_all(&[b'D', scale])?;
-            writer.write_all(&long.to_be_bytes())?;
-        }
-        FieldValue::ShortString(str) => {
-            writer.write_all(b"s")?;
-            shortstr(str, writer)?;
-        }
-        FieldValue::LongString(str) => {
-            writer.write_all(b"S")?;
-            longstr(str, writer)?;
-        }
-        FieldValue::FieldArray(array) => {
-            writer.write_all(b"A")?;
-            let len = u32::try_from(array.len()).context("array too long")?;
-            writer.write_all(&len.to_be_bytes())?;
+fn field_value<W: AsyncWriteExt + Unpin>(
+    value: FieldValue,
+    writer: &mut W,
+) -> Pin<Box<dyn Future<Output = Result<()>> + '_>> {
+    Box::pin(async {
+        match value {
+            FieldValue::Boolean(bool) => {
+                writer.write_all(&[b't', u8::from(bool)]).await?;
+            }
+            FieldValue::ShortShortInt(int) => {
+                writer.write_all(b"b").await?;
+                writer.write_all(&int.to_be_bytes()).await?;
+            }
+            FieldValue::ShortShortUInt(int) => {
+                writer.write_all(&[b'B', int]).await?;
+            }
+            FieldValue::ShortInt(int) => {
+                writer.write_all(b"U").await?;
+                writer.write_all(&int.to_be_bytes()).await?;
+            }
+            FieldValue::ShortUInt(int) => {
+                writer.write_all(b"u").await?;
+                writer.write_all(&int.to_be_bytes()).await?;
+            }
+            FieldValue::LongInt(int) => {
+                writer.write_all(b"I").await?;
+                writer.write_all(&int.to_be_bytes()).await?;
+            }
+            FieldValue::LongUInt(int) => {
+                writer.write_all(b"i").await?;
+                writer.write_all(&int.to_be_bytes()).await?;
+            }
+            FieldValue::LongLongInt(int) => {
+                writer.write_all(b"L").await?;
+                writer.write_all(&int.to_be_bytes()).await?;
+            }
+            FieldValue::LongLongUInt(int) => {
+                writer.write_all(b"l").await?;
+                writer.write_all(&int.to_be_bytes()).await?;
+            }
+            FieldValue::Float(float) => {
+                writer.write_all(b"f").await?;
+                writer.write_all(&float.to_be_bytes()).await?;
+            }
+            FieldValue::Double(float) => {
+                writer.write_all(b"d").await?;
+                writer.write_all(&float.to_be_bytes()).await?;
+            }
+            FieldValue::DecimalValue(scale, long) => {
+                writer.write_all(&[b'D', scale]).await?;
+                writer.write_all(&long.to_be_bytes()).await?;
+            }
+            FieldValue::ShortString(str) => {
+                writer.write_all(b"s").await?;
+                shortstr(str, writer).await?;
+            }
+            FieldValue::LongString(str) => {
+                writer.write_all(b"S").await?;
+                longstr(str, writer).await?;
+            }
+            FieldValue::FieldArray(array) => {
+                writer.write_all(b"A").await?;
+                let len = u32::try_from(array.len()).context("array too long")?;
+                writer.write_all(&len.to_be_bytes()).await?;
 
-            for element in array {
-                field_value(element, writer)?;
+                for element in array {
+                    field_value(element, writer).await?;
+                }
+            }
+            FieldValue::Timestamp(time) => {
+                writer.write_all(b"T").await?;
+                writer.write_all(&time.to_be_bytes()).await?;
+            }
+            FieldValue::FieldTable(_) => {
+                writer.write_all(b"F").await?;
+            }
+            FieldValue::Void => {
+                writer.write_all(b"V").await?;
             }
         }
-        FieldValue::Timestamp(time) => {
-            writer.write_all(b"T")?;
-            writer.write_all(&time.to_be_bytes())?;
-        }
-        FieldValue::FieldTable(_) => {
-            writer.write_all(b"F")?;
-        }
-        FieldValue::Void => {
-            writer.write_all(b"V")?;
-        }
-    }
-    Ok(())
+        Ok(())
+    })
 }
 
 #[cfg(test)]
 mod tests {
-    #[test]
-    fn pack_few_bits() {
+    #[tokio::test]
+    async fn pack_few_bits() {
         let bits = vec![true, false, true];
 
-        let mut buffer = [0u8; 1];
-        super::bit(bits, &mut buffer.as_mut_slice()).unwrap();
+        let mut buffer = Vec::new();
+        super::bit(&bits, &mut buffer).await.unwrap();
 
-        assert_eq!(buffer, [0b00000101])
+        assert_eq!(buffer.as_slice(), &[0b00000101])
     }
 
-    #[test]
-    fn pack_many_bits() {
+    #[tokio::test]
+    async fn pack_many_bits() {
         let bits = vec![
             /* first 8 */
             true, true, true, true, false, false, false, false, /* second 4 */
             true, false, true, true,
         ];
 
-        let mut buffer = [0u8; 2];
-        super::bit(bits, &mut buffer.as_mut_slice()).unwrap();
+        let mut buffer = Vec::new();
+        super::bit(&bits, &mut buffer).await.unwrap();
 
         assert_eq!(buffer, [0b00001111, 0b00001101]);
     }

amqp_transport/src/connection.rs

@@ -1,32 +1,39 @@
-use crate::error::{ProtocolError, TransError};
-use crate::frame;
-use crate::frame::FrameType;
+use crate::error::{ProtocolError, Result};
+use crate::frame::{Frame, FrameType};
+use crate::{classes, frame};
 use anyhow::Context;
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
 use tokio::net::TcpStream;
-use tracing::{debug, error};
+use tracing::{debug, error, warn};
+
+const MIN_MAX_FRAME_SIZE: usize = 4096;
 
 pub struct Connection {
     stream: TcpStream,
+    max_frame_size: usize,
 }
 
 impl Connection {
     pub fn new(stream: TcpStream) -> Self {
-        Self { stream }
+        Self {
+            stream,
+            max_frame_size: MIN_MAX_FRAME_SIZE,
+        }
     }
 
-    pub async fn start(mut self) {
+    pub async fn open_connection(mut self) {
         match self.run().await {
             Ok(()) => {}
             Err(err) => error!(%err, "Error during processing of connection"),
         }
     }
 
-    pub async fn run(&mut self) -> Result<(), TransError> {
+    pub async fn run(&mut self) -> Result<()> {
         self.negotiate_version().await?;
+        self.start().await?;
 
         loop {
-            let frame = frame::read_frame(&mut self.stream, 10000).await?;
+            let frame = frame::read_frame(&mut self.stream, self.max_frame_size).await?;
             debug!(?frame, "received frame");
             if frame.kind == FrameType::Method {
                 let class = super::classes::parse_method(&frame.payload)?;
@@ -35,7 +42,25 @@ impl Connection {
         }
     }
 
-    async fn negotiate_version(&mut self) -> Result<(), TransError> {
+    async fn start(&mut self) -> Result<()> {
+        let start_method = classes::Class::Connection(classes::Connection::Start {
+            version_major: 0,
+            version_minor: 9,
+            server_properties: Default::default(),
+            mechanisms: vec![],
+            locales: vec![],
+        });
+
+        let fut = classes::write::write_method(start_method, &mut self.stream);
+        warn!(size = %std::mem::size_of_val(&fut), "that future is big");
+        // todo fix out_buffer buffer things :spiral_eyes:
+        // maybe have a `size` method on `Class` and use `AsyncWrite`? oh god no that's horrible
+        // frame::write_frame(&mut self.stream, Frame {})?;
+
+        Ok(())
+    }
+
+    async fn negotiate_version(&mut self) -> Result<()> {
         const HEADER_SIZE: usize = 8;
         const SUPPORTED_PROTOCOL_VERSION: &[u8] = &[0, 9, 1];
         const OWN_PROTOCOL_HEADER: &[u8] = b"AMQP\0\0\x09\x01";

amqp_transport/src/error.rs

@@ -1,5 +1,7 @@
 use std::io::Error;
 
+pub type Result<T> = std::result::Result<T, TransError>;
+
 #[derive(Debug, thiserror::Error)]
 pub enum TransError {
     #[error("{0}")]

amqp_transport/src/frame.rs

@@ -1,6 +1,6 @@
-use crate::error::{ConException, ProtocolError, TransError};
+use crate::error::{ConException, ProtocolError, Result, TransError};
 use anyhow::Context;
-use tokio::io::AsyncReadExt;
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
 
 const REQUIRED_FRAME_END: u8 = 0xCE;
 
@@ -20,7 +20,7 @@ pub struct Frame {
     pub payload: Vec<u8>,
 }
 
-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, Copy, Clone, PartialEq, Eq)]
 #[repr(u8)]
 pub enum FrameType {
     Method = 1,
@@ -29,7 +29,21 @@ pub enum FrameType {
     Heartbeat = 8,
 }
 
-pub async fn read_frame<R>(r: &mut R, max_frame_size: usize) -> Result<Frame, TransError>
+pub async fn write_frame<W>(mut w: W, frame: &Frame) -> Result<()>
+where
+    W: AsyncWriteExt + Unpin,
+{
+    w.write_u8(frame.kind as u8).await?;
+    w.write_u16(frame.channel).await?;
+    w.write_u32(u32::try_from(frame.payload.len()).context("frame size too big")?)
+        .await?;
+    w.write_all(&frame.payload).await?;
+    w.write_u8(REQUIRED_FRAME_END).await?;
+
+    Ok(())
+}
+
+pub async fn read_frame<R>(r: &mut R, max_frame_size: usize) -> Result<Frame>
 where
     R: AsyncReadExt + Unpin,
 {
@@ -59,7 +73,7 @@ where
     })
 }
 
-fn parse_frame_type(kind: u8, channel: u16) -> Result<FrameType, TransError> {
+fn parse_frame_type(kind: u8, channel: u16) -> Result<FrameType> {
     match kind {
         frame_type::METHOD => Ok(FrameType::Method),
         frame_type::HEADER => Ok(FrameType::Header),

amqp_transport/src/lib.rs

@@ -22,6 +22,6 @@ pub async fn do_thing_i_guess() -> Result<()> {
 
         let connection = Connection::new(stream);
 
-        tokio::spawn(connection.start());
+        tokio::spawn(connection.open_connection());
     }
 }