diff --git a/.gitignore b/.gitignore
index 408b8a5..30fc9a3 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1,3 +1,4 @@
 /target
 Cargo.lock
-.idea
\ No newline at end of file
+.idea
+.DS_Store
diff --git a/Cargo.toml b/Cargo.toml
index 773e62a..ab32c32 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -31,3 +31,4 @@ dashmap = "4.0.2"
 [dev-dependencies]
 tracing-subscriber = "0.3.1"
 fake = { version = "2.4", features=['derive']}
+chrono = "0.4.19"
diff --git a/examples/complex_message.rs b/examples/complex_message.rs
new file mode 100644
index 0000000..2a840d7
--- /dev/null
+++ b/examples/complex_message.rs
@@ -0,0 +1,86 @@
+use std::sync::Arc;
+
+use chrono::Utc;
+use futures::StreamExt;
+use rabbitmq_stream_client::{
+    types::{ByteCapacity, Message, OffsetSpecification},
+    Environment,
+};
+use tokio::sync::Barrier;
+use tracing::{info, Level};
+use tracing_subscriber::FmtSubscriber;
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let subscriber = FmtSubscriber::builder()
+        .with_max_level(Level::TRACE)
+        .finish();
+
+    tracing::subscriber::set_global_default(subscriber).expect("setting default subscriber failed");
+    let environment = Environment::builder()
+        .host("localhost")
+        .port(5552)
+        .build()
+        .await?;
+
+    let stream_name = "complex_message";
+    let _ = environment.delete_stream(stream_name).await;
+    let message_count = 1;
+    environment
+        .stream_creator()
+        .max_length(ByteCapacity::GB(2))
+        .create(stream_name)
+        .await?;
+
+    let producer = environment.producer().build(stream_name).await?;
+
+    let barrier = Arc::new(Barrier::new(message_count + 1));
+    for i in 0..message_count {
+        let producer_cloned = producer.clone();
+
+        let barrier_cloned = barrier.clone();
+        tokio::task::spawn(async move {
+            let message = Message::builder()
+                .message_annotations()
+                .insert("test", i as i32)
+                .message_builder()
+                .application_properties()
+                .insert("test", i as i32)
+                .message_builder()
+                .properties()
+                .content_encoding("application/json")
+                .absolute_expiry_time(Utc::now())
+                .message_builder()
+                .body(format!("message{}", i))
+                .build();
+            producer_cloned.send_with_confirm(message).await.unwrap();
+
+            barrier_cloned.wait().await;
+        });
+    }
+
+    barrier.wait().await;
+
+    producer.close().await?;
+
+    let mut consumer = environment
+        .consumer()
+        .offset(OffsetSpecification::First)
+        .build(stream_name)
+        .await
+        .unwrap();
+
+    for _ in 0..message_count {
+        let delivery = consumer.next().await.unwrap()?;
+        info!(
+            "Got message: {:#?} with offset: {}",
+            delivery.message(),
+            delivery.offset(),
+        );
+    }
+
+    consumer.handle().close().await.unwrap();
+
+    environment.delete_stream(stream_name).await?;
+    Ok(())
+}
diff --git a/protocol/Cargo.toml b/protocol/Cargo.toml
index 83f4a68..797c458 100644
--- a/protocol/Cargo.toml
+++ b/protocol/Cargo.toml
@@ -6,12 +6,18 @@ edition = "2018"
 license = "Apache-2.0 OR MPL-2.0"
 
 
+[features]
+wasm-bindgen = ["uuid/wasm-bindgen"]
+
 [dependencies]
 byteorder = "1"
-ntex-amqp-codec = "= 0.7.2"
-ntex-bytes = "0.1"
-
+ordered-float = "2.10.0"
+uuid = "0.8.2"
+chrono = "0.4.19"
+num_enum = "0.5.7"
+derive_more = "0.99"
 
 [dev-dependencies]
-fake = { version = "2.4", features=['derive']}
+fake = { version = "2.4", features=['derive', 'chrono','uuid']}
 rand = "0.8"
+pretty_assertions = "1.2.0"
diff --git a/protocol/src/codec/decoder.rs b/protocol/src/codec/decoder.rs
index 53e98f1..64bb905 100644
--- a/protocol/src/codec/decoder.rs
+++ b/protocol/src/codec/decoder.rs
@@ -2,6 +2,7 @@ use std::collections::HashMap;
 
 use byteorder::ByteOrder;
 
+use crate::error::IncompleteError;
 use crate::message::Message;
 use crate::types::PublishedMessage;
 use crate::types::PublishingError;
@@ -12,42 +13,42 @@ use super::Decoder;
 
 impl Decoder for i8 {
     fn decode(input: &[u8]) -> Result<(&[u8], Self), crate::error::DecodeError> {
-        read_i8(input)
+        read_i8(input).map(Ok)?
     }
 }
 
 impl Decoder for i32 {
     fn decode(input: &[u8]) -> Result<(&[u8], Self), crate::error::DecodeError> {
-        read_i32(input)
+        read_i32(input).map(Ok)?
     }
 }
 
 impl Decoder for u8 {
     fn decode(input: &[u8]) -> Result<(&[u8], Self), crate::error::DecodeError> {
-        read_u8(input)
+        read_u8(input).map(Ok)?
     }
 }
 
 impl Decoder for u16 {
     fn decode(input: &[u8]) -> Result<(&[u8], Self), crate::error::DecodeError> {
-        read_u16(input)
+        read_u16(input).map(Ok)?
     }
 }
 
 impl Decoder for u32 {
     fn decode(input: &[u8]) -> Result<(&[u8], Self), crate::error::DecodeError> {
-        read_u32(input)
+        read_u32(input).map(Ok)?
     }
 }
 
 impl Decoder for u64 {
     fn decode(input: &[u8]) -> Result<(&[u8], Self), crate::error::DecodeError> {
-        read_u64(input)
+        read_u64(input).map(Ok)?
     }
 }
 impl Decoder for i64 {
     fn decode(input: &[u8]) -> Result<(&[u8], Self), crate::error::DecodeError> {
-        read_i64(input)
+        read_i64(input).map(Ok)?
     }
 }
 
@@ -199,9 +200,9 @@ impl Decoder for Vec<PublishingError> {
     }
 }
 
