diff --git a/.gitignore b/.gitignore
index 2923c5de..403d0d67 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1,5 +1,6 @@
 **/target
 **/Cargo.lock
+.idea/
 .vscode/
 .DS_Store
 
diff --git a/Cargo.toml b/Cargo.toml
index d0afd0fc..23e734ac 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -11,6 +11,7 @@ members = [
     "fe2o3-amqp-ws",
     "fe2o3-amqp-management",
     "fe2o3-amqp-cbs",
+    "fe2o3-amqp-http",
 ]
 
 exclude = [
@@ -67,9 +68,49 @@ log = "0.4"
 tracing = "0.1"
 tokio = { version = "1", default-features = false }
 tokio-util = "0.7"
+tokio-test = "0.4"
 futures-util = "0.3"
 uuid = "1"
 ordered-float = "4"
 pin-project-lite = "0.2"
 rand = "0.8"
-getrandom = { version = "0.2", features = ["js"] }
\ No newline at end of file
+getrandom = { version = "0.2", features = ["js"] }
+convert_case = "0.6.0"
+darling = "0.20"
+proc-macro2 = "1"
+quote = "1"
+syn = "2"
+url = "2"
+slab = "0.4"
+parking_lot = { version = "0.12", features = ["send_guard"] }
+sha-1 = "0.10"
+sha2 = "0.10"
+base64 = "0.22"
+stringprep = "0.1"
+hmac = "0.12"
+pbkdf2 = "0.12"
+webpki-roots = "0.26"
+tokio-rustls = "0.26"
+librustls = { package = "rustls", version = "0.23" }
+libnative-tls = { package = "native-tls", version = "0.2" }
+tokio-native-tls = "0.3"
+ring = "0.17"
+tokio-stream = "0.1"
+fluvio-wasm-timer = "0.2"
+testcontainers = "0.15.0"
+indexmap = "2"
+serde_json = "1"
+chrono = "0.4"
+time = "0.3"
+criterion = "0.5"
+http = "1"
+httpdate = "1"
+http-body = "1"
+http-body-util = "0.1"
+tungstenite = "0.24"
+tokio-tungstenite = "0.24"
+js-sys = "0.3"
+wasm-bindgen = "0.2"
+web-sys = "0.3"
+serde_repr = "0.1"
+trait-variant = "0.1.1"
\ No newline at end of file
diff --git a/fe2o3-amqp-cbs/Cargo.toml b/fe2o3-amqp-cbs/Cargo.toml
index c2b1b775..dd565097 100644
--- a/fe2o3-amqp-cbs/Cargo.toml
+++ b/fe2o3-amqp-cbs/Cargo.toml
@@ -13,6 +13,6 @@ readme = "Readme.md"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-fe2o3-amqp = { workspace = true }
-fe2o3-amqp-management = { workspace = true }
-trait-variant = "0.1.1"
\ No newline at end of file
+fe2o3-amqp.workspace = true
+fe2o3-amqp-management.workspace = true
+trait-variant.workspace = true
\ No newline at end of file
diff --git a/fe2o3-amqp-ext/Cargo.toml b/fe2o3-amqp-ext/Cargo.toml
index 15646611..f0fbb808 100644
--- a/fe2o3-amqp-ext/Cargo.toml
+++ b/fe2o3-amqp-ext/Cargo.toml
@@ -13,4 +13,4 @@ readme = "Readme.md"
 
 [dependencies]
 serde_amqp = { workspace = true, features = ["derive"] }
-fe2o3-amqp-types = { workspace = true }
+fe2o3-amqp-types.workspace = true
diff --git a/fe2o3-amqp-http/Cargo.toml b/fe2o3-amqp-http/Cargo.toml
new file mode 100644
index 00000000..99a2a314
--- /dev/null
+++ b/fe2o3-amqp-http/Cargo.toml
@@ -0,0 +1,19 @@
+[package]
+name = "fe2o3-amqp-http"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+fe2o3-amqp.workspace = true
+fe2o3-amqp-types.workspace = true
+serde.workspace = true
+thiserror.workspace = true
+
+http.workspace = true
+httpdate.workspace = true
+http-body.workspace = true
+http-body-util.workspace = true
+
+# Optional dependencies
+log = { workspace = true, optional = true }
+tracing = { workspace = true, optional = true }
\ No newline at end of file
diff --git a/fe2o3-amqp-http/src/client.rs b/fe2o3-amqp-http/src/client.rs
new file mode 100644
index 00000000..7bb62cec
--- /dev/null
+++ b/fe2o3-amqp-http/src/client.rs
@@ -0,0 +1,9 @@
+use fe2o3_amqp::{Receiver, Sender};
+
+use crate::projected::TryIntoProjected;
+
+#[derive(Debug)]
+pub struct Client {
+    sender: Sender,
+    receiver: Receiver,
+}
diff --git a/fe2o3-amqp-http/src/error.rs b/fe2o3-amqp-http/src/error.rs
new file mode 100644
index 00000000..9f8e5e99
--- /dev/null
+++ b/fe2o3-amqp-http/src/error.rs
@@ -0,0 +1,58 @@
+use http::{header::{InvalidHeaderValue, ToStrError}, method::InvalidMethod, status::InvalidStatusCode};
+use thiserror::Error;
+
+#[derive(Debug, Error)]
+pub enum TryIntoProjectedError<BE> {
+    #[error(transparent)]
+    ToStrError(#[from] ToStrError),
+
+    #[error(transparent)]
+    Date(#[from] httpdate::Error),
+
+    #[error(transparent)]
+    SystemTime(#[from] std::time::SystemTimeError),
+
+    #[error(transparent)]
+    Body(BE),
+}
+
+#[derive(Debug, Error)]
+pub enum TryFromProjectedError<BE> {
+    #[error("HTTP method not present")]
+    MethodNotPresent,
+
+    #[error(transparent)]
+    InvalidMethod(#[from] InvalidMethod),
+
+    #[error("HTTP status not present")]
+    StatusNotPresent,
+
+    #[error(transparent)]
+    InvalidStatusCode(#[from] InvalidStatusCode),
+
+    #[error("HTTP request target not present")]
+    RequestTargetNotPresent,
+
+    #[error("HTTP version not encoded as a string")]
+    VersionNotString,
+
+    #[error("HTTP version not valid")]
+    InvalidVersion,
+
+    #[error(transparent)]
+    HeaderValue(#[from] InvalidHeaderValue),
+
+    #[error(transparent)]
+    Body(BE),
+
+    #[error(transparent)]
+    Http(#[from] http::Error),
+}
+
+pub(crate) struct InvalidVersion;
+
+impl<T> From<InvalidVersion> for TryFromProjectedError<T> {
+    fn from(_: InvalidVersion) -> Self {
+        TryFromProjectedError::InvalidVersion
+    }
+}
\ No newline at end of file
diff --git a/fe2o3-amqp-http/src/lib.rs b/fe2o3-amqp-http/src/lib.rs
new file mode 100644
index 00000000..f76146fd
--- /dev/null
+++ b/fe2o3-amqp-http/src/lib.rs
@@ -0,0 +1,7 @@
+pub mod client;
+pub mod projected;
+pub mod tunneled;
+pub mod uri;
+pub mod error;
+
+mod util;
\ No newline at end of file
diff --git a/fe2o3-amqp-http/src/projected.rs b/fe2o3-amqp-http/src/projected.rs
new file mode 100644
index 00000000..b585516f
--- /dev/null
+++ b/fe2o3-amqp-http/src/projected.rs
@@ -0,0 +1,412 @@
+use std::str::FromStr;
+
+use fe2o3_amqp_types::{
+    messaging::{
+        map_builder::MapBuilder, properties, ApplicationProperties, Batch, Data, Message, Properties
+    },
+    primitives::{Binary, SimpleValue},
+};
+use http::{
+    header::{CONTENT_ENCODING, CONTENT_TYPE, DATE, EXPIRES, FROM}, HeaderName, HeaderValue, Method, Request, Response
+};
+
+use crate::{error::{TryFromProjectedError, TryIntoProjectedError}, util::{parse_http_version, timestamp_to_httpdate}};
+
+type PropertiesBuilder = properties::Builder;
+type ApplicationPropertiesBuilder = MapBuilder<String, SimpleValue, ApplicationProperties>;
+type ProjectedMessageBody = Batch<Data>;
+type ProjectedMessage = Message<ProjectedMessageBody>;
+
+
+/// Type Parameter B is the body type of the HTTP message.
+///
+/// > When interpreting the message content, it MUST be considered as
+/// equivalent to a single data section obtained by concatenating all the
+/// data sections, the data section boundaries MUST be ignored.
+pub trait TryIntoProjected<B>
+where
+    B: TryInto<ProjectedMessageBody>,
+    B::Error: std::error::Error,
+{
+    type Error: std::error::Error;
+
+    fn into_projected(self) -> Result<ProjectedMessage, Self::Error>;
+}
+
+fn project_request_line(
+    prop_builder: PropertiesBuilder,
+    app_prop_builder: ApplicationPropertiesBuilder,
+    method: &str,
+    uri: &str,
+    version: &http::Version,
+) -> (PropertiesBuilder, ApplicationPropertiesBuilder) {
+    // 4.1.1 Request Line
+    let prop_builder = prop_builder.subject(method).to(uri);
+    let app_prop_builder = app_prop_builder.insert("http:request", format!("{:?}", version)); // TODO: something better than Debug fmt?
+
+    (prop_builder, app_prop_builder)
+}
+
+fn project_status_line(
+    prop_builder: PropertiesBuilder,
+    app_prop_builder: ApplicationPropertiesBuilder,
+    status: &http::StatusCode,
+    version: &http::Version,
+) -> (PropertiesBuilder, ApplicationPropertiesBuilder) {
+    // 4.1.2 Status Line
+    let app_prop_builder = app_prop_builder.insert("http:response", format!("{:?}", version)); // TODO: something better than Debug fmt?
+    let prop_builder = prop_builder.subject(status.as_str());
+
+    (prop_builder, app_prop_builder)
+}
+
+fn project_headers<BE>(
+    mut prop_builder: PropertiesBuilder,
+    mut app_prop_builder: ApplicationPropertiesBuilder,
+    headers: &http::HeaderMap,
+) -> Result<(PropertiesBuilder, ApplicationPropertiesBuilder), TryIntoProjectedError<BE>> {
+    // 4.1.3 Headers
+    // The following HTTP Headers defined in RFC7230 MUST NOT be mapped into AMQP HTTP messages
+    // - TE
+    // - Trailer
+    // - Transfer-Encoding
+    // - Content-Length
+    // - Via
+    // - Connection
+    // - Upgrade
+    const EXCLUDED_HEADERS_LOWERCASE: [&str; 7] = [
+        "te",
+        "trailer",
+        "transfer-encoding",
+        "content-length",
+        "via",
+        "connection",
+        "upgrade",
+    ];
+
+    // The HTTP Host information MUST follow the addressing rules defined in section 3. While the Host header
+    // is required in RFC7230, it is OPTIONAL in HTTP AMQP because the container is already uniquely
+    // identified through other means. The Host value is set in the application-properties section, as value of the
+    // “http:host” string property. When the property is omitted, the default value is a container-defined scope
+    // identifier.
+
+    // The following RFC7231 headers have special mappings:
+    // - Content-Type maps to the properties section, content-type field.
+    if let Some(content_type) = headers.get(CONTENT_TYPE) {
+        prop_builder = prop_builder.content_type(content_type.to_str()?);
+    }
+    // - Content-Encoding maps to the properties section, content-encoding field.
+    if let Some(content_encoding) = headers.get(CONTENT_ENCODING) {
+        prop_builder = prop_builder.content_encoding(content_encoding.to_str()?);
+    }
+    // - Date maps to the properties section, creation-time field.
+    if let Some(date) = headers.get(DATE) {
+        let date_str = date.to_str()?;
+        let timestamp = crate::util::httpdate_to_timestamp(date_str)?;
+        prop_builder = prop_builder.creation_time(timestamp);
+    }
+    // - From maps to the properties section, user-id field.
+    if let Some(from) = headers.get(FROM) {
+        prop_builder = prop_builder.user_id(Binary::from(from.as_bytes()));
+    }
+
+    // The following RFC7234 header has a special mapping
+    // - The Expires value maps to the properties section, absolute-expiry-time field.
+    if let Some(expires) = headers.get(EXPIRES) {
+        let expires_str = expires.to_str()?;
+        let timestamp = crate::util::httpdate_to_timestamp(expires_str)?;
+        prop_builder = prop_builder.absolute_expiry_time(timestamp);
+    }
+
+    const SPECIAL_HEADERS_LOWERCASE: [&str; 5] = [
+        "content-type",
+        "content-encoding",
+        "date",
+        "from",
+        "expires",
+    ];
+
+    // All other HTTP headers, including those defined in RFC7231 and any
+    // headers defined by formal HTTP extensions as well as any application
+    // specific HTTP headers are added to the application-properties section
+    // of the message. The header names are not prefixed. Headers with
+    // special mappings MUST NOT be added to the application-properties
+    // section
+    //
+    // Because HTTP header names are case-insensitive but AMQP property names are case-sensitive, all
+    // HTTP header names MUST be converted to lower case as they are mapped to AMQP application
+    // property names. The type of all mapped header values is string.
+    let iter = headers
+        .iter()
+        .filter(|(name, _)| {
+            !EXCLUDED_HEADERS_LOWERCASE.contains(&name.as_str().to_lowercase().as_str())
+        }) // TODO: how to avoid creating new strings?
+        .filter(|(name, _)| {
+            !SPECIAL_HEADERS_LOWERCASE.contains(&name.as_str().to_lowercase().as_str())
+        });
+    for (name, value) in iter {
+        app_prop_builder = app_prop_builder.insert(name.as_str().to_lowercase(), value.to_str()?);
+    }
+
+    Ok((prop_builder, app_prop_builder))
+}
+
+impl<B> TryIntoProjected<B> for Request<B>
+where
+    B: TryInto<ProjectedMessageBody>,
+    B::Error: std::error::Error,
+{
+    type Error = TryIntoProjectedError<B::Error>;
+
+    /// Implement HTTP mapping to AMQP message in projected mode.
+    ///
+    /// See Section 4.1 for more details.
+    fn into_projected(self) -> Result<ProjectedMessage, Self::Error> {
+        //  An implementation MUST ignore and exclude all RFC7230 headers and
+        //  RFC7230 information items not explicitly covered below
+
+        // MUST include all headers and information items from RFC7231 and other
+        // HTTP extension specifications.
+
+        let (parts, body) = self.into_parts();
+
+        let prop_builder = Properties::builder();
+        let app_prop_builder = ApplicationProperties::builder();
+
+        let (prop_builder, app_prop_builder) = project_request_line(
+            prop_builder,
+            app_prop_builder,
+            parts.method.as_str(),
+            parts.uri.to_string().as_str(),
+            &parts.version,
+        );
+        let (prop_builder, app_prop_builder) =
+            project_headers(prop_builder, app_prop_builder, &parts.headers)?;
+
+        let properties = prop_builder.build();
+        let application_properties = app_prop_builder.build();
+
+        let data = body.try_into().map_err(TryIntoProjectedError::Body)?;
+        let msg = Message::builder()
+            .properties(properties)
+            .application_properties(application_properties)
+            .data_batch(data)
+            .build();
+
+        Ok(msg)
+    }
+}
+
+impl<B> TryIntoProjected<B> for Response<B>
+where
+    B: TryInto<ProjectedMessageBody>,
+    B::Error: std::error::Error,
+{
+    type Error = TryIntoProjectedError<B::Error>;
+
+    fn into_projected(self) -> Result<ProjectedMessage, Self::Error> {
+        let (parts, body) = self.into_parts();
+
+        let prop_builder = Properties::builder();
+        let app_prop_builder = ApplicationProperties::builder();
+
+        let (prop_builder, app_prop_builder) = project_status_line(
+            prop_builder,
+            app_prop_builder,
+            &parts.status,
+            &parts.version,
+        );
+        let (prop_builder, app_prop_builder) =
+            project_headers(prop_builder, app_prop_builder, &parts.headers)?;
+
+        let properties = prop_builder.build();
+        let application_properties = app_prop_builder.build();
+
+        let data = body.try_into().map_err(TryIntoProjectedError::Body)?;
+        let msg = Message::builder()
+            .properties(properties)
+            .application_properties(application_properties)
+            .data_batch(data)
+            .build();
+
+        Ok(msg)
+    }
+}
+
+pub trait TryFromProjected
+where
+    Self: Sized,
+{
+    type Body;
+    type Error: std::error::Error;
+
+    fn try_from_projected(msg: ProjectedMessage) -> Result<Self, Self::Error>;
+}
+
+impl<B> TryFromProjected for Request<B>
+where
+    B: TryFrom<ProjectedMessageBody>,
+    B::Error: std::error::Error,
+{
+    type Body = B;
+
+    type Error = TryFromProjectedError<B::Error>;
+
+    fn try_from_projected(msg: ProjectedMessage) -> Result<Self, Self::Error> {
+        // If properties are missing, method is definitely missing
+        let properties = msg.properties.ok_or(TryFromProjectedError::MethodNotPresent)?;
+        // There shouldn't be any allocation for the default 
+        let application_properties = msg.application_properties.unwrap_or_default();
+        
+        // Get request line
+        let method = properties
+            .subject
+            .map(|s| Method::from_str(&s))
+            .ok_or(TryFromProjectedError::MethodNotPresent)??;
+        let uri = properties
+            .to
+            .ok_or(TryFromProjectedError::RequestTargetNotPresent)?;
+        let version = application_properties
+            .get("http:request")
+            .map(|v| match v {
+                SimpleValue::String(s) => Ok(s.as_str()),
+                _ => Err(TryFromProjectedError::VersionNotString), 
+            })
+            .unwrap_or_else(|| Ok("1.1"))?; // default to 1.1
+
+        
+        let version = parse_http_version(version)?;
+        let mut builder = Request::builder()
+            .method(method)
+            .uri(uri)
+            .version(version);
+
+        // Get special headers
+        if let Some(content_type) = properties.content_type {
+            builder = builder.header(CONTENT_TYPE, HeaderValue::from_str(&content_type.0)?);
+        }
+        if let Some(content_encoding) = properties.content_encoding {
+            builder = builder.header(CONTENT_ENCODING, HeaderValue::from_str(&content_encoding.0)?);
+        }
+        if let Some(creation_time) = properties.creation_time {
+            let date_str = timestamp_to_httpdate(creation_time);
+            builder = builder.header(DATE, HeaderValue::from_str(&date_str)?);
+        }
+        if let Some(user_id) = properties.user_id {
+            builder = builder.header(FROM, HeaderValue::from_bytes(&user_id)?);
+        }
+        if let Some(absolute_expiry_time) = properties.absolute_expiry_time {
+            let expires_str = timestamp_to_httpdate(absolute_expiry_time);
+            builder = builder.header(EXPIRES, HeaderValue::from_str(&expires_str)?);
+        }
+
+        const HEADERS_TO_IGNORE_LOWERCASE: [&str; 1] = [
+            "http:request",
+        ];
+
+        // Get other headers
+        // Compare lowercased header names to SPECIAL_HEADERS_LOWERCASE
+        let iter = application_properties.iter().filter_map(|(name, value)| {
+            let name = name.to_lowercase();
+            if HEADERS_TO_IGNORE_LOWERCASE.contains(&name.as_str()) {
+                None
+            } else {
+                // TODO: return an error or ignore the wrong header?
+                let name = HeaderName::from_lowercase(name.as_bytes()).ok()?;
+                let value = match value {
+                    SimpleValue::String(s) => HeaderValue::from_str(s).ok()?,
+                    _ => return None,
+                };
+                Some((name, value))
+            }
+        });
+
+        for (name, value) in iter {
+            builder = builder.header(name, value);
+        }
+        
+        let body = B::try_from(msg.body).map_err(TryFromProjectedError::Body)?;
+        builder.body(body).map_err(Into::into)
+    }
+}
+
+impl<B> TryFromProjected for Response<B>
+where
+    B: TryFrom<ProjectedMessageBody>,
+    B::Error: std::error::Error,
+{
+    type Body = B;
+
+    type Error = TryFromProjectedError<B::Error>;
+
+    fn try_from_projected(msg: ProjectedMessage) -> Result<Self, Self::Error> {
+        let properties = msg.properties.ok_or(TryFromProjectedError::StatusNotPresent)?;
+        let application_properties = msg.application_properties.unwrap_or_default();
+        
+        // Get status line
+        let status = properties
+            .subject
+            .map(|s| http::StatusCode::from_bytes(s.as_bytes()))
+            .ok_or(TryFromProjectedError::StatusNotPresent)??;
+        let version = application_properties
+            .get("http:response")
+            .map(|v| match v {
+                SimpleValue::String(s) => Ok(s.as_str()),
+                _ => Err(TryFromProjectedError::VersionNotString), 
+            })
+            .unwrap_or_else(|| Ok("1.1"))?; // default to 1.1
+
+        let version = parse_http_version(version)?;
+        let mut builder = Response::builder()
+            .status(status)
+            .version(version);
+
+        // Get special headers
+        if let Some(content_type) = properties.content_type {
+            builder = builder.header(CONTENT_TYPE, HeaderValue::from_str(&content_type.0)?);
+        }
+        if let Some(content_encoding) = properties.content_encoding {
+            builder = builder.header(CONTENT_ENCODING, HeaderValue::from_str(&content_encoding.0)?);
+        }
+        if let Some(creation_time) = properties.creation_time {
+            let date_str = timestamp_to_httpdate(creation_time);
+            builder = builder.header(DATE, HeaderValue::from_str(&date_str)?);
+        }
+        if let Some(user_id) = properties.user_id {
+            builder = builder.header(FROM, HeaderValue::from_bytes(&user_id)?);
+        }
+        if let Some(absolute_expiry_time) = properties.absolute_expiry_time {
+            let expires_str = timestamp_to_httpdate(absolute_expiry_time);
+            builder = builder.header(EXPIRES, HeaderValue::from_str(&expires_str)?);
+        }
+
+        // TODO: other headers that should be ignored?
+        const HEADERS_TO_IGNORE_LOWERCASE: [&str; 1] = [
+            "http:response",
+        ];
+
+        // Get other headers
+        // Compare lowercased header names to SPECIAL_HEADERS_LOWERCASE
+        let iter = application_properties.iter().filter_map(|(name, value)| {
+            let name = name.to_lowercase();
+            if HEADERS_TO_IGNORE_LOWERCASE.contains(&name.as_str()) {
+                None
+            } else {
+                // TODO: return an error or ignore the wrong header?
+                let name = HeaderName::from_lowercase(name.as_bytes()).ok()?;
+                let value = match value {
+                    SimpleValue::String(s) => HeaderValue::from_str(s).ok()?,
+                    _ => return None,
+                };
+                Some((name, value))
+            }
+        });
+
+        for (name, value) in iter {
+            builder = builder.header(name, value);
+        }
+
+        let body = B::try_from(msg.body).map_err(TryFromProjectedError::Body)?;
+        builder.body(body).map_err(Into::into)
+    }
+}
diff --git a/fe2o3-amqp-http/src/tunneled.rs b/fe2o3-amqp-http/src/tunneled.rs
new file mode 100644
index 00000000..cf995049
--- /dev/null
+++ b/fe2o3-amqp-http/src/tunneled.rs
@@ -0,0 +1,5 @@
+use fe2o3_amqp_types::{messaging::Message, primitives::Value};
+
+pub trait IntoTunneled {
+    fn into_tunneled(self) -> Message<Value>;
+}
\ No newline at end of file
diff --git a/fe2o3-amqp-http/src/uri.rs b/fe2o3-amqp-http/src/uri.rs
new file mode 100644
index 00000000..e69de29b
diff --git a/fe2o3-amqp-http/src/util.rs b/fe2o3-amqp-http/src/util.rs
new file mode 100644
index 00000000..3868bd47
--- /dev/null
+++ b/fe2o3-amqp-http/src/util.rs
@@ -0,0 +1,35 @@
+use fe2o3_amqp_types::primitives::Timestamp;
+use http::Version;
+
+use crate::error::{InvalidVersion, TryFromProjectedError, TryIntoProjectedError};
+
+pub(crate) fn httpdate_to_timestamp<BE>(
+    httpdate: &str
+) -> Result<Timestamp, TryIntoProjectedError<BE>> {
+    let sys_time = httpdate::parse_http_date(httpdate)?;
+    let duration_since_unix_epoch = sys_time.duration_since(std::time::UNIX_EPOCH)?;
+
+    let millis = duration_since_unix_epoch.as_millis() as i64;
+    Ok(Timestamp::from_milliseconds(millis))
+}
+
+pub(crate) fn timestamp_to_httpdate(
+    timestamp: Timestamp
+) -> String {
+    let millis = timestamp.milliseconds();
+    let duration_since_unix_epoch = std::time::Duration::from_millis(millis as u64);
+    let sys_time = std::time::UNIX_EPOCH + duration_since_unix_epoch;
+
+    httpdate::fmt_http_date(sys_time)
+}
+
+pub(crate) fn parse_http_version(s: &str) -> Result<Version, InvalidVersion> {
+    match s {
+        "HTTP/0.9" => Ok(Version::HTTP_09),
+        "HTTP/1.0" => Ok(Version::HTTP_10),
+        "HTTP/1.1" => Ok(Version::HTTP_11),
+        "HTTP/2.0" => Ok(Version::HTTP_2),
+        "HTTP/3.0" => Ok(Version::HTTP_3),
+        _ => Err(InvalidVersion),
+    }
+}
\ No newline at end of file
diff --git a/fe2o3-amqp-management/Cargo.toml b/fe2o3-amqp-management/Cargo.toml
index 29d1d00b..014a03b9 100644
--- a/fe2o3-amqp-management/Cargo.toml
+++ b/fe2o3-amqp-management/Cargo.toml
@@ -13,10 +13,10 @@ readme = "Readme.md"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-fe2o3-amqp = { workspace = true }
+fe2o3-amqp.workspace = true
 fe2o3-amqp-types =  { workspace = true }
-serde = { workspace = true }
-thiserror = { workspace = true }
+serde.workspace = true
+thiserror.workspace = true
 
 log = { workspace = true, optional = true }
 tracing = { workspace = true, optional = true }
\ No newline at end of file
diff --git a/fe2o3-amqp-types/Cargo.toml b/fe2o3-amqp-types/Cargo.toml
index 80d8f0e7..6f9274ff 100644
--- a/fe2o3-amqp-types/Cargo.toml
+++ b/fe2o3-amqp-types/Cargo.toml
@@ -35,6 +35,6 @@ security = ["primitive"]
 [dependencies]
 serde_amqp = { workspace = true, features = ["derive", "extensions"] }
 serde = { workspace = true, features = ["derive"] }
-serde_bytes = { workspace = true }
+serde_bytes.workspace = true
 ordered-float = { workspace = true, features = ["serde"] }
-serde_repr = "0.1"
\ No newline at end of file
+serde_repr.workspace = true
\ No newline at end of file
diff --git a/fe2o3-amqp-ws/Cargo.toml b/fe2o3-amqp-ws/Cargo.toml
index d5f560bd..2877a67d 100644
--- a/fe2o3-amqp-ws/Cargo.toml
+++ b/fe2o3-amqp-ws/Cargo.toml
@@ -26,21 +26,21 @@ rustls-tls-webpki-roots = ["tokio-tungstenite/rustls-tls-webpki-roots"]
 
 [dependencies]
 futures-util = { workspace = true, features = ["sink"] }
-http = "1"
-pin-project-lite = { workspace = true }
-thiserror = { workspace = true }
-tungstenite = "0.24"
+http.workspace = true
+pin-project-lite.workspace = true
+thiserror.workspace = true
+tungstenite.workspace = true
 
 [target.'cfg(not(target_arch = "wasm32"))'.dependencies]
 tokio = { workspace = true, features = ["net"] }
-tokio-tungstenite = "0.24"
+tokio-tungstenite.workspace = true
 
 [target.'cfg(target_arch = "wasm32")'.dependencies]
 tokio = { workspace = true, features = ["sync", "macros"] }
-getrandom = { workspace = true }
-js-sys = "0.3"
-wasm-bindgen = "0.2"
-web-sys = { version = "0.3", features = [
+getrandom.workspace = true
+js-sys.workspace = true
+wasm-bindgen.workspace = true
+web-sys = { workspace = true, features = [
   "WebSocket",
   "MessageEvent",
   "CloseEvent",
@@ -49,7 +49,7 @@ web-sys = { version = "0.3", features = [
 ] }
 
 [dev-dependencies]
-fe2o3-amqp = { workspace = true }
+fe2o3-amqp.workspace = true
 
 [target.'cfg(not(target_arch = "wasm32"))'.dev-dependencies]
 tokio = { workspace = true, features = ["net", "rt-multi-thread"] }
diff --git a/fe2o3-amqp/Cargo.toml b/fe2o3-amqp/Cargo.toml
index f33a5afe..cc5f1294 100644
--- a/fe2o3-amqp/Cargo.toml
+++ b/fe2o3-amqp/Cargo.toml
@@ -40,19 +40,19 @@ acceptor = []
 scram = ["sha-1", "sha2", "rand", "base64", "stringprep", "hmac", "pbkdf2"]
 
 [dependencies]
-serde_amqp = { workspace = true }
-fe2o3-amqp-types = { workspace = true }
+serde_amqp.workspace = true
+fe2o3-amqp-types.workspace = true
 
-bytes = { workspace = true }
+bytes.workspace = true
 tokio-util = { workspace = true, features = ["codec"] } # tokio-rs/tokio#4816
-thiserror = { workspace = true }
-serde = { workspace = true }
+thiserror.workspace = true
+serde.workspace = true
 futures-util = { workspace = true, features = ["sink"] }
-pin-project-lite = "0.2"
-url = "2"
-slab = "0.4"
-serde_bytes = { workspace = true }
-parking_lot = { version = "0.12", features = ["send_guard"] }
+pin-project-lite.workspace = true
+url.workspace = true
+slab.workspace = true
+serde_bytes.workspace = true
+parking_lot = { workspace = true, features = ["send_guard"] }
 
 # Optinoal deps that are feature themselves
 tracing = { workspace = true, optional = true }
@@ -60,38 +60,38 @@ log = { workspace = true, optional = true }
 
 # Optional deps
 uuid = { workspace = true, features = ["v4"], optional = true }
-sha-1 = { version = "0.10", optional = true }
-sha2 = { version = "0.10", optional = true }
+sha-1 = { workspace = true, optional = true }
+sha2 = { workspace = true, optional = true }
 rand = { workspace = true, optional = true }
-base64 = { version = "0.22", optional = true } # TODO: replace with base64-simd?
-stringprep = { version = "0.1", optional = true }
-hmac = { version = "0.12", optional = true }
-pbkdf2 = { version = "0.12", default-features = false, optional = true }
-webpki-roots = { version = "0.26", optional = true }
-tokio-rustls = { version = "0.26", default-features = false, features = ["logging", "tls12", "ring"], optional = true }
-librustls = { package = "rustls", version = "0.23", default-features = false, features = ["logging", "std", "tls12", "ring"], optional = true }
+base64 = { workspace = true, optional = true } # TODO: replace with base64-simd?
+stringprep = { workspace = true, optional = true }
+hmac = { workspace = true, optional = true }
+pbkdf2 = { workspace = true, default-features = false, optional = true }
+webpki-roots = { workspace = true, optional = true }
+tokio-rustls = { workspace = true, default-features = false, features = ["logging", "tls12", "ring"], optional = true }
+librustls = { workspace = true, default-features = false, features = ["logging", "std", "tls12", "ring"], optional = true }
 
 [target.'cfg(not(target_arch = "wasm32"))'.dependencies]
 tokio = { workspace = true, features = ["sync", "io-util", "net", "rt", "macros", "time"] }
-libnative-tls = { package = "native-tls", version = "0.2", optional = true }
-tokio-native-tls = { version = "0.3", optional = true }
-ring = { version = "0.17", default-features = false, optional = true }
-tokio-stream = { version = "0.1", features = ["time"] }
+libnative-tls = { workspace = true, optional = true }
+tokio-native-tls = { workspace = true, optional = true }
+ring = { workspace = true, default-features = false, optional = true }
+tokio-stream = { workspace = true, features = ["time"] }
 
 [target.'cfg(target_arch = "wasm32")'.dependencies]
 tokio = { workspace = true, features = ["sync", "io-util", "rt", "macros"] } # "net" feature doesn't support wasm32
-ring = { version = "0.17", default-features = false, features = ["wasm32_unknown_unknown_js"], optional = true }
-getrandom = { workspace = true }
-fluvio-wasm-timer = "0.2"
+ring = { workspace = true, default-features = false, features = ["wasm32_unknown_unknown_js"], optional = true }
+getrandom.workspace = true
+fluvio-wasm-timer.workspace = true
 
 [dev-dependencies]
-tokio-test = { version = "0.4" }
-testcontainers = "0.15.0"
-fe2o3-amqp-ext = { workspace = true }
+tokio-test.workspace = true
+testcontainers.workspace = true
+fe2o3-amqp-ext.workspace = true
 
 [target.'cfg(not(target_arch = "wasm32"))'.dev-dependencies]
 tokio = { workspace = true, features = ["rt", "rt-multi-thread", "macros", "parking_lot"] }
 
 [target.'cfg(target_arch = "wasm32")'.dev-dependencies]
 tokio = { workspace = true, features = ["rt", "macros", "parking_lot"]}
-getrandom = { workspace = true }
+getrandom.workspace = true
diff --git a/serde_amqp/Cargo.toml b/serde_amqp/Cargo.toml
index ebb5b783..04511ca0 100644
--- a/serde_amqp/Cargo.toml
+++ b/serde_amqp/Cargo.toml
@@ -32,27 +32,27 @@ json = ["serde_json"]
 chrono = ["dep:chrono"]
 
 [dev-dependencies]
-criterion = "0.5"
-rand = { workspace = true }
+criterion.workspace = true
+rand.workspace = true
 uuid = { workspace = true, features = ["v4"] }
-bytes = { workspace = true }
+bytes.workspace = true
 
 [dependencies]
 ordered-float = { workspace = true, features = ["serde"] }
 serde = { workspace = true, features = ["derive"] }
-serde_bytes = { workspace = true }
-thiserror = { workspace = true }
-bytes = { workspace = true }
-indexmap = { version = "2", features = ["serde"] }
+serde_bytes.workspace = true
+thiserror.workspace = true
+bytes.workspace = true
+indexmap = { workspace = true, features = ["serde"] }
 
 # derive
 serde_amqp_derive = { workspace = true, optional = true }
 
-# Optinal dependencies
-serde_json = { version = "1", optional = true }
-chrono = { version = "0.4.30", optional = true }
+# Optional dependencies
+serde_json = { workspace = true, optional = true }
+chrono = { workspace = true, optional = true }
 uuid = { workspace = true, optional = true }
-time = { version = "0.3", optional = true }
+time = { workspace = true, optional = true }
 
 [[bench]]
 name = "serialize"
diff --git a/serde_amqp_derive/Cargo.toml b/serde_amqp_derive/Cargo.toml
index 7f4c975c..2c8a2a67 100644
--- a/serde_amqp_derive/Cargo.toml
+++ b/serde_amqp_derive/Cargo.toml
@@ -20,8 +20,8 @@ proc-macro = true
 serde = { workspace = true, features = ["derive"] }
 
 [dependencies]
-convert_case = "0.6.0"
-darling = "0.20"
-proc-macro2 = "1"
-quote = "1"
-syn = { version = "2", features = ["parsing", "derive"] } 
+convert_case.workspace = true
+darling.workspace = true
+proc-macro2.workspace = true
+quote.workspace = true
+syn = { workspace = true, features = ["parsing", "derive"] }