Factor common response Builder trait out for actix & warp

This proved very difficult due to the different ownership behavior
between each builder, i.e. warp is a consuming builder, actix isn't.

With the use of the Rc/Cell/RefCell, I worry that this is now more
complex than it was, and for all I know, I've introduced a memory leak
somewhere. :)

Since the reqwest builder is also consuming, it should be able to
follow the same "adapter" pattern as warp. Unfortunately, the reqwest
builder doesn't implement the Default trait, so I can't use take() and
I've yet to come up with another solution.

Since 2/3 of the builders are consumers, it's possible we might
simplify the code by having the new Builder trait reflect that,
i.e. using self instead of &self in the fn params. We'll investigate
that next.

For now, I'm just happy to have 2 builders sharing some formerly
redundant behavior.

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

Author: jcrossley3

src/binding/actix/mod.rs

@@ -48,4 +48,3 @@ pub use server_request::request_to_event;
 pub use server_request::HttpRequestExt;
 pub use server_response::event_to_response;
 pub use server_response::HttpResponseBuilderExt;
-pub use server_response::HttpResponseSerializer;

src/binding/actix/server_response.rs

@@ -1,79 +1,31 @@
-use crate::event::SpecVersion;
-use crate::message::{
-    BinaryDeserializer, BinarySerializer, MessageAttributeValue, Result, StructuredSerializer,
-};
+use crate::binding::http::{Builder, Serializer};
+use crate::message::{BinaryDeserializer, Result};
 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::StatusCode;
 use actix_web::HttpResponse;
 use async_trait::async_trait;
 use futures::future::LocalBoxFuture;
 use futures::FutureExt;
 
