Refactor redundant header logic to a shared lib (#146)

* Refactor redundant header logic to a shared lib

* Remove some unused code

* Share macro between actix and warp, eliminating actix header mod

Fixes #145

Signed-off-by: Jim Crossley <[email protected]>

Author: jcrossley3

Cargo.toml

@@ -17,10 +17,10 @@ categories = ["web-programming", "encoding", "data-structures"]
 name = "cloudevents"
 
 [features]
-actix = ["actix-web", "async-trait", "lazy_static", "bytes", "futures"]
-reqwest = ["reqwest-lib", "async-trait", "lazy_static", "bytes"]
-rdkafka = ["rdkafka-lib", "lazy_static", "bytes", "futures"]
-warp = ["warp-lib", "lazy_static", "bytes", "http", "hyper"]
+actix = ["actix-web", "async-trait", "bytes", "futures", "http"]
+reqwest = ["reqwest-lib", "async-trait", "bytes"]
+rdkafka = ["rdkafka-lib", "bytes", "futures"]
+warp = ["warp-lib", "bytes", "http", "hyper"]
 
 [dependencies]
 serde = { version = "^1.0", features = ["derive"] }
@@ -38,7 +38,6 @@ reqwest-lib = { version = "^0.11", default-features = false, features = ["rustls
 rdkafka-lib = { version = "^0.25", features = ["cmake-build"], optional = true, package = "rdkafka" }
 warp-lib = { version = "^0.3", optional = true, package = "warp" }
 async-trait = { version = "^0.1.33", optional = true }
-lazy_static = { version = "1.4.0", optional = true }
 bytes = { version = "^1.0", optional = true }
 futures = { version = "^0.3", optional = true }
 http = { version = "0.2", optional = true }

src/binding/actix/headers.rs

@@ -41,8 +41,6 @@
 
 #![deny(broken_intra_doc_links)]
 
-#[macro_use]
-mod headers;
 mod server_request;
 mod server_response;
 

src/binding/actix/mod.rs

@@ -1,11 +1,10 @@
-use super::headers;
+use crate::binding::http::SPEC_VERSION_HEADER;
 use crate::event::SpecVersion;
 use crate::message::{
     BinaryDeserializer, BinarySerializer, Encoding, MessageAttributeValue, MessageDeserializer,
     Result, StructuredDeserializer, StructuredSerializer,
 };
-use crate::{message, Event};
-use actix_web::http::HeaderName;
+use crate::{header_value_to_str, message, Event};
 use actix_web::web::{Bytes, BytesMut};
 use actix_web::{web, HttpMessage, HttpRequest};
 use async_trait::async_trait;
@@ -32,18 +31,21 @@ impl<'a> BinaryDeserializer for HttpRequestDeserializer<'a> {
         }
 
         let spec_version = SpecVersion::try_from(
-            unwrap_optional_header!(self.req.headers(), headers::SPEC_VERSION_HEADER).unwrap()?,
+            self.req
+                .headers()
+                .get(SPEC_VERSION_HEADER)
+                .map(|a| header_value_to_str!(a))
+                .unwrap()?,
         )?;
 
         visitor = visitor.set_spec_version(spec_version.clone())?;
 
         let attributes = spec_version.attribute_names();
 
-        for (hn, hv) in
-            self.req.headers().iter().filter(|(hn, _)| {
-                headers::SPEC_VERSION_HEADER.ne(hn) && hn.as_str().starts_with("ce-")
-            })
-        {
+        for (hn, hv) in self.req.headers().iter().filter(|(hn, _)| {
+            let key = hn.as_str();
+            SPEC_VERSION_HEADER.ne(key) && key.starts_with("ce-")
+        }) {
             let name = &hn.as_str()["ce-".len()..];
 
             if attributes.contains(&name) {
@@ -87,12 +89,7 @@ impl<'a> MessageDeserializer for HttpRequestDeserializer<'a> {
     fn encoding(&self) -> Encoding {
         if self.req.content_type() == "application/cloudevents+json" {
             Encoding::STRUCTURED
-        } else if self
-            .req
-            .headers()
-            .get::<&'static HeaderName>(&super::headers::SPEC_VERSION_HEADER)
-            .is_some()
-        {
+        } else if self.req.headers().get(SPEC_VERSION_HEADER).is_some() {
             Encoding::BINARY
         } else {
             Encoding::UNKNOWN

src/binding/actix/server_request.rs

@@ -1,16 +1,21 @@
-use super::headers;
 use crate::event::SpecVersion;
 use crate::message::{
     BinaryDeserializer, BinarySerializer, MessageAttributeValue, Result, StructuredSerializer,
 };
 use crate::Event;
+use crate::{
+    binding::{
+        http::{header_prefix, SPEC_VERSION_HEADER},
+        CLOUDEVENTS_JSON_HEADER,
+    },
+    str_to_header_value,
+};
 use actix_web::dev::HttpResponseBuilder;
-use actix_web::http::{HeaderName, HeaderValue, StatusCode};
+use actix_web::http::StatusCode;
 use actix_web::HttpResponse;
 use async_trait::async_trait;
 use futures::future::LocalBoxFuture;
 use futures::FutureExt;
-use std::str::FromStr;
 
 /// Wrapper for [`HttpResponseBuilder`] that implements [`StructuredSerializer`] and [`BinarySerializer`].
 pub struct HttpResponseSerializer {
@@ -25,26 +30,20 @@ impl HttpResponseSerializer {
 
 impl BinarySerializer<HttpResponse> for HttpResponseSerializer {
     fn set_spec_version(mut self, spec_version: SpecVersion) -> Result<Self> {
-        self.builder.set_header(
-            headers::SPEC_VERSION_HEADER.clone(),
-            str_to_header_value!(spec_version.as_str())?,
-        );
+        self.builder
+            .set_header(SPEC_VERSION_HEADER, str_to_header_value!(spec_version)?);
         Ok(self)
     }
 
     fn set_attribute(mut self, name: &str, value: MessageAttributeValue) -> Result<Self> {
-        self.builder.set_header(
-            headers::ATTRIBUTES_TO_HEADERS.get(name).unwrap().clone(),
-            str_to_header_value!(value.to_string().as_str())?,
-        );
+        self.builder
+            .set_header(&header_prefix(name), str_to_header_value!(value)?);
         Ok(self)
     }
 
     fn set_extension(mut self, name: &str, value: MessageAttributeValue) -> Result<Self> {
-        self.builder.set_header(
-            attribute_name_to_header!(name)?,
-            str_to_header_value!(value.to_string().as_str())?,
-        );
+        self.builder
+            .set_header(&header_prefix(name), str_to_header_value!(value)?);
         Ok(self)
     }
 
@@ -63,7 +62,7 @@ impl StructuredSerializer<HttpResponse> for HttpResponseSerializer {
             .builder
             .set_header(
                 actix_web::http::header::CONTENT_TYPE,
-                headers::CLOUDEVENTS_JSON_HEADER.clone(),
+                CLOUDEVENTS_JSON_HEADER,
             )
             .body(bytes))
     }

src/binding/actix/server_response.rs

@@ -8,3 +8,53 @@ pub mod rdkafka;
 pub mod reqwest;
 #[cfg(feature = "warp")]
 pub mod warp;
+
+#[cfg(feature = "rdkafka")]
+pub(crate) mod kafka {
+    pub static SPEC_VERSION_HEADER: &str = "ce_specversion";
+    pub fn header_prefix(name: &str) -> String {
+        super::header_prefix("ce_", name)
+    }
+}
+
+#[cfg(any(feature = "actix", feature = "warp", feature = "reqwest"))]
+pub(crate) mod http {
+    pub static SPEC_VERSION_HEADER: &str = "ce-specversion";
+    pub fn header_prefix(name: &str) -> String {
+        super::header_prefix("ce-", name)
+    }
+}
+
+#[cfg(any(feature = "actix", feature = "warp"))]
+#[macro_export]
+macro_rules! str_to_header_value {
+    ($header_value:expr) => {
+        http::header::HeaderValue::from_str(&$header_value.to_string()).map_err(|e| {
+            crate::message::Error::Other {
+                source: Box::new(e),
+            }
+        })
+    };
+}
+
+pub(crate) static CLOUDEVENTS_JSON_HEADER: &str = "application/cloudevents+json";
+pub(crate) static CONTENT_TYPE: &str = "content-type";
+
+fn header_prefix(prefix: &str, name: &str) -> String {
+    if name == "datacontenttype" {
+        CONTENT_TYPE.to_string()
+    } else {
+        [prefix, name].concat()
+    }
+}
+
+#[macro_export]
+macro_rules! header_value_to_str {
+    ($header_value:expr) => {
+        $header_value
+            .to_str()
+            .map_err(|e| crate::message::Error::Other {
+                source: Box::new(e),
+            })
+    };
+}

src/binding/mod.rs

@@ -1,6 +1,6 @@
 use rdkafka_lib as rdkafka;
 
-use super::headers;
+use crate::binding::{kafka::SPEC_VERSION_HEADER, CLOUDEVENTS_JSON_HEADER, CONTENT_TYPE};
 use crate::event::SpecVersion;
 use crate::message::{
     BinaryDeserializer, BinarySerializer, Encoding, MessageAttributeValue, MessageDeserializer,
@@ -47,17 +47,18 @@ impl BinaryDeserializer for ConsumerRecordDeserializer {
         }
 
         let spec_version = SpecVersion::try_from(
-            str::from_utf8(&self.headers.remove(headers::SPEC_VERSION_HEADER).unwrap()[..])
-                .map_err(|e| crate::message::Error::Other {
+            str::from_utf8(&self.headers.remove(SPEC_VERSION_HEADER).unwrap()).map_err(|e| {
+                crate::message::Error::Other {
                     source: Box::new(e),
-                })?,
+                }
+            })?,
         )?;
 
         visitor = visitor.set_spec_version(spec_version.clone())?;
 
         let attributes = spec_version.attribute_names();
 
-        if let Some(hv) = self.headers.remove(headers::CONTENT_TYPE) {
+        if let Some(hv) = self.headers.remove(CONTENT_TYPE) {
             visitor = visitor.set_attribute(
                 "datacontenttype",
                 MessageAttributeValue::String(String::from_utf8(hv).map_err(|e| {
@@ -71,7 +72,7 @@ impl BinaryDeserializer for ConsumerRecordDeserializer {
         for (hn, hv) in self
             .headers
             .into_iter()
-            .filter(|(hn, _)| headers::SPEC_VERSION_HEADER != *hn && hn.starts_with("ce_"))
+            .filter(|(hn, _)| SPEC_VERSION_HEADER != *hn && hn.starts_with("ce_"))
         {
             let name = &hn["ce_".len()..];
 
@@ -120,9 +121,9 @@ impl MessageDeserializer for ConsumerRecordDeserializer {
                 .get("content-type")
                 .map(|s| String::from_utf8(s.to_vec()).ok())
                 .flatten()
-                .map(|s| s.starts_with(headers::CLOUDEVENTS_JSON_HEADER))
+                .map(|s| s.starts_with(CLOUDEVENTS_JSON_HEADER))
                 .unwrap_or(false),
-            self.headers.get(headers::SPEC_VERSION_HEADER),
+            self.headers.get(SPEC_VERSION_HEADER),
         ) {
             (true, _) => Encoding::STRUCTURED,
             (_, Some(_)) => Encoding::BINARY,