-pub fn check_len(input: &[u8], size: usize) -> Result<(), DecodeError> {
+pub fn check_len(input: &[u8], size: usize) -> Result<(), IncompleteError> {
     if input.len() < size {
-        return Err(DecodeError::Incomplete(size));
+        return Err(IncompleteError(size));
     }
     Ok(())
 }
@@ -209,7 +210,7 @@ pub fn check_len(input: &[u8], size: usize) -> Result<(), DecodeError> {
 macro_rules! reader {
     ( $fn:ident, $size:expr, $ret:ty) => {
         #[allow(unused)]
-        pub fn $fn(input: &[u8]) -> Result<(&[u8], $ret), crate::error::DecodeError> {
+        pub fn $fn(input: &[u8]) -> Result<(&[u8], $ret), IncompleteError> {
             check_len(input, $size)?;
             let x = byteorder::BigEndian::$fn(input);
             Ok((&input[$size..], x))
@@ -217,19 +218,26 @@ macro_rules! reader {
     };
 }
 
-pub fn read_u8(input: &[u8]) -> Result<(&[u8], u8), DecodeError> {
+pub fn read_u8(input: &[u8]) -> Result<(&[u8], u8), IncompleteError> {
     check_len(input, 1)?;
     Ok((&input[1..], input[0]))
 }
 
-pub fn read_i8(input: &[u8]) -> Result<(&[u8], i8), DecodeError> {
+pub fn read_i8(input: &[u8]) -> Result<(&[u8], i8), IncompleteError> {
     check_len(input, 1)?;
     Ok((&input[1..], input[0] as i8))
 }
 
+pub fn read_exact(input: &[u8], len: usize) -> Result<(&[u8], &[u8]), IncompleteError> {
+    check_len(input, len)?;
+    Ok((&input[len..], &input[..len]))
+}
+
 reader!(read_i16, 2, i16);
 reader!(read_u16, 2, u16);
 reader!(read_u32, 4, u32);
 reader!(read_i32, 4, i32);
 reader!(read_u64, 8, u64);
 reader!(read_i64, 8, i64);
+reader!(read_f32, 4, f32);
+reader!(read_f64, 8, f64);
diff --git a/protocol/src/commands/deliver.rs b/protocol/src/commands/deliver.rs
index 6b5a357..c4a5e77 100644
--- a/protocol/src/commands/deliver.rs
+++ b/protocol/src/commands/deliver.rs
@@ -160,15 +160,11 @@ mod tests {
     use fake::{Dummy, Faker};
 
     use crate::{commands::tests::command_encode_decode_test, message::InternalMessage};
-    use ntex_amqp_codec::Message as AmpqMessage;
 
     use super::{DeliverCommand, Message};
     impl Dummy<Faker> for Message {
         fn dummy_with_rng<R: rand::Rng + ?Sized>(_config: &Faker, _rng: &mut R) -> Self {
-            Message::new(InternalMessage {
-                message: AmpqMessage::default(),
-                publishing_id: None,
-            })
+            Message::new(InternalMessage::default())
         }
     }
 
diff --git a/protocol/src/error.rs b/protocol/src/error.rs
index 8351816..bc10ed2 100644
--- a/protocol/src/error.rs
+++ b/protocol/src/error.rs
@@ -2,12 +2,13 @@ use std::string::FromUtf8Error;
 
 #[derive(Debug)]
 pub enum DecodeError {
-    Incomplete(usize),
+    Incomplete(IncompleteError),
     Utf8Error(FromUtf8Error),
     UnknownResponseCode(u16),
     UnsupportedResponseType(u16),
     MismatchSize(usize),
     MessageParse(String),
+    InvalidFormatCode(u8),
     Empty,
 }
 
@@ -28,3 +29,12 @@ impl From<FromUtf8Error> for DecodeError {
         DecodeError::Utf8Error(err)
     }
 }
+
+impl From<IncompleteError> for DecodeError {
+    fn from(err: IncompleteError) -> Self {
+        DecodeError::Incomplete(err)
+    }
+}
+
+#[derive(Debug)]
+pub struct IncompleteError(pub usize);
diff --git a/protocol/src/lib.rs b/protocol/src/lib.rs
index f98ed25..f806431 100644
--- a/protocol/src/lib.rs
+++ b/protocol/src/lib.rs
@@ -8,3 +8,5 @@ mod response;
 pub mod types;
 pub use request::*;
 pub use response::*;
+
+pub mod utils;
diff --git a/protocol/src/message.rs b/protocol/src/message.rs
deleted file mode 100644
index f32e929..0000000
--- a/protocol/src/message.rs
+++ /dev/null
@@ -1,100 +0,0 @@
-use std::sync::Arc;
-
-use ntex_amqp_codec::{Encode, Message as AmpqMessage};
-use ntex_bytes::BytesMut;
-
-use crate::{
-    codec::{Decoder, Encoder},
-    error::DecodeError,
-};
-
-#[derive(Debug, PartialEq, Clone)]
-pub struct Message(Arc<InternalMessage>);
-
-#[derive(Debug, PartialEq)]
-pub struct InternalMessage {
-    pub(crate) publishing_id: Option<u64>,
-    pub(crate) message: AmpqMessage,
-}
-
-unsafe impl Send for Message {}
-unsafe impl Sync for Message {}
-
-impl Encoder for Message {
-    fn encoded_size(&self) -> u32 {
-        self.0.message.encoded_size() as u32
-    }
-
-    fn encode(&self, writer: &mut impl std::io::Write) -> Result<(), crate::error::EncodeError> {
-        let mut buf = BytesMut::with_capacity(self.encoded_size() as usize);
-
-        ntex_amqp_codec::Encode::encode(&self.0.message, &mut buf);
-
-        writer.write_all(&buf)?;
-
-        Ok(())
-    }
-}
-
-impl Message {
-    #[cfg(test)]
-    pub(crate) fn new(internal: InternalMessage) -> Message {
-        Message(Arc::new(internal))
-    }
-    pub fn builder() -> MessageBuilder {
-        MessageBuilder(InternalMessage {
-            message: AmpqMessage::default(),
-            publishing_id: None,
-        })
-    }
-
-    pub fn data(&self) -> Option<&[u8]> {
-        self.0.message.body().data().map(|data| data.as_ref())
-    }
-
-    /// Get a reference to the message's publishing id.
-    pub fn publishing_id(&self) -> Option<&u64> {
-        self.0.publishing_id.as_ref()
-    }
-}
-
-impl Decoder for Message {
-    fn decode(input: &[u8]) -> Result<(&[u8], Self), crate::error::DecodeError> {
-        ntex_amqp_codec::Decode::decode(input)
-            .map_err(|err| DecodeError::MessageParse(err.to_string()))
-            .map(|message| {
-                (
-                    message.0,
-                    Message(Arc::new(InternalMessage {
-                        publishing_id: None,
-                        message: message.1,
-                    })),
-                )
-            })
-    }
-}
-
-pub struct MessageBuilder(InternalMessage);
-
-impl MessageBuilder {
-    pub fn body(mut self, data: impl Into<Vec<u8>>) -> Self {
-        self.0
-            .message
-            .set_body(|body| body.set_data(data.into().into()));
-        self
-    }
-
-    pub fn publising_id(mut self, publishing_id: u64) -> Self {
-        self.0.publishing_id = Some(publishing_id);
-        self
-    }
-    pub fn build(self) -> Message {
-        Message(Arc::new(self.0))
-    }
-}
-
-impl From<Message> for Vec<Message> {
-    fn from(message: Message) -> Self {
-        vec![message]
-    }
-}
diff --git a/protocol/src/message/amqp/body.rs b/protocol/src/message/amqp/body.rs
new file mode 100644
index 0000000..0694ea5
--- /dev/null
+++ b/protocol/src/message/amqp/body.rs
@@ -0,0 +1,89 @@
+use crate::message::amqp::codec::AmqpEncoder;
+
+use super::{
+    codec::constants::{MESSAGE_BODY_DATA, MESSAGE_BODY_SEQUENCE, MESSAGE_BODY_VALUE},
+    types::{List, Value},
+    AmqpEncodeError,
+};
+
+#[derive(Debug, Clone, Default, PartialEq)]
+pub struct MessageBody {
+    pub data: Vec<Vec<u8>>,
+    pub sequence: Vec<List>,
+    pub value: Option<Value>,
+}
+
+impl MessageBody {
+    pub fn data(&self) -> Option<&Vec<u8>> {
+        self.data.get(0)
+    }
+
+    pub fn value(&self) -> Option<&Value> {
+        self.value.as_ref()
+    }
+
+    #[cfg(test)]
+    pub fn set_value(&mut self, value: impl Into<Value>) -> &mut Self {
+        self.value = Some(value.into());
+        self
+    }
+
+    pub fn set_data(&mut self, data: impl Into<Vec<u8>>) -> &mut Self {
+        self.data.clear();
+        self.data.push(data.into());
+        self
+    }
+}
+
+impl AmqpEncoder for MessageBody {
+    fn encoded_size(&self) -> u32 {
+        let mut size = self.data.iter().fold(0, |a, d| {
+            a + d.encoded_size() + MESSAGE_BODY_DATA.encoded_size()
+        });
+        size += self.sequence.iter().fold(0, |a, seq| {
+            a + seq.encoded_size() + MESSAGE_BODY_SEQUENCE.encoded_size()
+        });
+
+        if let Some(ref val) = self.value {
+            size + val.encoded_size() + MESSAGE_BODY_VALUE.encoded_size()
+        } else {
+            size
+        }
+    }
+
+    fn encode(&self, writer: &mut impl std::io::Write) -> Result<(), AmqpEncodeError> {
+        for data in &self.data {
+            MESSAGE_BODY_DATA.encode(writer)?;
+            data.encode(writer)?;
+        }
+
+        for sequence in &self.sequence {
+            MESSAGE_BODY_SEQUENCE.encode(writer)?;
+            sequence.encode(writer)?;
+        }
+        if let Some(ref val) = self.value {
+            MESSAGE_BODY_VALUE.encode(writer)?;
+            val.encode(writer)?;
+        }
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+
+    use fake::{Dummy, Fake, Faker};
+
+    use crate::message::amqp::Value;
+
+    use super::MessageBody;
+    impl Dummy<Faker> for MessageBody {
+        fn dummy_with_rng<R: rand::Rng + ?Sized>(config: &Faker, rng: &mut R) -> Self {
+            MessageBody {
+                data: config.fake_with_rng(rng),
+                sequence: config.fake_with_rng(rng),
+                value: Some(Value::Simple(config.fake_with_rng(rng))),
+            }
+        }
+    }
+}
diff --git a/protocol/src/message/amqp/codec/constants.rs b/protocol/src/message/amqp/codec/constants.rs
new file mode 100644
index 0000000..6dd4b8b
--- /dev/null
+++ b/protocol/src/message/amqp/codec/constants.rs
@@ -0,0 +1,88 @@
+use std::convert::TryInto;
+
+use crate::{
+    codec::decoder::read_u8,
+    message::amqp::{
+        error::{AmqpDecodeError, AmqpEncodeError},
+        types::Descriptor,
+    },
+};
+
+pub const MESSAGE_HEADER: Descriptor = Descriptor::Ulong(112);
+pub const MESSAGE_DELIVERY_ANNOTATIONS: Descriptor = Descriptor::Ulong(113);
+pub const MESSAGE_ANNOTATIONS: Descriptor = Descriptor::Ulong(114);
+pub const MESSAGE_PROPERTIES: Descriptor = Descriptor::Ulong(115);
+pub const MESSAGE_APPLICATION_PROPERTIES: Descriptor = Descriptor::Ulong(116);
+pub const MESSAGE_BODY_DATA: Descriptor = Descriptor::Ulong(117);
+pub const MESSAGE_BODY_SEQUENCE: Descriptor = Descriptor::Ulong(118);
+pub const MESSAGE_BODY_VALUE: Descriptor = Descriptor::Ulong(119);
+pub const MESSAGE_FOOTER: Descriptor = Descriptor::Ulong(120);
+
+use byteorder::WriteBytesExt;
+use num_enum::{IntoPrimitive, TryFromPrimitive};
+
+use super::{AmqpDecoder, AmqpEncoder};
+
+#[derive(Debug, TryFromPrimitive, Clone, Copy, IntoPrimitive)]
+#[repr(u8)]
+pub enum TypeCode {
+    Described = 0x00,
+    Null = 0x40,
+    Boolean = 0x56,
+    BooleanTrue = 0x41,
+    BooleanFalse = 0x42,
+    UInt0 = 0x43,
+    ULong0 = 0x44,
+    UByte = 0x50,
+    UShort = 0x60,
+    UInt = 0x70,
+    ULong = 0x80,
+    Byte = 0x51,
+    Short = 0x61,
+    Int = 0x71,
+    Long = 0x81,
+    UIntSmall = 0x52,
+    ULongSmall = 0x53,
+    IntSmall = 0x54,
+    LongSmall = 0x55,
+    Float = 0x72,
+    Double = 0x82,
+    Char = 0x73,
+    Timestamp = 0x83,
+    Uuid = 0x98,
+    Binary8 = 0xa0,
+    Binary32 = 0xb0,
+    String8 = 0xa1,
+    String32 = 0xb1,
+    Symbol8 = 0xa3,
+    Symbol32 = 0xb3,
+    List0 = 0x45,
+    List8 = 0xc0,
+    List32 = 0xd0,
+    Map8 = 0xc1,
+    Map32 = 0xd1,
+    Array8 = 0xe0,
+    Array32 = 0xf0,
+}
+
+impl AmqpEncoder for TypeCode {
+    fn encoded_size(&self) -> u32 {
+        1
+    }
+
+    fn encode(&self, writer: &mut impl std::io::Write) -> Result<(), AmqpEncodeError> {
+        let code: u8 = (*self).into();
+        writer.write_u8(code).map(Ok)?
+    }
+}
+
+impl AmqpDecoder for TypeCode {
+    fn decode(input: &[u8]) -> Result<(&[u8], Self), AmqpDecodeError> {
+        let (input, code) = read_u8(input)?;
+        Ok((
+            input,
+            code.try_into()
+                .map_err(|_| AmqpDecodeError::InvalidTypeCode(code))?,
+        ))
+    }
+}
diff --git a/protocol/src/message/amqp/codec/decoder.rs b/protocol/src/message/amqp/codec/decoder.rs
new file mode 100644
index 0000000..8b13789
--- /dev/null
+++ b/protocol/src/message/amqp/codec/decoder.rs
@@ -0,0 +1 @@
+
diff --git a/protocol/src/message/amqp/codec/mod.rs b/protocol/src/message/amqp/codec/mod.rs
new file mode 100644
index 0000000..1379a3a
--- /dev/null
+++ b/protocol/src/message/amqp/codec/mod.rs
@@ -0,0 +1,27 @@
+use std::{any::type_name, io::Write};
+
+use self::constants::TypeCode;
+
+use super::error::{AmqpDecodeError, AmqpEncodeError};
+
+pub mod constants;
+mod decoder;
+
+pub trait AmqpEncoder {
+    fn encoded_size(&self) -> u32;
+    fn encode(&self, writer: &mut impl Write) -> Result<(), AmqpEncodeError>;
+}
+
+pub trait AmqpDecoder
+where
+    Self: Sized,
+{
+    fn decode(input: &[u8]) -> Result<(&[u8], Self), AmqpDecodeError>;
+
+    fn invalid_type_code(code: TypeCode) -> AmqpDecodeError {
+        AmqpDecodeError::InvalidTypeCodeFor {
+            target: type_name::<Self>().to_string(),
+            code,
+        }
+    }
+}
diff --git a/protocol/src/message/amqp/definitions.rs b/protocol/src/message/amqp/definitions.rs
new file mode 100644
index 0000000..e69de29
diff --git a/protocol/src/message/amqp/error.rs b/protocol/src/message/amqp/error.rs
new file mode 100644
index 0000000..1d73ce9
--- /dev/null
+++ b/protocol/src/message/amqp/error.rs
@@ -0,0 +1,43 @@
+use std::string::FromUtf8Error;
+
+use crate::error::IncompleteError;
+
+use super::codec::constants::TypeCode;
+
+#[derive(Debug)]
+pub enum AmqpEncodeError {
+    Io(std::io::Error),
+}
+
+impl From<std::io::Error> for AmqpEncodeError {
+    fn from(err: std::io::Error) -> Self {
+        AmqpEncodeError::Io(err)
+    }
+}
+
+#[derive(Debug)]
+pub enum AmqpDecodeError {
+    InvalidTypeCode(u8),
+    InvalidTypeCodeFor { target: String, code: TypeCode },
+    MessageParse(String),
+    Incomplete(IncompleteError),
+    Utf8Error(FromUtf8Error),
+    UuidError(uuid::Error),
+}
+
+impl From<IncompleteError> for AmqpDecodeError {
+    fn from(err: IncompleteError) -> Self {
+        AmqpDecodeError::Incomplete(err)
+    }
+}
+
+impl From<FromUtf8Error> for AmqpDecodeError {
+    fn from(err: FromUtf8Error) -> Self {
+        AmqpDecodeError::Utf8Error(err)
+    }
+}
+impl From<uuid::Error> for AmqpDecodeError {
+    fn from(err: uuid::Error) -> Self {
+        AmqpDecodeError::UuidError(err)
+    }
+}
diff --git a/protocol/src/message/amqp/fixtures/empty_message b/protocol/src/message/amqp/fixtures/empty_message
new file mode 100644
index 0000000..0b06f60
Binary files /dev/null and b/protocol/src/message/amqp/fixtures/empty_message differ
diff --git a/protocol/src/message/amqp/fixtures/header_amqpvalue_message b/protocol/src/message/amqp/fixtures/header_amqpvalue_message
new file mode 100644
index 0000000..36639cb
Binary files /dev/null and b/protocol/src/message/amqp/fixtures/header_amqpvalue_message differ
diff --git a/protocol/src/message/amqp/fixtures/message_body_250 b/protocol/src/message/amqp/fixtures/message_body_250
new file mode 100644
index 0000000..d4585c6
Binary files /dev/null and b/protocol/src/message/amqp/fixtures/message_body_250 differ
diff --git a/protocol/src/message/amqp/fixtures/message_body_700 b/protocol/src/message/amqp/fixtures/message_body_700
new file mode 100644
index 0000000..5e51a9c
Binary files /dev/null and b/protocol/src/message/amqp/fixtures/message_body_700 differ
diff --git a/protocol/src/message/amqp/fixtures/message_random_application_properties_300 b/protocol/src/message/amqp/fixtures/message_random_application_properties_300
new file mode 100644
index 0000000..980d714
Binary files /dev/null and b/protocol/src/message/amqp/fixtures/message_random_application_properties_300 differ
diff --git a/protocol/src/message/amqp/fixtures/message_random_application_properties_500 b/protocol/src/message/amqp/fixtures/message_random_application_properties_500
new file mode 100644
index 0000000..110ca77
Binary files /dev/null and b/protocol/src/message/amqp/fixtures/message_random_application_properties_500 differ
diff --git a/protocol/src/message/amqp/fixtures/message_random_application_properties_properties_900 b/protocol/src/message/amqp/fixtures/message_random_application_properties_properties_900
new file mode 100644
index 0000000..44f7855
Binary files /dev/null and b/protocol/src/message/amqp/fixtures/message_random_application_properties_properties_900 differ
diff --git a/protocol/src/message/amqp/fixtures/static_test_message_compare b/protocol/src/message/amqp/fixtures/static_test_message_compare
new file mode 100644
index 0000000..2fc584a
Binary files /dev/null and b/protocol/src/message/amqp/fixtures/static_test_message_compare differ
diff --git a/protocol/src/message/amqp/header.rs b/protocol/src/message/amqp/header.rs
new file mode 100644
index 0000000..0b5d317
--- /dev/null
+++ b/protocol/src/message/amqp/header.rs
@@ -0,0 +1,115 @@
+use byteorder::{BigEndian, WriteBytesExt};
+
+use crate::message::amqp::{
+    codec::constants::MESSAGE_HEADER,
+    types::{Descriptor, UInt},
+};
+
+use super::{
+    codec::constants::TypeCode,
+    types::Milliseconds,
+    types::{list_decoder, Boolean, List},
+    AmqpDecodeError, AmqpDecoder, AmqpEncodeError, AmqpEncoder,
+};
+#[cfg(test)]
+use fake::Fake;
+
+/// Header of the message
+#[derive(Clone, Debug, PartialEq)]
+#[cfg_attr(test, derive(fake::Dummy))]
+pub struct Header {
+    pub durable: bool,
+    pub priority: u8,
+    pub ttl: Option<Milliseconds>,
+    pub first_acquirer: bool,
+    pub delivery_count: u32,
+}
+
+impl Default for Header {
+    fn default() -> Self {
+        Self {
+            durable: Default::default(),
+            priority: 4,
+            ttl: Default::default(),
+            first_acquirer: Default::default(),
+            delivery_count: 0,
+        }
+    }
+}
+
+impl Header {
+    fn content_size(&self) -> u32 {
+        self.durable.encoded_size()
+            + self.priority.encoded_size()
+            + self.ttl.encoded_size()
+            + self.first_acquirer.encoded_size()
+            + self.delivery_count.encoded_size()
+    }
+}
+
+impl AmqpEncoder for Header {
+    fn encoded_size(&self) -> u32 {
+        let size = self.content_size() + MESSAGE_HEADER.encoded_size();
+        let fixed = if size > u8::MAX as u32 { 9 } else { 3 };
+        fixed + size
+    }
+
+    fn encode(&self, writer: &mut impl std::io::Write) -> Result<(), AmqpEncodeError> {
+        MESSAGE_HEADER.encode(writer)?;
+
+        let content_size = self.content_size();
+
+        if content_size + MESSAGE_HEADER.encoded_size() > u8::MAX as u32 {
+            TypeCode::List8.encode(writer)?;
+            writer.write_u32::<BigEndian>(content_size + 4)?;
+            writer.write_u32::<BigEndian>(5)?;
+        } else {
+            TypeCode::List8.encode(writer)?;
+            writer.write_u8((content_size + 4) as u8)?;
+            writer.write_u8(5)?;
+        }
+        self.durable.encode(writer)?;
+        self.priority.encode(writer)?;
+        self.ttl.encode(writer)?;
+        self.first_acquirer.encode(writer)?;
+        self.delivery_count.encode(writer)?;
+        Ok(())
+    }
+}
+
+impl AmqpDecoder for Header {
+    fn decode(input: &[u8]) -> Result<(&[u8], Self), AmqpDecodeError> {
+        match Descriptor::decode(input)? {
+            (input, MESSAGE_HEADER) => {
+                let header = Header::default();
+                List::decode_with_fields(input, list_decoder_header, header)
+            }
+            (_, descriptor) => Err(AmqpDecodeError::MessageParse(format!(
+                "Invalid descriptor for header {:?}",
+                descriptor
+            ))),
+        }
+    }
+}
+
+list_decoder!(Header, list_decoder_header,
+    {
+     0 => { durable, Boolean, false},
+     1 => { priority, u8, 4},
+     2 => { ttl, u32, None, true},
+     3 => { first_acquirer, Boolean, false},
+     4 => { delivery_count, UInt, 4}
+    }
+);
+
+#[cfg(test)]
+mod tests {
+    use crate::message::amqp::tests::type_encode_decode_test_fuzzy;
+
+    use super::Header;
+
+    #[test]
+    fn test_header_encode_decode() {
+        type_encode_decode_test_fuzzy::<Header>()
+    }
+}
diff --git a/protocol/src/message/amqp/message.rs b/protocol/src/message/amqp/message.rs
new file mode 100644
index 0000000..ce6434e
--- /dev/null
+++ b/protocol/src/message/amqp/message.rs
@@ -0,0 +1,509 @@
+use super::body::MessageBody;
+use super::codec::constants::{
+    MESSAGE_ANNOTATIONS, MESSAGE_APPLICATION_PROPERTIES, MESSAGE_DELIVERY_ANNOTATIONS,
+    MESSAGE_FOOTER,
+};
+use super::codec::AmqpEncoder;
+use super::error::AmqpEncodeError;
+use super::header::Header;
+use super::properties::Properties;
+use super::section::MessageSection;
+use super::types::{ApplicationProperties, DeliveryAnnotations, Footer, MessageAnnotations};
+use super::{AmqpDecodeError, AmqpDecoder};
+
+#[cfg(test)]
+use fake::Fake;
+
+#[derive(Debug, Clone, Default, PartialEq)]
+#[cfg_attr(test, derive(fake::Dummy))]
+pub struct Message {
+    header: Option<Header>,
+    delivery_annotations: Option<DeliveryAnnotations>,
+    message_annotations: Option<MessageAnnotations>,
+    properties: Option<Properties>,
+    application_properties: Option<ApplicationProperties>,
+    footer: Option<Footer>,
+    body: MessageBody,
+}
+
+impl Message {
+    /// Message body
+    pub fn body(&self) -> &MessageBody {
+        &self.body
+    }
+
+    pub fn set_body<F>(&mut self, f: F) -> &mut Self
+    where
+        F: FnOnce(&mut MessageBody),
+    {
+        f(&mut self.body);
+        self
+    }
+
+    #[cfg(test)]
+    pub fn with_header<F>(&mut self, f: F) -> &mut Self
+    where
+        F: FnOnce(&mut Header),
+    {
+        if self.header.is_none() {
+            self.header = Some(Header::default());
+        }
+
+        if let Some(ref mut header) = self.header.as_mut() {
+            f(header);
+        }
+
+        self
+    }
+
+    pub fn message_annotations_mut(&mut self) -> &mut MessageAnnotations {
+        if self.message_annotations.is_none() {
+            self.message_annotations = Some(MessageAnnotations::default());
+        }
+        self.message_annotations.as_mut().unwrap()
+    }
+    pub fn application_properties_mut(&mut self) -> &mut ApplicationProperties {
+        if self.application_properties.is_none() {
+            self.application_properties = Some(ApplicationProperties::default());
+        }
+        self.application_properties.as_mut().unwrap()
+    }
+    #[cfg(test)]
+    pub fn with_footer<F>(&mut self, f: F) -> &mut Self
+    where
+        F: FnOnce(&mut Footer),
+    {
+        if self.footer.is_none() {
+            self.footer = Some(Footer::default());
+        }
+
+        if let Some(ref mut footer) = self.footer.as_mut() {
+            f(footer)
+        }
+
+        self
+    }
+
+    #[cfg(test)]
+    pub(crate) fn set_properties(&mut self, properties: Properties) -> &mut Self {
+        self.properties = Some(properties);
+        self
+    }
+    pub fn with_properties<F>(&mut self, f: F) -> &mut Self
+    where
+        F: FnOnce(&mut Properties),
+    {
+        if self.properties.is_none() {
+            self.properties = Some(Properties::default());
+        }
+
+        if let Some(ref mut properties) = self.properties.as_mut() {
+            f(properties);
+        }
+
+        self
+    }
+
+    /// Get a reference to the message's application properties.
+    #[must_use]
+    pub fn application_properties(&self) -> Option<&ApplicationProperties> {
+        self.application_properties.as_ref()
+    }
+
+    /// Get a reference to the message's properties.
+    #[must_use]
+    pub fn properties(&self) -> Option<&Properties> {
+        self.properties.as_ref()
+    }
+
+    /// Get a reference to the message's header.
+    #[must_use]
+    pub fn header(&self) -> Option<&Header> {
+        self.header.as_ref()
+    }
+
+    /// Get a reference to the message's message annotations.
+    #[must_use]
+    pub fn message_annotations(&self) -> Option<&MessageAnnotations> {
+        self.message_annotations.as_ref()
+    }
+
+    /// Get a reference to the message's delivery annotations.
+    #[must_use]
+    pub fn delivery_annotations(&self) -> Option<&DeliveryAnnotations> {
+        self.delivery_annotations.as_ref()
+    }
+}
+
+impl AmqpDecoder for Message {
+    fn decode(input: &[u8]) -> Result<(&[u8], Self), AmqpDecodeError> {
+        let mut message = Message::default();
+        let mut input = input;
+        loop {
+            if input.is_empty() {
+                break;
+            }
+            let (inner, sec) = MessageSection::decode(input)?;
+            input = inner;
+            match sec {
+                MessageSection::Header(val) => {
+                    message.header = Some(val);
+                }
+                MessageSection::DeliveryAnnotations(val) => {
+                    message.delivery_annotations = Some(val);
+                }
+                MessageSection::MessageAnnotations(val) => {
+                    message.message_annotations = Some(val);
+                }
+                MessageSection::ApplicationProperties(val) => {
+                    message.application_properties = Some(val);
+                }
+                MessageSection::Footer(val) => {
+                    message.footer = Some(val);
+                }
+                MessageSection::Properties(val) => {
+                    message.properties = Some(val);
+                }
+                // body
+                MessageSection::AmqpSequence(val) => {
+                    message.body.sequence.push(val);
+                }
+                MessageSection::AmqpValue(val) => {
+                    message.body.value = Some(val);
+                }
+                MessageSection::Data(val) => {
+                    message.body.data.push(val);
+                }
+            }
+        }
+        Ok((input, message))
+    }
+}
+
+impl AmqpEncoder for Message {
+    fn encoded_size(&self) -> u32 {
+        let mut size = self.body.encoded_size();
+
+        if let Some(ref h) = self.header {
+            size += h.encoded_size();
+        }
+        if let Some(ref da) = self.delivery_annotations {
+            size += da.encoded_size() + MESSAGE_DELIVERY_ANNOTATIONS.encoded_size();
+        }
+        if let Some(ref ma) = self.message_annotations {
+            size += ma.encoded_size() + MESSAGE_ANNOTATIONS.encoded_size();
+        }
+        if let Some(ref p) = self.properties {
+            size += p.encoded_size();
+        }
+        if let Some(ref ap) = self.application_properties {
+            size += ap.encoded_size() + MESSAGE_APPLICATION_PROPERTIES.encoded_size();
+        }
+        if let Some(ref f) = self.footer {
+            size += f.encoded_size() + MESSAGE_FOOTER.encoded_size();
+        }
+
+        size as u32
+    }
+
+    fn encode(&self, writer: &mut impl std::io::Write) -> Result<(), AmqpEncodeError> {
+        if let Some(ref h) = self.header {
+            h.encode(writer)?;
+        }
+        if let Some(ref da) = self.delivery_annotations {
+            MESSAGE_DELIVERY_ANNOTATIONS.encode(writer)?;
+            da.encode(writer)?;
+        }
+        if let Some(ref ma) = self.message_annotations {
+            MESSAGE_ANNOTATIONS.encode(writer)?;
+            ma.encode(writer)?;
+        }
+        if let Some(ref p) = self.properties {
+            p.encode(writer)?;
+        }
+        if let Some(ref ap) = self.application_properties {
+            MESSAGE_APPLICATION_PROPERTIES.encode(writer)?;
+            ap.encode(writer)?;
+        }
+
+        self.body.encode(writer)?;
+
+        if let Some(ref f) = self.footer {
+            MESSAGE_FOOTER.encode(writer)?;
+            f.encode(writer)?;
+        }
+
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::convert::TryInto;
+
+    use fake::Fake;
+    use fake::Faker;
+
+    use super::Message;
+    use crate::message::amqp::codec::AmqpDecoder;
+    use crate::message::amqp::codec::AmqpEncoder;
+    use crate::message::amqp::tests::type_encode_decode_test_fuzzy;
+    use crate::message::amqp::types::List;
+    use crate::message::amqp::AnnonationKey;
+    use crate::message::amqp::Value;
+
+    #[test]
+    fn message_with_body_test() {
+        let mut message = Message::default();
+
+        message.set_body(|body| {
+            body.set_data(b"test:w".as_ref());
+        });
+
+        let mut buf = vec![];
+        message.encode(&mut buf).unwrap();
+
+        let (remaining, decoded) = Message::decode(&buf).unwrap();
+
+        assert_eq!(0, remaining.len());
+        assert_eq!(message, decoded);
+    }
+    #[test]
+    fn message_with_header_test() {
+        let mut message = Message::default();
+        message
+            .set_body(|body| {
+                body.set_data(b"test:w".as_ref());
+            })
+            .with_header(|header| {
+                header.ttl = Some(3000);
+                header.delivery_count = 32;
+                header.first_acquirer = true;
+            });
+
+        let mut buf = vec![];
+        message.encode(&mut buf).unwrap();
+
+        let (remaining, decoded) = Message::decode(&buf).unwrap();
+
+        assert_eq!(0, remaining.len());
+        assert_eq!(message, decoded);
+    }
+    #[test]
+    fn message_with_properties_test() {
+        let mut message = Message::default();
+        message
+            .set_body(|body| {
+                body.set_data(b"test:w".as_ref());
+            })
+            .set_properties(Faker.fake());
+
+        let mut buf = vec![];
+        message.encode(&mut buf).unwrap();
+
+        let (remaining, decoded) = match Message::decode(&buf) {
+            Ok(result) => result,
+            Err(err) => panic!(
+                "Failed to decode message {:#?} with error : {:?}",
+                message, err
+            ),
+        };
+
+        assert_eq!(0, remaining.len());
+        assert_eq!(message, decoded);
+    }
+    #[test]
+    fn message_with_footer_test() {
+        let mut message = Message::default();
+        message
+            .set_body(|body| {
+                body.set_data(b"test:w".as_ref());
+            })
+            .with_footer(|footer| {
+                let mut list = List::new();
+
+                list.push(1);
+                list.push("test");
+                footer.put("test", 1);
+                footer.put(
+                    (0..300).map(|idx| format!("{}", idx)).collect::<String>(),
+                    1,
+                );
+                footer.put(1, "test");
+                footer.put(1000, "test");
+                footer.put("list", list);
+            });
+
+        check_message_encode_decode(&message);
+    }
+    #[test]
+    fn message_with_body_value_test() {
+        let mut message = Message::default();
+        message.set_body(|body| {
+            body.set_data(b"test:w".as_ref()).set_value(10);
+        });
+
+        check_message_encode_decode(&message);
+    }
+
+    #[test]
+    fn test_message_encode_decode() {
+        type_encode_decode_test_fuzzy::<Message>()
+    }
+    #[test]
+    fn test_message_empty() {
+        let content = include_bytes!("fixtures/empty_message");
+
+        let (remaining, message) = Message::decode(content).unwrap();
+
+        assert!(remaining.is_empty());
+
+        assert_eq!(message.body().data(), Some(&vec![]));
+    }
+    #[test]
+    fn test_message_header_amqp_value() {
+        let content = include_bytes!("fixtures/header_amqpvalue_message");
+
+        let (remaining, message) = Message::decode(content).unwrap();
+
+        assert!(remaining.is_empty());
+
+        let header = message.header().unwrap();
+
+        assert_eq!(true, header.durable);
+        assert_eq!(true, header.first_acquirer);
+        assert_eq!(100, header.priority);
+        assert_eq!(300, header.delivery_count);
+        assert_eq!(0, header.ttl.unwrap_or_default());
+
+        assert!(message.body().data().is_some());
+    }
+    #[test]
+    fn test_message_random_application_properties_300() {
+        let content = include_bytes!("fixtures/message_random_application_properties_300");
+
+        let (remaining, message) = Message::decode(content).unwrap();
+
+        assert!(remaining.is_empty());
+
+        assert!(message.application_properties().unwrap().len() > 0);
+
+        assert!(message.body().data().unwrap().len() > u8::MAX as usize);
+    }
+    #[test]
+    fn test_message_random_application_properties_500() {
+        let content = include_bytes!("fixtures/message_random_application_properties_500");
+
+        let (remaining, message) = Message::decode(content).unwrap();
+
+        assert!(remaining.is_empty());
+
+        assert!(message.application_properties().unwrap().len() > 0);
+
+        assert!(message.body().data().unwrap().len() > u8::MAX as usize);
+    }
+    #[test]
+    fn test_message_random_application_properties_properties_900() {
+        let content =
+            include_bytes!("fixtures/message_random_application_properties_properties_900");
+
+        let (remaining, message) = Message::decode(content).unwrap();
+
+        assert!(remaining.is_empty());
+
+        assert!(message.application_properties().unwrap().len() > 0);
+        assert!(message.body().data().unwrap().len() > u8::MAX as usize);
+
+        for (k, v) in message.application_properties().unwrap().iter() {
+            assert!(k.len() >= 900);
+            let v: &String = v.try_into().unwrap();
+            assert!(v.len() >= 900);
+        }
+
+        let properties = message.properties().unwrap();
+
+        assert!(properties.reply_to.as_ref().unwrap().len() > 0);
+        assert!(properties.content_encoding.as_ref().unwrap().len() > 0);
+        assert!(properties.content_type.as_ref().unwrap().len() > 0);
+        assert!(properties.group_id.as_ref().unwrap().len() > 0);
+        assert!(properties.reply_to.as_ref().unwrap().len() > 0);
+
+        assert!(properties.absolute_expiry_time.is_some());
+        assert!(properties.creation_time.is_some());
+    }
+    #[test]
+    fn test_message_static_test_compare() {
+        let content = include_bytes!("fixtures/static_test_message_compare");
+
+        let (remaining, message) = Message::decode(content).unwrap();
+
+        assert!(remaining.is_empty());
+
+        assert_eq!(
+            Some(b"test".as_slice()),
+            message.body().data().map(|data| data.as_slice())
+        );
+
+        let properties = message.properties().unwrap();
+
+        let msg_id: &String = properties.message_id.as_ref().unwrap().try_into().unwrap();
+        assert_eq!("test", properties.subject.as_ref().unwrap());
+        assert_eq!("test", msg_id);
+        assert_eq!("test", properties.reply_to.as_ref().unwrap());
+        assert_eq!("test", properties.content_encoding.as_ref().unwrap());
+        assert_eq!("test", properties.content_type.as_ref().unwrap());
+        assert_eq!("test", properties.group_id.as_ref().unwrap());
+        assert_eq!("test", properties.reply_to.as_ref().unwrap());
+        assert_eq!(b"test".as_slice(), properties.user_id.as_ref().unwrap());
+
+        for (k, v) in message.application_properties().unwrap().iter() {
+            assert_eq!("test", k);
+            let v: &String = v.try_into().unwrap();
+            assert_eq!("test", v);
+        }
+
+        let message_annotations = message.message_annotations().unwrap();
+
+        let key: AnnonationKey = "test".into();
+        assert_eq!(Value::from("test"), message_annotations[&key]);
+
+        let key = AnnonationKey::Long(100_000);
+        assert_eq!(Value::from(100_000i64), message_annotations[&key]);
+    }
+    #[test]
+    fn test_message_body_250() {
+        let content = include_bytes!("fixtures/message_body_250");
+
+        let (remaining, message) = Message::decode(content).unwrap();
+
+        assert!(remaining.is_empty());
+
+        assert!(message.body().data().unwrap().len() <= u8::MAX as usize);
+    }
+
+    #[test]
+    fn test_message_body_700() {
+        let content = include_bytes!("fixtures/message_body_700");
+
+        let (remaining, message) = Message::decode(content).unwrap();
+
+        assert!(remaining.is_empty());
+
+        assert!(message.body().data().unwrap().len() > u8::MAX as usize);
+    }
+    fn check_message_encode_decode(message: &Message) {
+        let mut buf = vec![];
+        message.encode(&mut buf).unwrap();
+
+        let (remaining, decoded) = match Message::decode(&buf) {
+            Ok(result) => result,
+            Err(err) => panic!(
+                "Failed to decode message {:#?} with error : {:?}",
+                message, err
+            ),
+        };
+
+        assert_eq!(0, remaining.len());
+        assert_eq!(message, &decoded);
+    }
+}
diff --git a/protocol/src/message/amqp/mod.rs b/protocol/src/message/amqp/mod.rs
new file mode 100644
index 0000000..6c6eca7
--- /dev/null
+++ b/protocol/src/message/amqp/mod.rs
@@ -0,0 +1,62 @@
+mod body;
+mod codec;
+mod error;
+mod header;
+mod message;
+mod properties;
+mod section;
+mod types;
+
+pub use codec::{AmqpDecoder, AmqpEncoder};
+pub use error::{AmqpDecodeError, AmqpEncodeError};
+
+pub use header::Header;
+pub use message::Message as AmqpMessage;
+pub use properties::Properties;
+pub use types::{
+    Address, AnnonationKey, Annotations, ApplicationProperties, DeliveryAnnotations, Footer, Map,
+    MessageAnnotations, MessageId, SequenceNo, SimpleValue, Str, Symbol, Timestamp, Value,
+};
+
+#[cfg(test)]
+mod tests {
+
+    use std::{fmt::Debug, ops::Range};
+
+    use pretty_assertions::assert_eq;
+
+    use fake::{Dummy, Fake, Faker};
+
+    const DEFAULT_LEN_RANGE: Range<usize> = 0..10;
+
+    use super::{AmqpDecoder, AmqpEncoder};
+    pub(crate) fn type_encode_decode_test<T>()
+    where
+        T: Dummy<Faker> + AmqpDecoder + AmqpEncoder + Debug + PartialEq + Debug,
+    {
+        let mut buffer = vec![];
+
+        let value: T = Faker.fake();
+
+        let _ = value.encode(&mut buffer).unwrap();
+
+        let (remaining, decoded) = T::decode(&buffer).unwrap();
+
+        assert_eq!(value, decoded);
+
+        assert!(remaining.is_empty());
+
+        assert_eq!(buffer.len() as u32, value.encoded_size(), "{:#?}", value);
+    }
+    pub(crate) fn type_encode_decode_test_fuzzy<T>()
+    where
+        T: Dummy<Faker> + AmqpDecoder + AmqpEncoder + Debug + PartialEq,
+    {
+        let mut rng = rand::thread_rng();
+        let len: usize = DEFAULT_LEN_RANGE.fake_with_rng(&mut rng);
+
+        for _ in 0..len {
+            type_encode_decode_test::<T>()
+        }
+    }
+}
diff --git a/protocol/src/message/amqp/properties.rs b/protocol/src/message/amqp/properties.rs
new file mode 100644
index 0000000..2c8141b
--- /dev/null
+++ b/protocol/src/message/amqp/properties.rs
@@ -0,0 +1,140 @@
+use byteorder::{BigEndian, WriteBytesExt};
+
+use crate::message::amqp::codec::constants::TypeCode;
+use crate::message::amqp::codec::constants::MESSAGE_PROPERTIES;
+use crate::message::amqp::types::list_decoder;
+use crate::message::amqp::types::Binary;
+use crate::message::amqp::types::Descriptor;
+use crate::message::amqp::types::List;
+use crate::message::amqp::types::Str;
+use crate::message::amqp::AmqpDecodeError;
+
+use super::AmqpDecoder;
+use super::{
+    types::{Address, MessageId, SequenceNo, Symbol, Timestamp},
+    AmqpEncodeError, AmqpEncoder,
+};
+
+#[cfg(test)]
+use fake::Fake;
+
+/// Properties of the message
+#[derive(Clone, Debug, PartialEq, Default)]
+#[cfg_attr(test, derive(fake::Dummy))]
+pub struct Properties {
+    pub message_id: Option<MessageId>,
+    pub user_id: Option<Vec<u8>>,
+    pub to: Option<Address>,
+    pub subject: Option<String>,
+    pub reply_to: Option<Address>,
+    pub correlation_id: Option<MessageId>,
+    pub content_type: Option<Symbol>,
+    pub content_encoding: Option<Symbol>,
+    pub absolute_expiry_time: Option<Timestamp>,
+    pub creation_time: Option<Timestamp>,
+    pub group_id: Option<String>,
+    pub group_sequence: Option<SequenceNo>,
+    pub reply_to_group_id: Option<String>,
+}
+
+impl Properties {
+    fn content_size(&self) -> u32 {
+        self.message_id.encoded_size()
+            + self.user_id.encoded_size()
+            + self.to.encoded_size()
+            + self.subject.encoded_size()
+            + self.reply_to.encoded_size()
+            + self.correlation_id.encoded_size()
+            + self.content_type.encoded_size()
+            + self.content_encoding.encoded_size()
+            + self.absolute_expiry_time.encoded_size()
+            + self.creation_time.encoded_size()
+            + self.group_id.encoded_size()
+            + self.group_sequence.encoded_size()
+            + self.reply_to_group_id.encoded_size()
+    }
+}
+impl AmqpEncoder for Properties {
+    fn encoded_size(&self) -> u32 {
+        let size = MESSAGE_PROPERTIES.encoded_size() + self.content_size();
+
+        let header = if size > u8::MAX as u32 { 9 } else { 3 };
+
+        header + size
+    }
+
+    fn encode(&self, writer: &mut impl std::io::Write) -> Result<(), AmqpEncodeError> {
+        MESSAGE_PROPERTIES.encode(writer)?;
+
+        let content_size = self.content_size();
+
+        if content_size + 1 > u8::MAX as u32 {
+            TypeCode::List8.encode(writer)?;
+            writer.write_u32::<BigEndian>(content_size + 4)?;
+            writer.write_u32::<BigEndian>(13)?;
+        } else {
+            TypeCode::List8.encode(writer)?;
+            writer.write_u8((content_size + 1) as u8)?;
+            writer.write_u8(13)?;
+        }
+        self.message_id.encode(writer)?;
+        self.user_id.encode(writer)?;
+        self.to.encode(writer)?;
+        self.subject.encode(writer)?;
+        self.reply_to.encode(writer)?;
+        self.correlation_id.encode(writer)?;
+        self.content_type.encode(writer)?;
+        self.content_encoding.encode(writer)?;
+        self.absolute_expiry_time.encode(writer)?;
+        self.creation_time.encode(writer)?;
+        self.group_id.encode(writer)?;
+        self.group_sequence.encode(writer)?;
+        self.reply_to_group_id.encode(writer)?;
+
+        Ok(())
+    }
+}
+impl AmqpDecoder for Properties {
+    fn decode(input: &[u8]) -> Result<(&[u8], Self), super::AmqpDecodeError> {
+        match Descriptor::decode(input)? {
+            (input, MESSAGE_PROPERTIES) => {
+                let properties = Properties::default();
+                List::decode_with_fields(input, list_decoder_properties, properties)
+            }
+            (_, descriptor) => Err(AmqpDecodeError::MessageParse(format!(
+                "Invalid descriptor for properties {:?}",
+                descriptor
+            ))),
+        }
+    }
+}
+
+list_decoder!(Properties, list_decoder_properties,
+    {
+     0 => { message_id, MessageId, None, true},
+     1 => { user_id, Binary, None, true},
+     2 => { to, Address, None, true},
+     3 => { subject, Str, None, true},
+     4 => { reply_to, Address, None, true},
+     5 => { correlation_id, MessageId, None, true},
+     6 => { content_type, Symbol, None, true},
+     7 => { content_encoding, Symbol, None, true},
+     8 => { absolute_expiry_time, Timestamp, None, true},
+     9 => { creation_time, Timestamp, None, true},
+     10 => { group_id, Str, None, true},
+     11 => { group_sequence, SequenceNo, None, true},
+     12 => { reply_to_group_id, Str, None, true}
+    }
+);
+
+#[cfg(test)]
+mod tests {
+    use crate::message::amqp::tests::type_encode_decode_test_fuzzy;
+
+    use super::Properties;
+
+    #[test]
+    fn test_properties_encode_decode() {
+        type_encode_decode_test_fuzzy::<Properties>()
+    }
+}
diff --git a/protocol/src/message/amqp/section.rs b/protocol/src/message/amqp/section.rs
new file mode 100644
index 0000000..732821a
--- /dev/null
+++ b/protocol/src/message/amqp/section.rs
@@ -0,0 +1,62 @@
+use crate::{
+    message::amqp::codec::constants::MESSAGE_BODY_DATA,
+    message::amqp::{types::Binary, AmqpDecoder},
+    utils::TupleMapperSecond,
+};
+
+use super::{
+    codec::constants::{
+        MESSAGE_ANNOTATIONS, MESSAGE_APPLICATION_PROPERTIES, MESSAGE_BODY_SEQUENCE,
+        MESSAGE_BODY_VALUE, MESSAGE_DELIVERY_ANNOTATIONS, MESSAGE_FOOTER, MESSAGE_HEADER,
+        MESSAGE_PROPERTIES,
+    },
+    header::Header,
+    properties::Properties,
+    types::{
+        AmqpSequence, AmqpValue, ApplicationProperties, DeliveryAnnotations, Descriptor, Footer,
+        MessageAnnotations, Value,
+    },
+    AmqpDecodeError,
+};
+
+#[derive(Clone, Debug, PartialEq)]
+#[allow(clippy::large_enum_variant)]
+pub enum MessageSection {
+    Header(Header),
+    DeliveryAnnotations(DeliveryAnnotations),
+    MessageAnnotations(MessageAnnotations),
+    ApplicationProperties(ApplicationProperties),
+    Data(Binary),
+    AmqpSequence(AmqpSequence),
+    AmqpValue(AmqpValue),
+    Footer(Footer),
+    Properties(Properties),
+}
+
+impl AmqpDecoder for MessageSection {
+    fn decode(input: &[u8]) -> Result<(&[u8], Self), AmqpDecodeError> {
+        let (remaining, described) = Descriptor::decode(input)?;
+
+        match described {
+            MESSAGE_HEADER => Header::decode(input).map_second(MessageSection::Header),
+            MESSAGE_BODY_DATA => Binary::decode(remaining).map_second(MessageSection::Data),
+            MESSAGE_PROPERTIES => Properties::decode(input).map_second(MessageSection::Properties),
+            MESSAGE_BODY_VALUE => Value::decode(remaining).map_second(MessageSection::AmqpValue),
+            MESSAGE_APPLICATION_PROPERTIES => ApplicationProperties::decode(remaining)
+                .map_second(MessageSection::ApplicationProperties),
+            MESSAGE_ANNOTATIONS => {
+                MessageAnnotations::decode(remaining).map_second(MessageSection::MessageAnnotations)
+            }
+            MESSAGE_BODY_SEQUENCE => {
+                AmqpSequence::decode(remaining).map_second(MessageSection::AmqpSequence)
+            }
+            MESSAGE_DELIVERY_ANNOTATIONS => DeliveryAnnotations::decode(remaining)
+                .map_second(MessageSection::DeliveryAnnotations),
+            MESSAGE_FOOTER => Footer::decode(remaining).map_second(MessageSection::Footer),
+            descriptor => Err(AmqpDecodeError::MessageParse(format!(
+                "Invalid section {:?}",
+                descriptor
+            ))),
+        }
+    }
+}
diff --git a/protocol/src/message/amqp/types/annotations.rs b/protocol/src/message/amqp/types/annotations.rs
new file mode 100644
index 0000000..6fad44f
--- /dev/null
+++ b/protocol/src/message/amqp/types/annotations.rs
@@ -0,0 +1,85 @@
+use crate::{
+    message::amqp::{codec::constants::TypeCode, AmqpDecoder, AmqpEncoder},
+    utils::TupleMapperSecond,
+};
+
+use super::{Long, Map, Symbol, ULong, Value};
+#[cfg(test)]
+use fake::Fake;
+
+/// Key for annotations [`Map`]
+#[derive(Debug, Clone, Hash, PartialEq, Eq)]
+#[cfg_attr(test, derive(fake::Dummy))]
+pub enum AnnonationKey {
+    Symbol(Symbol),
+    ULong(ULong),
+    Long(Long),
+}
+
+pub type Annotations = Map<AnnonationKey, Value>;
+
+impl Annotations {
+    pub fn put<K, V>(&mut self, key: K, value: V) -> Option<Value>
+    where
+        K: Into<AnnonationKey>,
+        V: Into<Value>,
+    {
+        self.0.insert(key.into(), value.into())
+    }
+}
+
+impl From<&str> for AnnonationKey {
+    fn from(string: &str) -> Self {
+        AnnonationKey::Symbol(Symbol::from(string))
+    }
+}
+
+impl From<String> for AnnonationKey {
+    fn from(string: String) -> Self {
+        AnnonationKey::Symbol(Symbol::from(string))
+    }
+}
+impl From<u64> for AnnonationKey {
+    fn from(number: u64) -> Self {
+        AnnonationKey::ULong(number)
+    }
+}
+
+impl AmqpDecoder for AnnonationKey {
+    fn decode(input: &[u8]) -> Result<(&[u8], Self), crate::message::amqp::AmqpDecodeError> {
+        let (_, code) = TypeCode::decode(input)?;
+
+        match code {
+            TypeCode::Symbol8 | TypeCode::Symbol32 => {
+                Symbol::decode(input).map_second(AnnonationKey::Symbol)
+            }
+            TypeCode::ULong | TypeCode::ULong0 | TypeCode::ULongSmall => {
+                ULong::decode(input).map_second(AnnonationKey::ULong)
+            }
+            TypeCode::Long | TypeCode::LongSmall => {
+                Long::decode(input).map_second(AnnonationKey::Long)
+            }
+            _ => Err(Self::invalid_type_code(code)),
+        }
+    }
+}
+impl AmqpEncoder for AnnonationKey {
+    fn encoded_size(&self) -> u32 {
+        match self {
+            AnnonationKey::Symbol(symbol) => symbol.encoded_size(),
+            AnnonationKey::ULong(long) => long.encoded_size(),
+            AnnonationKey::Long(long) => long.encoded_size(),
+        }
+    }
+
+    fn encode(
+        &self,
+        writer: &mut impl std::io::Write,
+    ) -> Result<(), crate::message::amqp::AmqpEncodeError> {
+        match self {
+            AnnonationKey::Symbol(symbol) => symbol.encode(writer),
+            AnnonationKey::ULong(long) => long.encode(writer),
+            AnnonationKey::Long(long) => long.encode(writer),
+        }
+    }
+}
diff --git a/protocol/src/message/amqp/types/definitions.rs b/protocol/src/message/amqp/types/definitions.rs
new file mode 100644
index 0000000..16194d0
--- /dev/null
+++ b/protocol/src/message/amqp/types/definitions.rs
@@ -0,0 +1,19 @@
+use super::{annotations::Annotations, List, Map, SimpleValue, Str, Value};
+
+pub type SequenceNo = u32;
+pub type Address = Str;
+
+/// Footer type definition [`Annotations`]
+pub type Footer = Annotations;
+
+/// Delivery annotations type definition [`Annotations`]
+pub type DeliveryAnnotations = Annotations;
+
+/// Message annotations type definition [`Annotations`]
+pub type MessageAnnotations = Annotations;
+
+/// Application properties type definition [`Map`]
+pub type ApplicationProperties = Map<Str, SimpleValue>;
+
+pub type AmqpSequence = List;
+pub type AmqpValue = Value;
diff --git a/protocol/src/message/amqp/types/descriptor.rs b/protocol/src/message/amqp/types/descriptor.rs
new file mode 100644
index 0000000..3b8e971
--- /dev/null
+++ b/protocol/src/message/amqp/types/descriptor.rs
@@ -0,0 +1,53 @@
+use super::primitives::ULong;
+use super::symbol::Symbol;
+use crate::message::amqp::codec::constants::TypeCode;
+use crate::message::amqp::codec::{AmqpDecoder, AmqpEncoder};
+use crate::message::amqp::error::{AmqpDecodeError, AmqpEncodeError};
+use crate::utils::TupleMapperSecond;
+
+#[cfg(test)]
+use fake::Fake;
+
+#[derive(Debug, PartialEq, Eq, Clone, Hash)]
+#[cfg_attr(test, derive(fake::Dummy))]
+pub enum Descriptor {
+    Ulong(u64),
+    Symbol(Symbol),
+}
+
+impl AmqpEncoder for Descriptor {
+    fn encoded_size(&self) -> u32 {
+        match *self {
+            Descriptor::Ulong(v) => 1 + v.encoded_size(),
+            Descriptor::Symbol(ref v) => 1 + v.encoded_size(),
+        }
+    }
+
+    fn encode(&self, writer: &mut impl std::io::Write) -> Result<(), AmqpEncodeError> {
+        TypeCode::Described.encode(writer)?;
+        match *self {
+            Descriptor::Ulong(v) => v.encode(writer),
+            Descriptor::Symbol(ref v) => v.encode(writer),
+        }
+    }
+}
+
+impl AmqpDecoder for Descriptor {
+    fn decode(input: &[u8]) -> Result<(&[u8], Self), AmqpDecodeError> {
+        match TypeCode::decode(input)? {
+            (input, TypeCode::Described) => {
+                let (remaining, code) = TypeCode::decode(input)?;
+                match code {
+                    TypeCode::ULong => ULong::decode(input).map_second(Descriptor::Ulong),
+                    TypeCode::ULong0 => Ok((remaining, Descriptor::Ulong(0))),
+                    TypeCode::ULongSmall => ULong::decode(input).map_second(Descriptor::Ulong),
+                    TypeCode::Symbol8 | TypeCode::Symbol32 => {
+                        Symbol::decode(input).map_second(Descriptor::Symbol)
+                    }
+                    _ => Err(Self::invalid_type_code(code)),
+                }
+            }
+            (_, code) => Err(Self::invalid_type_code(code)),
+        }
+    }
+}
diff --git a/protocol/src/message/amqp/types/mod.rs b/protocol/src/message/amqp/types/mod.rs
new file mode 100644
index 0000000..027c360
--- /dev/null
+++ b/protocol/src/message/amqp/types/mod.rs
@@ -0,0 +1,78 @@
+use crate::utils::TupleMapperSecond;
+pub use definitions::*;
+use derive_more::{From, TryInto};
+pub use descriptor::Descriptor;
+pub use primitives::*;
+
+use super::{codec::constants::TypeCode, AmqpDecoder, AmqpEncodeError, AmqpEncoder};
+
+#[cfg(test)]
+use fake::Fake;
+
+mod annotations;
+mod definitions;
+mod descriptor;
+mod primitives;
+mod symbol;
+
+pub use annotations::{AnnonationKey, Annotations};
+pub use primitives::{SimpleValue, Value};
+pub use symbol::Symbol;
+
+pub use primitives::Str;
+
+pub type Milliseconds = u32;
+
+#[derive(Debug, Eq, PartialEq, Clone, From, TryInto)]
+#[try_into(owned, ref, ref_mut)]
+#[cfg_attr(test, derive(fake::Dummy))]
+pub enum MessageId {
+    ULong(ULong),
+    Long(Long),
+    Uuid(Uuid),
+    Binary(Binary),
+    String(Str),
+}
+
+impl AmqpEncoder for MessageId {
+    fn encoded_size(&self) -> u32 {
+        match self {
+            MessageId::ULong(id) => id.encoded_size(),
+            MessageId::Uuid(id) => id.encoded_size(),
+            MessageId::Binary(id) => id.encoded_size(),
+            MessageId::String(id) => id.encoded_size(),
+            MessageId::Long(id) => id.encoded_size(),
+        }
+    }
+
+    fn encode(&self, writer: &mut impl std::io::Write) -> Result<(), AmqpEncodeError> {
+        match self {
+            MessageId::ULong(id) => id.encode(writer),
+            MessageId::Uuid(id) => id.encode(writer),
+            MessageId::Binary(id) => id.encode(writer),
+            MessageId::String(id) => id.encode(writer),
+            MessageId::Long(id) => id.encode(writer),
+        }
+    }
+}
+
+impl AmqpDecoder for MessageId {
+    fn decode(input: &[u8]) -> Result<(&[u8], Self), super::AmqpDecodeError> {
+        let (_, code) = TypeCode::decode(input)?;
+
+        match code {
+            TypeCode::ULong0 | TypeCode::ULong | TypeCode::ULongSmall => {
+                ULong::decode(input).map_second(MessageId::ULong)
+            }
+            TypeCode::Long | TypeCode::LongSmall => Long::decode(input).map_second(MessageId::Long),
+            TypeCode::Uuid => Uuid::decode(input).map_second(MessageId::Uuid),
+            TypeCode::Binary8 | TypeCode::Binary32 => {
+                Binary::decode(input).map_second(MessageId::Binary)
+            }
+            TypeCode::String8 | TypeCode::String32 => {
+                Str::decode(input).map_second(MessageId::String)
+            }
+            _ => Err(Self::invalid_type_code(code)),
+        }
+    }
+}
diff --git a/protocol/src/message/amqp/types/primitives/list.rs b/protocol/src/message/amqp/types/primitives/list.rs
new file mode 100644
index 0000000..ed23d3a
--- /dev/null
+++ b/protocol/src/message/amqp/types/primitives/list.rs
@@ -0,0 +1,199 @@
+use crate::{
+    codec::decoder::{read_u32, read_u8},
+    message::amqp::{
+        codec::constants::TypeCode, AmqpDecodeError, AmqpDecoder, AmqpEncodeError, AmqpEncoder,
+    },
+    utils::TupleMapperSecond,
+};
+
+#[derive(Debug, PartialEq, Eq, Clone, Hash, Default)]
+pub struct List(pub Vec<Value>);
+
+impl List {
+    pub fn new() -> Self {
+        Self(vec![])
+    }
+    pub fn push(&mut self, elem: impl Into<Value>) {
+        self.0.push(elem.into())
+    }
+
+    fn content_size(&self) -> u32 {
+        self.0.iter().fold(0, |acc, item| acc + item.encoded_size())
+    }
+}
+
+impl AmqpEncoder for List {
+    fn encoded_size(&self) -> u32 {
+        let content_size = self.content_size();
+
+        let header = if content_size + 1 > u8::MAX as u32 {
+            9
+        } else {
+            3
+        };
+
+        header + content_size
+    }
+
+    fn encode(&self, writer: &mut impl std::io::Write) -> Result<(), AmqpEncodeError> {
+        let content_size = self.content_size();
+
+        if content_size + 1 > u8::MAX as u32 {
+            TypeCode::List8.encode(writer)?;
+            writer.write_u32::<BigEndian>(content_size + 4)?;
+            writer.write_u32::<BigEndian>(self.0.len() as u32)?;
+        } else {
+            TypeCode::List8.encode(writer)?;
+            writer.write_u8((content_size + 1) as u8)?;
+            writer.write_u8(self.0.len() as u8)?;
+        }
+
+        for item in &self.0 {
+            item.encode(writer)?;
+        }
+        Ok(())
+    }
+}
+
+impl AmqpDecoder for List {
+    fn decode(input: &[u8]) -> Result<(&[u8], Self), AmqpDecodeError> {
+        let (input, code) = TypeCode::decode(input)?;
+
+        let (mut input, count) = match code {
+            TypeCode::List0 => return Ok((input, List::new())),
+            TypeCode::List8 => {
+                let (input, _) = read_u8(input)?;
+                read_u8(input)
+                    .map_second(u32::from)
+                    .map_err(AmqpDecodeError::from)?
+            }
+            TypeCode::List32 => {
+                let (input, _) = read_u32(input)?;
+                read_u32(input).map_err(AmqpDecodeError::from)?
+            }
+            _ => return Err(Self::invalid_type_code(code)),
+        };
+
+        let mut list = List::new();
+
+        for _ in 0..count {
+            let (input_inner, elem) = Value::decode(input)?;
+            list.0.push(elem);
+            input = input_inner;
+        }
+
+        Ok((input, list))
+    }
+}
+
+impl List {
+    pub fn decode_with_fields<'a, T, F>(
+        input: &'a [u8],
+        f: F,
+        mut dest: T,
+    ) -> Result<(&'a [u8], T), AmqpDecodeError>
+    where
+        F: Fn(&'a [u8], u32, &mut T) -> Result<&'a [u8], AmqpDecodeError>,
+    {
+        let (input, code) = TypeCode::decode(input)?;
+
+        let (mut input, count) = match code {
+            TypeCode::List0 => return Ok((input, dest)),
+            TypeCode::List8 => {
+                let (input, _) = read_u8(input)?;
+                read_u8(input).map_second(u32::from)?
+            }
+            TypeCode::List32 => {
+                let (input, _) = read_u32(input)?;
+                read_u32(input)?
+            }
+
+            _ => {
+                return Err(AmqpDecodeError::MessageParse(format!(
+                    "Invalid type code {:?} for list",
+                    code
+                )))
+            }
+        };
+
+        for idx in 0..count {
+            let input1 = f(input, idx as u32, &mut dest)?;
+            input = input1;
+        }
+        Ok((input, dest))
+    }
+}
+#[macro_export(local_inner_macros)]
+macro_rules! list_decoder {
+    ($ty:ty, $fname:ident, { $($key:expr => $b:tt),* }) => {
+        fn $fname<'a>(
+            input: &'a [u8],
+            idx: u32,
+            dest: &mut $ty,
+        ) -> Result<&'a [u8], AmqpDecodeError> {
+
+            match idx {
+                 $($key => {
+                      list_decoder!(@internal dest, input, $b)
+                 })*
+                _ => {
+                    Ok(input)
+                }
+            }
+        }
+    };
+    (@internal $dest:ident, $input:ident, { $name:ident, $dt:ident, $def:expr }) => {
+         {
+             let (input, value) = Option::<$dt>::decode($input)?;
+             $dest.$name = value.unwrap_or($def);
+             Ok(input)
+         }
+    };
+    (@internal $dest:ident, $input:ident, { $name:ident, $dt:ident, $def:expr, $optional:expr }) => {
+        {
+            let (input, value) = Option::<$dt>::decode($input)?;
+            $dest.$name = value;
+            Ok(input)
+        }
+    }
+}
+
+use byteorder::{BigEndian, WriteBytesExt};
+pub use list_decoder;
+
+use super::Value;
+
+#[cfg(test)]
+mod tests {
+    use std::ops::Range;
+
+    use fake::{Dummy, Fake, Faker};
+    use rand::Rng;
+
+    use crate::message::amqp::{
+        tests::type_encode_decode_test_fuzzy,
+        types::{SimpleValue, Value},
+    };
+
+    use super::List;
+    const DEFAULT_LEN_RANGE: Range<usize> = 0..10;
+
+    impl Dummy<Faker> for List {
+        fn dummy_with_rng<R: Rng + ?Sized>(config: &Faker, rng: &mut R) -> Self {
+            let len: usize = DEFAULT_LEN_RANGE.fake_with_rng(rng);
+            let mut m = List::new();
+
+            for _ in 0..len {
+                let simple: SimpleValue = config.fake_with_rng(rng);
+
+                m.0.push(Value::Simple(simple));
+            }
+            m
+        }
+    }
+
+    #[test]
+    fn list_encode_decode_test() {
+        type_encode_decode_test_fuzzy::<List>()
+    }
+}
diff --git a/protocol/src/message/amqp/types/primitives/map.rs b/protocol/src/message/amqp/types/primitives/map.rs
new file mode 100644
index 0000000..246f7cc
--- /dev/null
+++ b/protocol/src/message/amqp/types/primitives/map.rs
@@ -0,0 +1,186 @@
+use std::{collections::HashMap, fmt::Debug, hash::Hash, ops::Deref};
+
+use byteorder::{BigEndian, WriteBytesExt};
+
+use crate::{
+    codec::decoder::{read_u32, read_u8},
+    message::amqp::{
+        codec::constants::TypeCode, AmqpDecodeError, AmqpDecoder, AmqpEncodeError, AmqpEncoder,
+    },
+    utils::TupleMapperSecond,
+};
+
+/// API for attaching additional metadata to messages
+#[derive(PartialEq, Eq, Clone, Debug)]
+pub struct Map<K, V>(pub(crate) HashMap<K, V>)
+where
+    K: Eq + PartialEq + Hash,
+    V: PartialEq;
+
+impl<K, V> Default for Map<K, V>
+where
+    K: Eq + PartialEq + Hash,
+    V: PartialEq,
+{
+    fn default() -> Self {
+        Self(Default::default())
+    }
+}
+
+impl<K, V> Map<K, V>
+where
+    K: Eq + PartialEq + Hash + AmqpEncoder,
+    V: PartialEq + AmqpEncoder,
+{
+    fn content_size(&self) -> u32 {
+        self.0
+            .iter()
+            .fold(0, |acc, (k, v)| acc + k.encoded_size() + v.encoded_size())
+    }
+}
+
+impl<K, V> Deref for Map<K, V>
+where
+    K: Eq + PartialEq + Hash + AmqpEncoder,
+    V: PartialEq + AmqpEncoder,
+{
+    type Target = HashMap<K, V>;
+
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
+impl<K, V> AmqpDecoder for Map<K, V>
+where
+    K: Eq + PartialEq + Hash + AmqpDecoder + Debug,
+    V: PartialEq + AmqpDecoder + Debug,
+{
+    fn decode(input: &[u8]) -> Result<(&[u8], Self), AmqpDecodeError> {
+        let (input, code) = TypeCode::decode(input)?;
+
+        let (mut input, mut count) = match code {
+            TypeCode::Map8 => {
+                let (input, _) = read_u8(input)?;
+                read_u8(input).map_second(u32::from)?
+            }
+            TypeCode::Map32 => {
+                let (input, _) = read_u32(input)?;
+                read_u32(input)?
+            }
+            _ => return Err(Self::invalid_type_code(code)),
+        };
+
+        let mut map = Map::default();
+
+        while count > 0 {
+            count -= 2;
+            let (input_inner, key) = K::decode(input)?;
+            let (input_inner, value) = V::decode(input_inner)?;
+            map.0.insert(key, value);
+            input = input_inner;
+        }
+
+        Ok((input, map))
+    }
+}
+
+impl<K, V> AmqpEncoder for Map<K, V>
+where
+    K: Eq + PartialEq + Hash + AmqpEncoder,
+    V: PartialEq + AmqpEncoder,
+{
+    fn encoded_size(&self) -> u32 {
+        let content_size = self.content_size();
+
+        let type_size = if content_size + 1 > u8::MAX as u32 {
+            9
+        } else {
+            3
+        };
+
+        content_size + type_size
+    }
+
+    fn encode(&self, writer: &mut impl std::io::Write) -> Result<(), AmqpEncodeError> {
+        let count = self.0.len() * 2;
+        let content_size = self.content_size();
+
+        if content_size + 1 > u8::MAX as u32 {
+            TypeCode::Map32.encode(writer)?;
+            writer.write_u32::<BigEndian>(content_size + 4)?;
+            writer.write_u32::<BigEndian>(count as u32)?;
+        } else {
+            TypeCode::Map8.encode(writer)?;
+            writer.write_u8((content_size + 1) as u8)?;
+            writer.write_u8(count as u8)?;
+        }
+
+        for (k, v) in &self.0 {
+            k.encode(writer)?;
+            v.encode(writer)?;
+        }
+        Ok(())
+    }
+}
+#[cfg(test)]
+mod tests {
+
+    use crate::message::amqp::tests::type_encode_decode_test_fuzzy;
+
+    use super::Map;
+
+    use crate::message::amqp::types::{SimpleValue, Value};
+
+    use fake::{Dummy, Fake, Faker};
+    use rand::Rng;
+
+    use std::{hash::Hash, ops::Range};
+
+    const DEFAULT_LEN_RANGE: Range<usize> = 0..10;
+
+    impl<K> Dummy<Faker> for Map<K, Value>
+    where
+        K: Dummy<Faker> + Hash + Eq,
+    {
+        fn dummy_with_rng<R: Rng + ?Sized>(config: &Faker, rng: &mut R) -> Self {
+            let len: usize = DEFAULT_LEN_RANGE.fake_with_rng(rng);
+            let mut m = Map::default();
+
+            for _ in 0..len {
+                let simple: SimpleValue = config.fake_with_rng(rng);
+                m.0.insert(config.fake_with_rng(rng), Value::Simple(simple));
+            }
+            m
+        }
+
+        fn dummy(config: &Faker) -> Self {
+            let mut r = rand::thread_rng();
+            Dummy::<Faker>::dummy_with_rng(config, &mut r)
+        }
+    }
+
+    impl<K> Dummy<Faker> for Map<K, SimpleValue>
+    where
+        K: Dummy<Faker> + Hash + Eq,
+    {
+        fn dummy_with_rng<R: Rng + ?Sized>(config: &Faker, rng: &mut R) -> Self {
+            let len: usize = DEFAULT_LEN_RANGE.fake_with_rng(rng);
+            let mut m = Map::default();
+
+            for _ in 0..len {
+                let simple: SimpleValue = config.fake_with_rng(rng);
+                m.0.insert(config.fake_with_rng(rng), simple);
+            }
+            m
+        }
+
+        fn dummy(config: &Faker) -> Self {
+            let mut r = rand::thread_rng();
+            Dummy::<Faker>::dummy_with_rng(config, &mut r)
+        }
+    }
+    #[test]
+    fn map_encode_decode_test() {
+        type_encode_decode_test_fuzzy::<Map<SimpleValue, Value>>()
+    }
+}
diff --git a/protocol/src/message/amqp/types/primitives/mod.rs b/protocol/src/message/amqp/types/primitives/mod.rs
new file mode 100644
index 0000000..6fe7b11
--- /dev/null
+++ b/protocol/src/message/amqp/types/primitives/mod.rs
@@ -0,0 +1,9 @@
+mod list;
+mod map;
+mod simple;
+mod value;
+
+pub use list::{list_decoder, List};
+pub use map::Map;
+pub use simple::*;
+pub use value::*;
diff --git a/protocol/src/message/amqp/types/primitives/simple.rs b/protocol/src/message/amqp/types/primitives/simple.rs
new file mode 100644
index 0000000..8a00a21
--- /dev/null
+++ b/protocol/src/message/amqp/types/primitives/simple.rs
@@ -0,0 +1,707 @@
+use byteorder::{BigEndian, WriteBytesExt};
+use chrono::{DateTime, TimeZone, Utc};
+use derive_more::{From, TryInto};
+use ordered_float::OrderedFloat;
+
+use crate::{
+    codec::decoder::{
+        read_exact, read_f32, read_f64, read_i16, read_i32, read_i64, read_i8, read_u16, read_u32,
+        read_u64, read_u8,
+    },
+    message::amqp::{
+        codec::{constants::TypeCode, AmqpEncoder},
+        error::AmqpEncodeError,
+        types::Symbol,
+        AmqpDecodeError, AmqpDecoder,
+    },
+    utils::TupleMapperSecond,
+};
+
+#[cfg(test)]
+use fake::Fake;
+
+/// Primitive AMQP 1.0 data type
+#[derive(Debug, Eq, PartialEq, Hash, Clone, From, TryInto)]
+#[try_into(owned, ref, ref_mut)]
+#[cfg_attr(test, derive(fake::Dummy))]
+pub enum SimpleValue {
+    Null,
+    Boolean(Boolean),
+    Ubyte(UByte),
+    Ushort(UShort),
+    Uint(UInt),
+    Ulong(ULong),
+    Byte(Byte),
+    Short(Short),
+    Int(Int),
+    Long(Long),
+    Float(Float),
+    Double(Double),
+    Char(Char),
+    Timestamp(Timestamp),
+    Uuid(Uuid),
+    Binary(Binary),
+    String(Str),
+    Symbol(Symbol),
+}
+
+impl From<&str> for SimpleValue {
+    fn from(string: &str) -> Self {
+        SimpleValue::String(string.into())
+    }
+}
+impl AmqpEncoder for SimpleValue {
+    fn encoded_size(&self) -> u32 {
+        match self {
+            SimpleValue::Null => 1,
+            SimpleValue::Boolean(value) => value.encoded_size(),
+            SimpleValue::Ubyte(value) => value.encoded_size(),
+            SimpleValue::Ushort(value) => value.encoded_size(),
+            SimpleValue::Uint(value) => value.encoded_size(),
+            SimpleValue::Ulong(value) => value.encoded_size(),
+            SimpleValue::Byte(value) => value.encoded_size(),
+            SimpleValue::Short(value) => value.encoded_size(),
+            SimpleValue::Int(value) => value.encoded_size(),
+            SimpleValue::Long(value) => value.encoded_size(),
+            SimpleValue::Float(value) => value.encoded_size(),
+            SimpleValue::Double(value) => value.encoded_size(),
+            SimpleValue::Char(value) => value.encoded_size(),
+            SimpleValue::Timestamp(value) => value.encoded_size(),
+            SimpleValue::Uuid(value) => value.encoded_size(),
+            SimpleValue::Binary(value) => value.encoded_size(),
+            SimpleValue::String(value) => value.encoded_size(),
+            SimpleValue::Symbol(value) => value.encoded_size(),
+        }
+    }
+
+    fn encode(&self, writer: &mut impl std::io::Write) -> Result<(), AmqpEncodeError> {
+        match self {
+            SimpleValue::Null => TypeCode::Null.encode(writer),
+            SimpleValue::Boolean(value) => value.encode(writer),
+            SimpleValue::Ubyte(value) => value.encode(writer),
+            SimpleValue::Ushort(value) => value.encode(writer),
+            SimpleValue::Uint(value) => value.encode(writer),
+            SimpleValue::Ulong(value) => value.encode(writer),
+            SimpleValue::Byte(value) => value.encode(writer),
+            SimpleValue::Short(value) => value.encode(writer),
+            SimpleValue::Int(value) => value.encode(writer),
+            SimpleValue::Long(value) => value.encode(writer),
+            SimpleValue::Float(value) => value.encode(writer),
+            SimpleValue::Double(value) => value.encode(writer),
+            SimpleValue::Char(value) => value.encode(writer),
+            SimpleValue::Timestamp(value) => value.encode(writer),
+            SimpleValue::Uuid(value) => value.encode(writer),
+            SimpleValue::Binary(value) => value.encode(writer),
+            SimpleValue::String(value) => value.encode(writer),
+            SimpleValue::Symbol(value) => value.encode(writer),
+        }
+    }
+}
+
+impl AmqpDecoder for SimpleValue {
+    fn decode(input: &[u8]) -> Result<(&[u8], Self), AmqpDecodeError> {
+        let (remaining, code) = TypeCode::decode(input)?;
+        match code {
+            TypeCode::Null => Ok((remaining, SimpleValue::Null)),
+            TypeCode::Boolean | TypeCode::BooleanFalse | TypeCode::BooleanTrue => {
+                Boolean::decode(input).map_second(SimpleValue::Boolean)
+            }
+            TypeCode::UInt0 | TypeCode::UIntSmall | TypeCode::UInt => {
+                UInt::decode(input).map_second(SimpleValue::Uint)
+            }
+            TypeCode::ULong0 | TypeCode::ULongSmall | TypeCode::ULong => {
+                ULong::decode(input).map_second(SimpleValue::Ulong)
+            }
+            TypeCode::UByte => UByte::decode(input).map_second(SimpleValue::Ubyte),
+            TypeCode::UShort => UShort::decode(input).map_second(SimpleValue::Ushort),
+            TypeCode::Byte => Byte::decode(input).map_second(SimpleValue::Byte),
+            TypeCode::Short => Short::decode(input).map_second(SimpleValue::Short),
+            TypeCode::IntSmall | TypeCode::Int => Int::decode(input).map_second(SimpleValue::Int),
+            TypeCode::LongSmall | TypeCode::Long => {
+                Long::decode(input).map_second(SimpleValue::Long)
+            }
+            TypeCode::Float => Float::decode(input).map_second(SimpleValue::Float),
+            TypeCode::Double => Double::decode(input).map_second(SimpleValue::Double),
+            TypeCode::Char => Char::decode(input).map_second(SimpleValue::Char),
+            TypeCode::Timestamp => Timestamp::decode(input).map_second(SimpleValue::Timestamp),
+            TypeCode::Uuid => Uuid::decode(input).map_second(SimpleValue::Uuid),
+            TypeCode::Binary8 | TypeCode::Binary32 => {
+                Binary::decode(input).map_second(SimpleValue::Binary)
+            }
+            TypeCode::String8 | TypeCode::String32 => {
+                Str::decode(input).map_second(SimpleValue::String)
+            }
+            TypeCode::Symbol8 | TypeCode::Symbol32 => {
+                Symbol::decode(input).map_second(SimpleValue::Symbol)
+            }
+            _ => Err(Self::invalid_type_code(code)),
+        }
+    }
+}
+
+pub type Boolean = bool;
+
+pub type UByte = u8;
+
+impl AmqpEncoder for UByte {
+    fn encoded_size(&self) -> u32 {
+        2
+    }
+
+    fn encode(&self, writer: &mut impl std::io::Write) -> Result<(), AmqpEncodeError> {
+        TypeCode::UByte.encode(writer)?;
+        writer.write_u8(*self).map(Ok)?
+    }
+}
+
+impl AmqpDecoder for UByte {
+    fn decode(input: &[u8]) -> Result<(&[u8], Self), AmqpDecodeError> {
+        match TypeCode::decode(input)? {
+            (input, TypeCode::UByte) => read_u8(input).map_err(AmqpDecodeError::from),
+            (_, code) => Err(Self::invalid_type_code(code)),
+        }
+    }
+}
+
+pub type UShort = u16;
+
+impl AmqpEncoder for UShort {
+    fn encoded_size(&self) -> u32 {
+        3
+    }
+
+    fn encode(&self, writer: &mut impl std::io::Write) -> Result<(), AmqpEncodeError> {
+        TypeCode::UShort.encode(writer)?;
+        writer.write_u16::<BigEndian>(*self)?;
+        Ok(())
+    }
+}
+
+impl AmqpDecoder for UShort {
+    fn decode(input: &[u8]) -> Result<(&[u8], Self), AmqpDecodeError> {
+        match TypeCode::decode(input)? {
+            (input, TypeCode::UShort) => read_u16(input).map_err(AmqpDecodeError::from),
+            (_, code) => Err(Self::invalid_type_code(code)),
+        }
+    }
+}
+pub type UInt = u32;
+
+impl AmqpEncoder for UInt {
+    fn encoded_size(&self) -> u32 {
+        if *self == 0 {
+            1
+        } else if *self > u32::from(u8::MAX) {
+            5
+        } else {
+            2
+        }
+    }
+
+    fn encode(&self, writer: &mut impl std::io::Write) -> Result<(), AmqpEncodeError> {
+        if *self == 0 {
+            TypeCode::UInt0.encode(writer).map(Ok)?
+        } else if *self > u32::from(u8::MAX) {
+            TypeCode::UInt.encode(writer)?;
+            writer.write_u32::<BigEndian>(*self).map(Ok)?
+        } else {
+            TypeCode::UIntSmall.encode(writer)?;
+            writer.write_u8(*self as u8).map(Ok)?
+        }
+    }
+}
+
+impl AmqpDecoder for UInt {
+    fn decode(input: &[u8]) -> Result<(&[u8], Self), AmqpDecodeError> {
+        let (input, code) = TypeCode::decode(input)?;
+        match code {
+            TypeCode::UInt0 => Ok((input, 0)),
+            TypeCode::UIntSmall => read_u8(input)
+                .map_second(UInt::from)
+                .map_err(AmqpDecodeError::from),
+            TypeCode::UInt => read_u32(input).map_err(AmqpDecodeError::from),
+            _ => Err(Self::invalid_type_code(code)),
+        }
+    }
+}
+
+pub type ULong = u64;
+
+impl AmqpEncoder for ULong {
+    fn encoded_size(&self) -> u32 {
+        if *self == 0 {
+            1
+        } else if *self > u64::from(u8::MAX) {
+            9
+        } else {
+            2
+        }
+    }
+
+    fn encode(&self, writer: &mut impl std::io::Write) -> Result<(), AmqpEncodeError> {
+        if *self == 0 {
+            TypeCode::ULong0.encode(writer).map(Ok)?
+        } else if *self > u64::from(u8::MAX) {
+            TypeCode::ULong.encode(writer)?;
+            writer.write_u64::<BigEndian>(*self).map(Ok)?
+        } else {
+            TypeCode::ULongSmall.encode(writer)?;
+            writer.write_u8(*self as u8).map(Ok)?
+        }
+    }
+}
+
+impl AmqpDecoder for ULong {
+    fn decode(input: &[u8]) -> Result<(&[u8], Self), AmqpDecodeError> {
+        match TypeCode::decode(input)? {
+            (input, TypeCode::ULong) => read_u64(input)
+                .map_second(ULong::from)
+                .map_err(AmqpDecodeError::from),
+            (input, TypeCode::ULongSmall) => read_u8(input)
+                .map_second(ULong::from)
+                .map_err(AmqpDecodeError::from),
+            (input, TypeCode::ULong0) => Ok((input, 0)),
+            (_, code) => Err(Self::invalid_type_code(code)),
+        }
+    }
+}
+
+pub type Byte = i8;
+
+impl AmqpDecoder for Byte {
+    fn decode(input: &[u8]) -> Result<(&[u8], Self), AmqpDecodeError> {
+        match TypeCode::decode(input)? {
+            (input, TypeCode::Byte) => read_i8(input).map_err(AmqpDecodeError::from),
+            (_, code) => Err(Self::invalid_type_code(code)),
+        }
+    }
+}
+impl AmqpEncoder for Byte {
+    fn encoded_size(&self) -> u32 {
+        2
+    }
+
+    fn encode(&self, writer: &mut impl std::io::Write) -> Result<(), AmqpEncodeError> {
+        TypeCode::Byte.encode(writer)?;
+        writer.write_i8(*self)?;
+        Ok(())
+    }
+}
+
+pub type Short = i16;
+
+impl AmqpDecoder for Short {
+    fn decode(input: &[u8]) -> Result<(&[u8], Self), AmqpDecodeError> {
+        let (input, code) = TypeCode::decode(input)?;
+        match code {
+            TypeCode::Short => read_i16(input).map_err(AmqpDecodeError::from),
+            _ => Err(Self::invalid_type_code(code)),
+        }
+    }
+}
+impl AmqpEncoder for Short {
+    fn encoded_size(&self) -> u32 {
+        3
+    }
+
+    fn encode(&self, writer: &mut impl std::io::Write) -> Result<(), AmqpEncodeError> {
+        TypeCode::Short.encode(writer)?;
+        writer.write_i16::<BigEndian>(*self)?;
+        Ok(())
+    }
+}
+
+pub type Int = i32;
+
+impl AmqpEncoder for Int {
+    fn encoded_size(&self) -> u32 {
+        if *self > i32::from(i8::MAX) || *self < i32::from(i8::MIN) {
+            5
+        } else {
+            2
+        }
+    }
+
+    fn encode(&self, writer: &mut impl std::io::Write) -> Result<(), AmqpEncodeError> {
+        if *self > i32::from(i8::MAX) || *self < i32::from(i8::MIN) {
+            TypeCode::Int.encode(writer)?;
+            writer.write_i32::<BigEndian>(*self)?;
+        } else {
+            TypeCode::IntSmall.encode(writer)?;
+            writer.write_i8(*self as i8)?;
+        }
+        Ok(())
+    }
+}
+impl AmqpDecoder for Int {
+    fn decode(input: &[u8]) -> Result<(&[u8], Self), AmqpDecodeError> {
+        let (input, code) = TypeCode::decode(input)?;
+        match code {
+            TypeCode::Int => read_i32(input).map_err(AmqpDecodeError::from),
+            TypeCode::IntSmall => read_u8(input)
+                .map_second(i32::from)
+                .map_err(AmqpDecodeError::from),
+            _ => Err(Self::invalid_type_code(code)),
+        }
+    }
+}
+
+pub type Long = i64;
+
+impl AmqpDecoder for Long {
+    fn decode(input: &[u8]) -> Result<(&[u8], Self), AmqpDecodeError> {
+        let (input, code) = TypeCode::decode(input)?;
+        match code {
+            TypeCode::Long => read_i64(input).map_err(AmqpDecodeError::from),
+            TypeCode::LongSmall => read_i8(input)
+                .map_second(i64::from)
+                .map_err(AmqpDecodeError::from),
+            _ => Err(Self::invalid_type_code(code)),
+        }
+    }
+}
+impl AmqpEncoder for Long {
+    fn encoded_size(&self) -> u32 {
+        if *self > i64::from(i8::MAX) || *self < i64::from(i8::MIN) {
+            9
+        } else {
+            3
+        }
+    }
+
+    fn encode(&self, writer: &mut impl std::io::Write) -> Result<(), AmqpEncodeError> {
+        if *self > i64::from(i8::MAX) || *self < i64::from(i8::MIN) {
+            TypeCode::Long.encode(writer)?;
+            writer.write_i64::<BigEndian>(*self)?;
+        } else {
+            TypeCode::LongSmall.encode(writer)?;
+            writer.write_i8(*self as i8)?;
+        }
+        Ok(())
+    }
+}
+
+#[derive(Debug, PartialEq, Hash, Eq, Clone)]
+pub struct Float(OrderedFloat<f32>);
+impl From<f32> for Float {
+    fn from(float: f32) -> Self {
+        Float(float.into())
+    }
+}
+
+impl AmqpDecoder for Float {
+    fn decode(input: &[u8]) -> Result<(&[u8], Self), AmqpDecodeError> {
+        let (input, code) = TypeCode::decode(input)?;
+        match code {
+            TypeCode::Float => read_f32(input)
+                .map_second(Float::from)
+                .map_err(AmqpDecodeError::from),
+            _ => Err(Self::invalid_type_code(code)),
+        }
+    }
+}
+impl AmqpEncoder for Float {
+    fn encoded_size(&self) -> u32 {
+        5
+    }
+
+    fn encode(&self, writer: &mut impl std::io::Write) -> Result<(), AmqpEncodeError> {
+        TypeCode::Float.encode(writer)?;
+        writer.write_f32::<BigEndian>(self.0 .0)?;
+        Ok(())
+    }
+}
+
+#[derive(Debug, PartialEq, Hash, Eq, Clone)]
+pub struct Double(OrderedFloat<f64>);
+
+impl From<f64> for Double {
+    fn from(double: f64) -> Self {
+        Double(double.into())
+    }
+}
+
+impl AmqpDecoder for Double {
+    fn decode(input: &[u8]) -> Result<(&[u8], Self), AmqpDecodeError> {
+        let (input, code) = TypeCode::decode(input)?;
+        match code {
+            TypeCode::Double => read_f64(input)
+                .map_second(Double::from)
+                .map_err(AmqpDecodeError::from),
+            _ => Err(Self::invalid_type_code(code)),
+        }
+    }
+}
+impl AmqpEncoder for Double {
+    fn encoded_size(&self) -> u32 {
+        9
+    }
+
+    fn encode(&self, writer: &mut impl std::io::Write) -> Result<(), AmqpEncodeError> {
+        TypeCode::Double.encode(writer)?;
+        writer.write_f64::<BigEndian>(self.0 .0)?;
+        Ok(())
+    }
+}
+
+pub type Char = char;
+
+impl AmqpDecoder for Char {
+    fn decode(input: &[u8]) -> Result<(&[u8], Self), AmqpDecodeError> {
+        let (input, code) = TypeCode::decode(input)?;
+
+        match code {
+            TypeCode::Char => read_u32(input)
+                .map_second(Char::from_u32)
+                .map_err(AmqpDecodeError::from)
+                .and_then(|(input, c)| {
+                    Ok((input, c.ok_or_else(|| Self::invalid_type_code(code))?))
+                }),
+            _ => Err(Self::invalid_type_code(code)),
+        }
+    }
+}
+impl AmqpEncoder for Char {
+    fn encoded_size(&self) -> u32 {
+        5
+    }
+
+    fn encode(&self, writer: &mut impl std::io::Write) -> Result<(), AmqpEncodeError> {
+        TypeCode::Char.encode(writer)?;
+        writer.write_u32::<BigEndian>(*self as u32)?;
+        Ok(())
+    }
+}
+
+pub type Uuid = uuid::Uuid;
+
+impl AmqpDecoder for Uuid {
+    fn decode(input: &[u8]) -> Result<(&[u8], Self), AmqpDecodeError> {
+        let (input, code) = TypeCode::decode(input)?;
+
+        match code {
+            TypeCode::Uuid => read_exact(input, 16)
+                .map_err(AmqpDecodeError::from)
+                .and_then(|(input, bytes)| Ok((input, Uuid::from_slice(bytes)?))),
+            _ => Err(Self::invalid_type_code(code)),
+        }
+    }
+}
+impl AmqpEncoder for Uuid {
+    fn encoded_size(&self) -> u32 {
+        TypeCode::Uuid.encoded_size() + 16
+    }
+
+    fn encode(&self, writer: &mut impl std::io::Write) -> Result<(), AmqpEncodeError> {
+        TypeCode::Uuid.encode(writer)?;
+        writer.write_all(self.as_bytes())?;
+        Ok(())
+    }
+}
+// Binary
+pub type Binary = Vec<u8>;
+
+impl<T: AmqpEncoder> AmqpEncoder for Option<T> {
+    fn encoded_size(&self) -> u32 {
+        self.as_ref().map(T::encoded_size).unwrap_or(1)
+    }
+
+    fn encode(&self, writer: &mut impl std::io::Write) -> Result<(), AmqpEncodeError> {
+        match self {
+            Some(value) => value.encode(writer),
+            None => TypeCode::Null.encode(writer),
+        }
+    }
+}
+
+impl<T: AmqpDecoder> AmqpDecoder for Option<T> {
+    fn decode(input: &[u8]) -> Result<(&[u8], Self), AmqpDecodeError> {
+        if let (input, TypeCode::Null) = TypeCode::decode(input)? {
+            Ok((input, None))
+        } else {
+            T::decode(input).map_second(Some)
+        }
+    }
+}
+
+impl AmqpEncoder for Boolean {
+    fn encoded_size(&self) -> u32 {
+        1
+    }
+
+    fn encode(&self, writer: &mut impl std::io::Write) -> Result<(), AmqpEncodeError> {
+        let code = if *self {
+            TypeCode::BooleanTrue
+        } else {
+            TypeCode::BooleanFalse
+        };
+        code.encode(writer)?;
+        Ok(())
+    }
+}
+
+impl AmqpDecoder for Boolean {
+    fn decode(input: &[u8]) -> Result<(&[u8], Self), AmqpDecodeError> {
+        let (input, code) = TypeCode::decode(input)?;
+
+        match code {
+            TypeCode::Boolean => read_u8(input)
+                .map_second(|boolean| boolean != 0)
+                .map_err(AmqpDecodeError::from),
+            TypeCode::BooleanTrue => Ok((input, true)),
+            TypeCode::BooleanFalse => Ok((input, false)),
+            _ => Err(Self::invalid_type_code(code)),
+        }
+    }
+}
+
+impl AmqpEncoder for Binary {
+    fn encoded_size(&self) -> u32 {
+        let len = self.len() as u32;
+        let size = if len > u8::MAX as u32 { 5 } else { 2 };
+        size + len
+    }
+
+    fn encode(&self, writer: &mut impl std::io::Write) -> Result<(), AmqpEncodeError> {
+        let len = self.len() as u32;
+        if len > u8::MAX as u32 {
+            TypeCode::Binary32.encode(writer)?;
+            writer.write_u32::<BigEndian>(len)?;
+        } else {
+            TypeCode::Binary8.encode(writer)?;
+            writer.write_u8(len as u8)?;
+        };
+        writer.write_all(self)?;
+        Ok(())
+    }
+}
+
+impl AmqpDecoder for Binary {
+    fn decode(input: &[u8]) -> Result<(&[u8], Self), AmqpDecodeError> {
+        let (input, code) = TypeCode::decode(input)?;
+        match code {
+            TypeCode::Binary8 => {
+                let (input, len) = read_u8(input)?;
+                read_exact(input, len as usize)
+                    .map_second(|bytes| bytes.to_vec())
+                    .map(Ok)?
+            }
+            TypeCode::Binary32 => {
+                let (input, len) = read_u32(input)?;
+                read_exact(input, len as usize)
+                    .map_second(|bytes| bytes.to_vec())
+                    .map(Ok)?
+            }
+            _ => Err(Self::invalid_type_code(code)),
+        }
+    }
+}
+
+pub type Str = String;
+
+impl AmqpEncoder for Str {
+    fn encoded_size(&self) -> u32 {
+        let length = self.len() as u32;
+        let size = if length > u8::MAX as u32 { 5 } else { 2 };
+        size + length
+    }
+
+    fn encode(&self, writer: &mut impl std::io::Write) -> Result<(), AmqpEncodeError> {
+        let length = self.len();
+        if length > u8::MAX as usize {
+            TypeCode::String32.encode(writer)?;
+            writer.write_u32::<BigEndian>(length as u32)?;
+        } else {
+            TypeCode::String8.encode(writer)?;
+            writer.write_u8(length as u8)?;
+        }
+
+        writer.write_all(self.as_bytes())?;
+        Ok(())
+    }
+}
+impl AmqpDecoder for Str {
+    fn decode(input: &[u8]) -> Result<(&[u8], Self), AmqpDecodeError> {
+        let (input, code) = TypeCode::decode(input)?;
+        match code {
+            TypeCode::String8 => {
+                let (input, len) = read_u8(input)?;
+                read_exact(input, len as usize)
+                    .map_second(|bytes| bytes.to_vec())
+                    .map(Ok)?
+                    .and_then(|(input, bytes)| Ok((input, String::from_utf8(bytes)?)))
+            }
+            TypeCode::String32 => {
+                let (input, len) = read_u32(input)?;
+                read_exact(input, len as usize)
+                    .map_second(|bytes| bytes.to_vec())
+                    .map(Ok)?
+                    .and_then(|(input, bytes)| Ok((input, String::from_utf8(bytes)?)))
+            }
+            _ => Err(Self::invalid_type_code(code)),
+        }
+    }
+}
+
+#[derive(Debug, PartialEq, Hash, Eq, Clone, From)]
+pub struct Timestamp(DateTime<Utc>);
+
+impl AmqpDecoder for Timestamp {
+    fn decode(input: &[u8]) -> Result<(&[u8], Self), AmqpDecodeError> {
+        match TypeCode::decode(input)? {
+            (input, TypeCode::Timestamp) => read_i64(input)
+                .map_second(|millis| Timestamp(Utc.timestamp_millis(millis)))
+                .map(Ok)?,
+            (_, code) => Err(Self::invalid_type_code(code)),
+        }
+    }
+}
+
+impl AmqpEncoder for Timestamp {
+    fn encoded_size(&self) -> u32 {
+        TypeCode::Timestamp.encoded_size() + 8
+    }
+
+    fn encode(&self, writer: &mut impl std::io::Write) -> Result<(), AmqpEncodeError> {
+        TypeCode::Timestamp.encode(writer)?;
+        writer.write_i64::<BigEndian>(self.0.timestamp_millis())?;
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+
+    use chrono::{DateTime, Timelike, Utc};
+    use fake::{Dummy, Fake, Faker};
+    use ordered_float::OrderedFloat;
+
+    use super::{Double, Float, Timestamp};
+    use crate::message::amqp::{tests::type_encode_decode_test_fuzzy, types::SimpleValue};
+
+    impl Dummy<Faker> for Float {
+        fn dummy_with_rng<R: rand::Rng + ?Sized>(config: &Faker, rng: &mut R) -> Self {
+            let num: f32 = config.fake_with_rng(rng);
+            Float(OrderedFloat::from(num))
+        }
+    }
+    impl Dummy<Faker> for Double {
+        fn dummy_with_rng<R: rand::Rng + ?Sized>(config: &Faker, rng: &mut R) -> Self {
+            let num: f64 = config.fake_with_rng(rng);
+            Double(OrderedFloat::from(num))
+        }
+    }
+
+    impl Dummy<Faker> for Timestamp {
+        fn dummy_with_rng<R: rand::Rng + ?Sized>(config: &Faker, rng: &mut R) -> Self {
+            let mut dt: DateTime<Utc> = config.fake_with_rng(rng);
+            dt = dt.with_nanosecond(0).unwrap();
+            Timestamp(dt)
+        }
+    }
+
+    #[test]
+    fn simple_value_encode_decode_test() {
+        type_encode_decode_test_fuzzy::<SimpleValue>()
+    }
+}
diff --git a/protocol/src/message/amqp/types/primitives/value.rs b/protocol/src/message/amqp/types/primitives/value.rs
new file mode 100644
index 0000000..8192ee3
--- /dev/null
+++ b/protocol/src/message/amqp/types/primitives/value.rs
@@ -0,0 +1,235 @@
+use std::{convert::TryFrom, hash::Hash};
+
+use super::{
+    Binary, Boolean, Byte, Char, Double, Float, Int, List, Long, Map, Short, SimpleValue, Str,
+    Timestamp, UByte, UInt, ULong, UShort, Uuid,
+};
+use crate::{
+    error::DecodeError,
+    message::amqp::{
+        codec::constants::TypeCode, types::Descriptor, AmqpDecodeError, AmqpDecoder, AmqpEncoder,
+        Symbol,
+    },
+    utils::TupleMapperSecond,
+};
+use derive_more::From;
+
+#[cfg(test)]
+use fake::Fake;
+
+/// AMQP 1.0 data types
+#[derive(Debug, Eq, PartialEq, Hash, Clone)]
+#[cfg_attr(test, derive(fake::Dummy))]
+pub enum Value {
+    Simple(SimpleValue),
+    Collection(CollectionValue),
+    Described(DescribedValue),
+}
+
+#[derive(Debug, Eq, PartialEq, Hash, Clone)]
+#[cfg_attr(test, derive(fake::Dummy))]
+pub struct DescribedValue {
+    descriptor: Descriptor,
+    value: Box<Value>,
+}
+
+impl AmqpEncoder for DescribedValue {
+    fn encoded_size(&self) -> u32 {
+        1 + self.descriptor.encoded_size() + self.value.encoded_size()
+    }
+
+    fn encode(
+        &self,
+        writer: &mut impl std::io::Write,
+    ) -> Result<(), crate::message::amqp::AmqpEncodeError> {
+        TypeCode::Described.encode(writer)?;
+        self.descriptor.encode(writer)?;
+        self.value.encode(writer)?;
+        Ok(())
+    }
+}
+
+impl AmqpDecoder for DescribedValue {
+    fn decode(input: &[u8]) -> Result<(&[u8], Self), AmqpDecodeError> {
+        let (input, code) = TypeCode::decode(input)?;
+
+        match code {
+            TypeCode::Described => {
+                let (input, descriptor) = Descriptor::decode(input)?;
+                let (input, value) = Value::decode(input)?;
+                Ok((
+                    input,
+                    DescribedValue {
+                        descriptor,
+                        value: Box::new(value),
+                    },
+                ))
+            }
+            _ => Err(Self::invalid_type_code(code)),
+        }
+    }
+}
+
+#[derive(Debug, Eq, PartialEq, Hash, Clone, From)]
+#[cfg_attr(test, derive(fake::Dummy))]
+pub enum CollectionValue {
+    List(List),
+    Map(Map<Value, Value>),
+}
+
+impl AmqpDecoder for CollectionValue {
+    fn decode(input: &[u8]) -> Result<(&[u8], Self), crate::message::amqp::AmqpDecodeError> {
+        let (_, code) = TypeCode::decode(input)?;
+
+        match code {
+            TypeCode::List0 | TypeCode::List8 | TypeCode::List32 => {
+                List::decode(input).map_second(CollectionValue::List)
+            }
+            TypeCode::Map8 | TypeCode::Map32 => Map::decode(input).map_second(CollectionValue::Map),
+            _ => Err(Self::invalid_type_code(code)),
+        }
+    }
+}
+
+impl AmqpEncoder for CollectionValue {
+    fn encoded_size(&self) -> u32 {
+        match self {
+            CollectionValue::List(list) => list.encoded_size(),
+            CollectionValue::Map(map) => map.encoded_size(),
+        }
+    }
+
+    fn encode(
+        &self,
+        writer: &mut impl std::io::Write,
+    ) -> Result<(), crate::message::amqp::AmqpEncodeError> {
+        match self {
+            CollectionValue::List(list) => list.encode(writer),
+            CollectionValue::Map(map) => map.encode(writer),
+        }
+    }
+}
+
+impl Hash for Map<Value, Value> {
+    fn hash<H: std::hash::Hasher>(&self, _state: &mut H) {
+        unimplemented!()
+    }
+}
+
+impl AmqpDecoder for Value {
+    fn decode(input: &[u8]) -> Result<(&[u8], Self), crate::message::amqp::AmqpDecodeError> {
+        let (_, code) = TypeCode::decode(input)?;
+
+        match code {
+            TypeCode::Described => DescribedValue::decode(input).map_second(Value::Described),
+            TypeCode::List0
+            | TypeCode::List8
+            | TypeCode::List32
+            | TypeCode::Map8
+            | TypeCode::Map32
+            | TypeCode::Array8
+            | TypeCode::Array32 => CollectionValue::decode(input).map_second(Value::Collection),
+
+            _ => SimpleValue::decode(input).map_second(Value::Simple),
+        }
+    }
+}
+
+impl AmqpEncoder for Value {
+    fn encoded_size(&self) -> u32 {
+        match self {
+            Value::Simple(simple) => simple.encoded_size(),
+            Value::Collection(collection) => collection.encoded_size(),
+            Value::Described(described) => described.encoded_size(),
+        }
+    }
+
+    fn encode(
+        &self,
+        writer: &mut impl std::io::Write,
+    ) -> Result<(), crate::message::amqp::AmqpEncodeError> {
+        match self {
+            Value::Simple(simple) => simple.encode(writer),
+            Value::Collection(collection) => collection.encode(writer),
+            Value::Described(described) => described.encode(writer),
+        }
+    }
+}
+
+macro_rules! impl_simple_value {
+    ($type:ty) => {
+        impl From<$type> for Value {
+            fn from(value: $type) -> Value {
+                Value::Simple(value.into())
+            }
+        }
+    };
+}
+macro_rules! impl_collection_value {
+    ($type:ty) => {
+        impl From<$type> for Value {
+            fn from(value: $type) -> Value {
+                Value::Collection(value.into())
+            }
+        }
+    };
+}
+
+macro_rules! impl_try_from_simple_value_ref {
+    ($type:ty) => {
+        impl<'a> TryFrom<&'a Value> for $type {
+            type Error = DecodeError;
+
+            fn try_from(value: &'a Value) -> Result<Self, Self::Error> {
+                use std::convert::TryInto;
+                match value {
+                    Value::Simple(simple) => simple
+                        .try_into()
+                        .map_err(|err: &str| DecodeError::MessageParse(err.to_string())),
+                    _ => Err(DecodeError::MessageParse(
+                        "Failed to cast Value to simple type".to_string(),
+                    )),
+                }
+            }
+        }
+    };
+}
+
+impl_try_from_simple_value_ref!(&'a Boolean);
+impl_try_from_simple_value_ref!(&'a UByte);
+impl_try_from_simple_value_ref!(&'a UShort);
+impl_try_from_simple_value_ref!(&'a UInt);
+impl_try_from_simple_value_ref!(&'a ULong);
+impl_try_from_simple_value_ref!(&'a Byte);
+impl_try_from_simple_value_ref!(&'a Short);
+impl_try_from_simple_value_ref!(&'a Int);
+impl_try_from_simple_value_ref!(&'a Long);
+impl_try_from_simple_value_ref!(&'a Float);
+impl_try_from_simple_value_ref!(&'a Double);
+impl_try_from_simple_value_ref!(&'a Char);
+impl_try_from_simple_value_ref!(&'a Timestamp);
+impl_try_from_simple_value_ref!(&'a Uuid);
+impl_try_from_simple_value_ref!(&'a Binary);
+impl_try_from_simple_value_ref!(&'a Str);
+impl_try_from_simple_value_ref!(&'a Symbol);
+
+impl_simple_value!(Boolean);
+impl_simple_value!(UByte);
+impl_simple_value!(UShort);
+impl_simple_value!(UInt);
+impl_simple_value!(ULong);
+impl_simple_value!(Byte);
+impl_simple_value!(Short);
+impl_simple_value!(Int);
+impl_simple_value!(Long);
+impl_simple_value!(Float);
+impl_simple_value!(Double);
+impl_simple_value!(Char);
+impl_simple_value!(Timestamp);
+impl_simple_value!(Uuid);
+impl_simple_value!(Binary);
+impl_simple_value!(Str);
+impl_simple_value!(Symbol);
+impl_simple_value!(&str);
+
+impl_collection_value!(List);
diff --git a/protocol/src/message/amqp/types/symbol.rs b/protocol/src/message/amqp/types/symbol.rs
new file mode 100644
index 0000000..4b55697
--- /dev/null
+++ b/protocol/src/message/amqp/types/symbol.rs
@@ -0,0 +1,91 @@
+use std::ops::Deref;
+
+use crate::{
+    codec::decoder::{read_exact, read_u32, read_u8},
+    message::amqp::{
+        codec::{constants::TypeCode, AmqpDecoder, AmqpEncoder},
+        error::{AmqpDecodeError, AmqpEncodeError},
+    },
+    utils::TupleMapperSecond,
+};
+
+use byteorder::{BigEndian, WriteBytesExt};
+#[cfg(test)]
+use fake::Fake;
+
+use super::Str;
+
+#[derive(Debug, Clone, Eq, PartialEq, Hash)]
+#[cfg_attr(test, derive(fake::Dummy))]
+pub struct Symbol(Str);
+
+impl Deref for Symbol {
+    type Target = Str;
+
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
+
+impl AmqpEncoder for Symbol {
+    fn encoded_size(&self) -> u32 {
+        let length = self.0.len() as u32;
+        let size = if length > u8::MAX as u32 { 5 } else { 2 };
+        size + length
+    }
+
+    fn encode(&self, writer: &mut impl std::io::Write) -> Result<(), AmqpEncodeError> {
+        let length = self.0.len();
+        if length > u8::MAX as usize {
+            TypeCode::Symbol32.encode(writer)?;
+            writer.write_u32::<BigEndian>(length as u32)?;
+        } else {
+            TypeCode::Symbol8.encode(writer)?;
+            writer.write_u8(length as u8)?;
+        }
+
+        writer.write_all(self.0.as_bytes())?;
+        Ok(())
+    }
+}
+impl AmqpDecoder for Symbol {
+    fn decode(input: &[u8]) -> Result<(&[u8], Self), AmqpDecodeError> {
+        let (input, code) = TypeCode::decode(input)?;
+
+        match code {
+            TypeCode::Symbol8 => {
+                let (input, len) = read_u8(input)?;
+                read_exact(input, len as usize)
+                    .map_second(|bytes| bytes.to_vec())
+                    .map(Ok)?
+                    .and_then(|(input, bytes)| Ok((input, String::from_utf8(bytes)?.into())))
+            }
+            TypeCode::Symbol32 => {
+                let (input, len) = read_u32(input)?;
+                read_exact(input, len as usize)
+                    .map_second(|bytes| bytes.to_vec())
+                    .map(Ok)?
+                    .and_then(|(input, bytes)| Ok((input, String::from_utf8(bytes)?.into())))
+            }
+            _ => Err(Self::invalid_type_code(code)),
+        }
+    }
+}
+
+impl From<Str> for Symbol {
+    fn from(string: Str) -> Self {
+        Symbol(string)
+    }
+}
+
+impl From<&str> for Symbol {
+    fn from(string: &str) -> Self {
+        Symbol(string.to_string())
+    }
+}
+
+impl PartialEq<Symbol> for str {
+    fn eq(&self, other: &Symbol) -> bool {
+        self == other.0
+    }
+}
diff --git a/protocol/src/message/builder.rs b/protocol/src/message/builder.rs
new file mode 100644
index 0000000..7afe103
--- /dev/null
+++ b/protocol/src/message/builder.rs
@@ -0,0 +1,176 @@
+use std::sync::Arc;
+
+use super::{InternalMessage, Message};
+
+use super::amqp::{
+    Address, AnnonationKey, Annotations, MessageId, SequenceNo, SimpleValue, Str, Symbol,
+    Timestamp, Value,
+};
+pub struct MessageBuilder(pub(crate) InternalMessage);
+
+impl MessageBuilder {
+    pub fn body(mut self, data: impl Into<Vec<u8>>) -> Self {
+        self.0.message.set_body(|body| {
+            body.set_data(data);
+        });
+        self
+    }
+
+    pub fn properties(self) -> PropertiesBuilder {
+        PropertiesBuilder(self)
+    }
+
+    pub fn message_annotations(self) -> AnnotationBuider {
+        AnnotationBuider(
+            self,
+            Box::new(|builder| builder.0.message.message_annotations_mut()),
+        )
+    }
+
+    pub fn application_properties(self) -> ApplicationPropertiesBuider {
+        ApplicationPropertiesBuider(self)
+    }
+    pub fn publising_id(mut self, publishing_id: u64) -> Self {
+        self.0.publishing_id = Some(publishing_id);
+        self
+    }
+    pub fn build(self) -> Message {
+        Message(Arc::new(self.0))
+    }
+}
+
+pub struct PropertiesBuilder(MessageBuilder);
+
+impl PropertiesBuilder {
+    pub fn message_builder(self) -> MessageBuilder {
+        self.0
+    }
+    pub fn message_id(mut self, message_id: impl Into<MessageId>) -> Self {
+        self.0
+             .0
+            .message
+            .with_properties(|p| p.message_id = Some(message_id.into()));
+        self
+    }
+    pub fn user_id(mut self, user_id: impl Into<Vec<u8>>) -> Self {
+        self.0
+             .0
+            .message
+            .with_properties(|p| p.user_id = Some(user_id.into()));
+        self
+    }
+    pub fn to(mut self, address: impl Into<Address>) -> Self {
+        self.0
+             .0
+            .message
+            .with_properties(|p| p.to = Some(address.into()));
+        self
+    }
+    pub fn subject(mut self, subject: impl Into<String>) -> Self {
+        self.0
+             .0
+            .message
+            .with_properties(|p| p.subject = Some(subject.into()));
+        self
+    }
+    pub fn reply_to(mut self, address: impl Into<Address>) -> Self {
+        self.0
+             .0
+            .message
+            .with_properties(|p| p.reply_to = Some(address.into()));
+        self
+    }
+    pub fn correlation_id(mut self, correlation_id: impl Into<MessageId>) -> Self {
+        self.0
+             .0
+            .message
+            .with_properties(|p| p.correlation_id = Some(correlation_id.into()));
+        self
+    }
+    pub fn content_type(mut self, content_type: impl Into<Symbol>) -> Self {
+        self.0
+             .0
+            .message
+            .with_properties(|p| p.content_type = Some(content_type.into()));
+        self
+    }
+    pub fn content_encoding(mut self, content_encoding: impl Into<Symbol>) -> Self {
+        self.0
+             .0
+            .message
+            .with_properties(|p| p.content_encoding = Some(content_encoding.into()));
+        self
+    }
+    pub fn absolute_expiry_time(mut self, expiry_time: impl Into<Timestamp>) -> Self {
+        self.0
+             .0
+            .message
+            .with_properties(|p| p.absolute_expiry_time = Some(expiry_time.into()));
+        self
+    }
+    pub fn creation_time(mut self, creation_time: impl Into<Timestamp>) -> Self {
+        self.0
+             .0
+            .message
+            .with_properties(|p| p.creation_time = Some(creation_time.into()));
+        self
+    }
+    pub fn group_id(mut self, group_id: impl Into<String>) -> Self {
+        self.0
+             .0
+            .message
+            .with_properties(|p| p.group_id = Some(group_id.into()));
+        self
+    }
+    pub fn group_sequence(mut self, group_sequence: SequenceNo) -> Self {
+        self.0
+             .0
+            .message
+            .with_properties(|p| p.group_sequence = Some(group_sequence));
+        self
+    }
+    pub fn reply_to_group_id(mut self, reply_to_group_id: impl Into<String>) -> Self {
+        self.0
+             .0
+            .message
+            .with_properties(|p| p.reply_to_group_id = Some(reply_to_group_id.into()));
+        self
+    }
+}
+
+pub struct AnnotationBuider(
+    MessageBuilder,
+    Box<dyn Fn(&mut MessageBuilder) -> &mut Annotations>,
+);
+
+impl AnnotationBuider {
+    pub fn insert<K, V>(mut self, key: K, value: V) -> Self
+    where
+        K: Into<AnnonationKey>,
+        V: Into<Value>,
+    {
+        let annotations = self.1(&mut self.0);
+        annotations.put(key, value);
+        self
+    }
+    pub fn message_builder(self) -> MessageBuilder {
+        self.0
+    }
+}
+
+pub struct ApplicationPropertiesBuider(MessageBuilder);
+
+impl ApplicationPropertiesBuider {
+    pub fn insert<K, V>(mut self, key: K, value: V) -> Self
+    where
+        K: Into<Str>,
+        V: Into<SimpleValue>,
+    {
+        let app_properties = self.0 .0.message.application_properties_mut();
+        app_properties.0.insert(key.into(), value.into());
+        self
+    }
+    pub fn message_builder(self) -> MessageBuilder {
+        self.0
+    }
+}
diff --git a/protocol/src/message/mod.rs b/protocol/src/message/mod.rs
new file mode 100644
index 0000000..5e01179
--- /dev/null
+++ b/protocol/src/message/mod.rs
@@ -0,0 +1,145 @@
+use std::convert::TryFrom;
+use std::sync::Arc;
+
+mod amqp;
+mod builder;
+
+use crate::codec::{Decoder, Encoder};
+use crate::error::{DecodeError, EncodeError};
+use crate::message::amqp::{AmqpDecoder, AmqpEncoder};
+
+use amqp::AmqpMessage;
+
+pub use self::amqp::{
+    AmqpDecodeError, AmqpEncodeError, AnnonationKey, Annotations, ApplicationProperties,
+    DeliveryAnnotations, Footer, Header, Map, MessageAnnotations, Properties, SimpleValue, Value,
+};
+
+use self::builder::MessageBuilder;
+
+/// API for inbound and outbound messages
+#[derive(Debug, PartialEq, Clone)]
+pub struct Message(Arc<InternalMessage>);
+
+#[derive(Debug, PartialEq, Default)]
+pub struct InternalMessage {
+    pub(crate) publishing_id: Option<u64>,
+    pub(crate) message: AmqpMessage,
+}
+
+impl Encoder for Message {
+    fn encoded_size(&self) -> u32 {
+        self.0.message.encoded_size()
+    }
+
+    fn encode(&self, writer: &mut impl std::io::Write) -> Result<(), crate::error::EncodeError> {
+        self.0.message.encode(writer)?;
+        Ok(())
+    }
+}
+
+impl Message {
+    #[cfg(test)]
+    pub(crate) fn new(internal: InternalMessage) -> Message {
+        Message(Arc::new(internal))
+    }
+
+    /// Builder for creating [`Message`]
+    pub fn builder() -> MessageBuilder {
+        MessageBuilder(InternalMessage {
+            message: AmqpMessage::default(),
+            publishing_id: None,
+        })
+    }
+
+    /// Extract a value as reference from the `amqp-value` section of the body if present
+    pub fn value_ref<'a, T>(&'a self) -> Result<Option<T>, DecodeError>
+    where
+        T: TryFrom<&'a Value, Error = DecodeError>,
+    {
+        self.0
+            .message
+            .body()
+            .value()
+            .map(|value| T::try_from(value))
+            .transpose()
+    }
+
+    /// Get the data associated to the message if any
+    pub fn data(&self) -> Option<&[u8]> {
+        self.0.message.body().data().map(|data| data.as_slice())
+    }
+
+    /// Get the properties of the message
+    pub fn properties(&self) -> Option<&Properties> {
+        self.0.message.properties()
+    }
+    /// Get the header of the message
+    pub fn header(&self) -> Option<&Header> {
+        self.0.message.header()
+    }
+
+    /// Get the annotations of the message
+    pub fn message_annotations(&self) -> Option<&MessageAnnotations> {
+        self.0.message.message_annotations()
+    }
+    /// Get the application properties of the message
+    pub fn application_properties(&self) -> Option<&ApplicationProperties> {
+        self.0.message.application_properties()
+    }
+
+    /// Get the delivery annotations of the message
+    pub fn delivery_annotations(&self) -> Option<&DeliveryAnnotations> {
+        self.0.message.delivery_annotations()
+    }
+
+    /// Get a reference to the message's publishing id.
+    pub fn publishing_id(&self) -> Option<&u64> {
+        self.0.publishing_id.as_ref()
+    }
+}
+
+impl Decoder for Message {
+    fn decode(input: &[u8]) -> Result<(&[u8], Self), crate::error::DecodeError> {
+        AmqpMessage::decode(input)
+            .map(|(remaining, message)| {
+                (
+                    remaining,
+                    Message(Arc::new(InternalMessage {
+                        publishing_id: None,
+                        message,
+                    })),
+                )
+            })
+            .map(Ok)?
+    }
+}
+
+impl From<Message> for Vec<Message> {
+    fn from(message: Message) -> Self {
+        vec![message]
+    }
+}
+
+impl From<AmqpEncodeError> for EncodeError {
+    fn from(err: AmqpEncodeError) -> Self {
+        match err {
+            AmqpEncodeError::Io(err) => EncodeError::Io(err),
+        }
+    }
+}
+
+impl From<AmqpDecodeError> for DecodeError {
+    fn from(err: AmqpDecodeError) -> Self {
+        match err {
+            AmqpDecodeError::InvalidTypeCode(code) => DecodeError::InvalidFormatCode(code),
+            AmqpDecodeError::MessageParse(err) => DecodeError::MessageParse(err),
+            AmqpDecodeError::Incomplete(err) => DecodeError::Incomplete(err),
+            AmqpDecodeError::Utf8Error(err) => DecodeError::Utf8Error(err),
+            AmqpDecodeError::UuidError(err) => DecodeError::MessageParse(err.to_string()),
+            AmqpDecodeError::InvalidTypeCodeFor { target, code } => {
+                DecodeError::MessageParse(format!("Invalid type code {:?} for {}", code, target))
+            }
+        }
+    }
+}
diff --git a/protocol/src/message/types.rs b/protocol/src/message/types.rs
new file mode 100644
index 0000000..e69de29
diff --git a/protocol/src/utils.rs b/protocol/src/utils.rs
new file mode 100644
index 0000000..b030231
--- /dev/null
+++ b/protocol/src/utils.rs
@@ -0,0 +1,25 @@
+pub trait TupleMapperFirst<Input, Output, Result>
+where
+    Self: Sized,
+{
+    fn map_first(self, mapper: impl Fn(Input) -> Output) -> Result;
+}
+
+pub trait TupleMapperSecond<Input, Output, Result>
+where
+    Self: Sized,
+{
+    fn map_second(self, mapper: impl FnOnce(Input) -> Output) -> Result;
+}
+
+impl<S, E, I, O> TupleMapperFirst<I, O, Result<(O, S), E>> for Result<(I, S), E> {
+    fn map_first(self, mapper: impl Fn(I) -> O) -> Result<(O, S), E> {
+        self.map(|(first, second)| (mapper(first), second))
+    }
+}
+
+impl<S, E, I, O> TupleMapperSecond<I, O, Result<(S, O), E>> for Result<(S, I), E> {
+    fn map_second(self, mapper: impl FnOnce(I) -> O) -> Result<(S, O), E> {
+        self.map(|(first, second)| (first, mapper(second)))
+    }
+}
diff --git a/src/consumer.rs b/src/consumer.rs
index 8df3dc2..4288c39 100644
--- a/src/consumer.rs
+++ b/src/consumer.rs
@@ -222,6 +222,7 @@ impl MessageHandler for ConsumerMessageHandler {
         Ok(())
     }
 }
+/// Envelope from incoming message
 #[derive(Debug)]
 pub struct Delivery {
     subscription_id: u8,
diff --git a/src/lib.rs b/src/lib.rs
index 4d914fd..9585cd3 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -96,4 +96,9 @@ pub mod types {
     pub use crate::stream_creator::StreamCreator;
     pub use rabbitmq_stream_protocol::message::Message;
     pub use rabbitmq_stream_protocol::{Response, ResponseCode, ResponseKind};
+
+    pub use rabbitmq_stream_protocol::message::{
+        AnnonationKey, ApplicationProperties, DeliveryAnnotations, Footer, Header, Map, Properties,
+        SimpleValue, Value,
+    };
 }
diff --git a/tests/integration/producer_test.rs b/tests/integration/producer_test.rs
index c69a135..e148906 100644
--- a/tests/integration/producer_test.rs
+++ b/tests/integration/producer_test.rs
@@ -171,3 +171,51 @@ async fn producer_batch_send() {
 
     producer.close().await.unwrap();
 }
+
+#[tokio::test(flavor = "multi_thread")]
+async fn producer_send_with_complex_message_ok() {
+    let env = TestEnvironment::create().await;
+
+    let producer = env.env.producer().build(&env.stream).await.unwrap();
+
+    let mut consumer = env
+        .env
+        .consumer()
+        .offset(OffsetSpecification::Next)
+        .build(&env.stream)
+        .await
+        .unwrap();
+
+    let _ = producer
+        .send_with_confirm(
+            Message::builder()
+                .body(b"message".to_vec())
+                .properties()
+                .message_id(32u64)
+                .message_builder()
+                .message_annotations()
+                .insert("test", "test")
+                .insert("test", true)
+                .insert("test", 3u8)
+                .message_builder()
+                .build(),
+        )
+        .await
+        .unwrap();
+
+    producer.close().await.unwrap();
+
+    let delivery = consumer.next().await.unwrap().unwrap();
+    assert_eq!(1, delivery.subscription_id());
+
+    let message = delivery.message();
+    assert_eq!(Some(b"message".as_ref()), message.data());
+    let properties = message.properties();
+
+    assert_eq!(
+        Some(32u64.into()),
+        properties.and_then(|properties| properties.message_id.clone())
+    );
+
+    consumer.handle().close().await.unwrap();
+}