-/// Wrapper for [`HttpResponseBuilder`] that implements [`StructuredSerializer`] and [`BinarySerializer`].
-pub struct HttpResponseSerializer {
-    builder: HttpResponseBuilder,
-}
-
-impl HttpResponseSerializer {
-    pub fn new(builder: HttpResponseBuilder) -> HttpResponseSerializer {
-        HttpResponseSerializer { builder }
-    }
-}
-
-impl BinarySerializer<HttpResponse> for HttpResponseSerializer {
-    fn set_spec_version(mut self, spec_version: SpecVersion) -> Result<Self> {
-        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(&header_prefix(name), str_to_header_value!(value)?);
-        Ok(self)
+impl Builder<HttpResponse> for HttpResponseBuilder {
+    fn header(&mut self, key: &str, value: http::header::HeaderValue) {
+        self.set_header(key, value);
     }
-
-    fn set_extension(mut self, name: &str, value: MessageAttributeValue) -> Result<Self> {
-        self.builder
-            .set_header(&header_prefix(name), str_to_header_value!(value)?);
-        Ok(self)
-    }
-
-    fn end_with_data(mut self, bytes: Vec<u8>) -> Result<HttpResponse> {
-        Ok(self.builder.body(bytes))
+    fn body(&mut self, bytes: Vec<u8>) -> Result<HttpResponse> {
+        Ok(HttpResponseBuilder::body(self, bytes))
     }
-
-    fn end(mut self) -> Result<HttpResponse> {
-        Ok(self.builder.finish())
-    }
-}
-
-impl StructuredSerializer<HttpResponse> for HttpResponseSerializer {
-    fn set_structured_event(mut self, bytes: Vec<u8>) -> Result<HttpResponse> {
-        Ok(self
-            .builder
-            .set_header(
-                actix_web::http::header::CONTENT_TYPE,
-                CLOUDEVENTS_JSON_HEADER,
-            )
-            .body(bytes))
+    fn finish(&mut self) -> Result<HttpResponse> {
+        Ok(HttpResponseBuilder::finish(self))
     }
 }
 
 /// Method to fill an [`HttpResponseBuilder`] with an [`Event`].
-pub async fn event_to_response(
+pub async fn event_to_response<T: Builder<HttpResponse> + 'static>(
     event: Event,
-    response: HttpResponseBuilder,
+    response: T,
 ) -> std::result::Result<HttpResponse, actix_web::error::Error> {
-    BinaryDeserializer::deserialize_binary(event, HttpResponseSerializer::new(response))
+    BinaryDeserializer::deserialize_binary(event, Serializer::new(response))
         .map_err(actix_web::error::ErrorBadRequest)
 }
 

src/binding/http/mod.rs

@@ -7,6 +7,10 @@ use crate::{
 };
 use deserializer::Deserializer;
 pub use headers::Headers;
+mod serializer;
+
+pub use serializer::Builder;
+pub use serializer::Serializer;
 
 pub static SPEC_VERSION_HEADER: &str = "ce-specversion";
 

src/binding/http/serializer.rs

@@ -0,0 +1,77 @@
+use std::{cell::RefCell, rc::Rc};
+
+use crate::binding::{
+    http::{header_prefix, SPEC_VERSION_HEADER},
+    CLOUDEVENTS_JSON_HEADER,
+};
+use crate::event::SpecVersion;
+use crate::message::{BinarySerializer, MessageAttributeValue, Result, StructuredSerializer};
+
+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 trait Builder<R> {
+    fn header(&mut self, key: &str, value: http::header::HeaderValue);
+    fn body(&mut self, bytes: Vec<u8>) -> Result<R>;
+    fn finish(&mut self) -> Result<R>;
+}
+
+pub struct Serializer<T> {
+    builder: Rc<RefCell<dyn Builder<T>>>,
+}
+
+impl<T> Serializer<T> {
+    pub fn new<B: Builder<T> + 'static>(delegate: B) -> Serializer<T> {
+        let builder = Rc::new(RefCell::new(delegate));
+        Serializer { builder }
+    }
+}
+
+impl<T> BinarySerializer<T> for Serializer<T> {
+    fn set_spec_version(self, spec_version: SpecVersion) -> Result<Self> {
+        self.builder
+            .borrow_mut()
+            .header(SPEC_VERSION_HEADER, str_to_header_value!(spec_version)?);
+        Ok(self)
+    }
+
+    fn set_attribute(self, name: &str, value: MessageAttributeValue) -> Result<Self> {
+        self.builder
+            .borrow_mut()
+            .header(&header_prefix(name), str_to_header_value!(value)?);
+        Ok(self)
+    }
+
+    fn set_extension(self, name: &str, value: MessageAttributeValue) -> Result<Self> {
+        self.builder
+            .borrow_mut()
+            .header(&header_prefix(name), str_to_header_value!(value)?);
+        Ok(self)
+    }
+
+    fn end_with_data(self, bytes: Vec<u8>) -> Result<T> {
+        self.builder.borrow_mut().body(bytes)
+    }
+
+    fn end(self) -> Result<T> {
+        self.builder.borrow_mut().finish()
+    }
+}
+
+impl<T> StructuredSerializer<T> for Serializer<T> {
+    fn set_structured_event(self, bytes: Vec<u8>) -> Result<T> {
+        let mut builder = self.builder.borrow_mut();
+        builder.header(
+            http::header::CONTENT_TYPE.as_str(),
+            http::HeaderValue::from_static(CLOUDEVENTS_JSON_HEADER),
+        );
+        builder.body(bytes)
+    }
+}

src/binding/mod.rs

@@ -19,18 +19,6 @@ pub(crate) mod kafka {
     }
 }
 
-#[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";
 

src/binding/reqwest/client_request.rs

@@ -11,6 +11,15 @@ use crate::message::{
 use crate::Event;
 use reqwest::RequestBuilder;
 
+// TODO: Ideally, we'd only need to implement binding::http::Builder
+// for reqwest::RequestBuilder here, but because the latter is a
+// consuming builder, we'd need an intermediate struct similar to
+// warp's to adapt that interface. Unfortunately, the reqwest builder
+// doesn't implement the Default trait, so I can't use take() as
+// warp's Adapter does, and I've yet to come up with another
+// solution. So for now, we continue to implement BinarySerializer
+// directly in here.
+
 /// Wrapper for [`RequestBuilder`] that implements [`StructuredSerializer`] & [`BinarySerializer`] traits.
 pub struct RequestSerializer {
     req: RequestBuilder,

src/binding/warp/server_response.rs

@@ -1,83 +1,40 @@
+use std::cell::Cell;
+
 use warp_lib as warp;
 
-use crate::binding::{
-    http::{header_prefix, SPEC_VERSION_HEADER},
-    CLOUDEVENTS_JSON_HEADER,
-};
-use crate::event::SpecVersion;
-use crate::message::{
-    BinaryDeserializer, BinarySerializer, Error, MessageAttributeValue, Result,
-    StructuredSerializer,
-};
-use crate::{str_to_header_value, Event};
+use crate::binding::http::{Builder, Serializer};
+use crate::message::{BinaryDeserializer, Error, Result};
+use crate::Event;
 
 use warp::hyper::Body;
 use warp::reply::Response;
 
-use http::response::Builder;
-
-pub struct ResponseSerializer {
-    builder: Builder,
-}
-
-impl ResponseSerializer {
-    fn new() -> Self {
-        ResponseSerializer {
-            builder: http::Response::builder(),
-        }
-    }
+struct Adapter {
+    builder: Cell<http::response::Builder>,
 }
 
-impl BinarySerializer<Response> for ResponseSerializer {
-    fn set_spec_version(mut self, spec_version: SpecVersion) -> Result<Self> {
-        self.builder = self
-            .builder
-            .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 = self
-            .builder
-            .header(&header_prefix(name), str_to_header_value!(value)?);
-        Ok(self)
-    }
-
-    fn set_extension(mut self, name: &str, value: MessageAttributeValue) -> Result<Self> {
-        self.builder = self
-            .builder
-            .header(&header_prefix(name), str_to_header_value!(value)?);
-        Ok(self)
+impl Builder<Response> for Adapter {
+    fn header(&mut self, key: &str, value: http::header::HeaderValue) {
+        self.builder.set(self.builder.take().header(key, value));
     }
-
-    fn end_with_data(self, bytes: Vec<u8>) -> Result<Response> {
+    fn body(&mut self, bytes: Vec<u8>) -> Result<Response> {
         self.builder
+            .take()
             .body(Body::from(bytes))
             .map_err(|e| crate::message::Error::Other {
                 source: Box::new(e),
             })
     }
-
-    fn end(self) -> Result<Response> {
-        self.builder
-            .body(Body::empty())
-            .map_err(|e| crate::message::Error::Other {
-                source: Box::new(e),
-            })
-    }
-}
-
-impl StructuredSerializer<Response> for ResponseSerializer {
-    fn set_structured_event(self, bytes: Vec<u8>) -> Result<Response> {
-        self.builder
-            .header(http::header::CONTENT_TYPE, CLOUDEVENTS_JSON_HEADER)
-            .body(Body::from(bytes))
-            .map_err(|e| crate::message::Error::Other {
-                source: Box::new(e),
-            })
+    fn finish(&mut self) -> Result<Response> {
+        self.body(Vec::new())
     }
 }
 
 pub fn event_to_response(event: Event) -> std::result::Result<Response, Error> {
-    BinaryDeserializer::deserialize_binary(event, ResponseSerializer::new())
+    BinaryDeserializer::deserialize_binary(
+        event,
+        Serializer::new(Adapter {
+            builder: Cell::new(http::Response::builder()),
+        }),
+    )
 }