Merge pull request cloudevents#33 from slinkydeveloper/fetch_implementation

Reqwest integration

Author: Lazzaretti

Cargo.lock

@@ -33,5 +33,6 @@ name = "cloudevents"
 [workspace]
 members = [
   ".",
-  "cloudevents-sdk-actix-web"
+  "cloudevents-sdk-actix-web",
+  "cloudevents-sdk-reqwest"
 ]

Cargo.toml

@@ -20,6 +20,7 @@ Work in progress SDK for [CloudEvents](https://github.com/cloudevents/spec)
 
 * `cloudevents-sdk`: Provides Event data structure, JSON Event format implementation
 * `cloudevents-sdk-actix-web`: Integration with [Actix Web](https://github.com/actix/actix-web)
+* `cloudevents-sdk-reqwest`: Integration with [reqwest](https://github.com/seanmonstar/reqwest)
 
 ## Development & Contributing
 

README.md

@@ -4,8 +4,8 @@ use actix_web::web::{Bytes, BytesMut};
 use actix_web::{web, HttpMessage, HttpRequest};
 use cloudevents::event::SpecVersion;
 use cloudevents::message::{
-    BinaryDeserializer, BinarySerializer, Encoding, Error, MessageAttributeValue,
-    MessageDeserializer, StructuredDeserializer, StructuredSerializer,
+    BinaryDeserializer, BinarySerializer, Encoding, MessageAttributeValue, MessageDeserializer,
+    Result, StructuredDeserializer, StructuredSerializer,
 };
 use cloudevents::{message, Event};
 use futures::StreamExt;
@@ -24,10 +24,7 @@ impl HttpRequestDeserializer<'_> {
 }
 
 impl<'a> BinaryDeserializer for HttpRequestDeserializer<'a> {
-    fn deserialize_binary<R: Sized, V: BinarySerializer<R>>(
-        self,
-        mut visitor: V,
-    ) -> Result<R, Error> {
+    fn deserialize_binary<R: Sized, V: BinarySerializer<R>>(self, mut visitor: V) -> Result<R> {
         if self.encoding() != Encoding::BINARY {
             return Err(message::Error::WrongEncoding {});
         }
@@ -36,7 +33,7 @@ impl<'a> BinaryDeserializer for HttpRequestDeserializer<'a> {
             unwrap_optional_header!(self.req.headers(), headers::SPEC_VERSION_HEADER).unwrap()?,
         )?;
 
-        visitor.set_spec_version(spec_version.clone())?;
+        visitor = visitor.set_spec_version(spec_version.clone())?;
 
         let attributes = cloudevents::event::spec_version::ATTRIBUTE_NAMES
             .get(&spec_version)
@@ -50,20 +47,20 @@ impl<'a> BinaryDeserializer for HttpRequestDeserializer<'a> {
             let name = &hn.as_str()["ce-".len()..];
 
             if attributes.contains(&name) {
-                visitor.set_attribute(
+                visitor = visitor.set_attribute(
                     name,
                     MessageAttributeValue::String(String::from(header_value_to_str!(hv)?)),
                 )?
             } else {
-                visitor.set_extension(
+                visitor = visitor.set_extension(
                     name,
                     MessageAttributeValue::String(String::from(header_value_to_str!(hv)?)),
                 )?
             }
         }
 
         if let Some(hv) = self.req.headers().get("content-type") {
-            visitor.set_attribute(
+            visitor = visitor.set_attribute(
                 "datacontenttype",
                 MessageAttributeValue::String(String::from(header_value_to_str!(hv)?)),
             )?
@@ -78,10 +75,7 @@ impl<'a> BinaryDeserializer for HttpRequestDeserializer<'a> {
 }
 
 impl<'a> StructuredDeserializer for HttpRequestDeserializer<'a> {
-    fn deserialize_structured<R: Sized, V: StructuredSerializer<R>>(
-        self,
-        visitor: V,
-    ) -> Result<R, Error> {
+    fn deserialize_structured<R: Sized, V: StructuredSerializer<R>>(self, visitor: V) -> Result<R> {
         if self.encoding() != Encoding::STRUCTURED {
             return Err(message::Error::WrongEncoding {});
         }
@@ -110,7 +104,7 @@ impl<'a> MessageDeserializer for HttpRequestDeserializer<'a> {
 pub async fn request_to_event(
     req: &HttpRequest,
     mut payload: web::Payload,
-) -> Result<Event, actix_web::error::Error> {
+) -> std::result::Result<Event, actix_web::error::Error> {
     let mut bytes = BytesMut::new();
     while let Some(item) = payload.next().await {
         bytes.extend_from_slice(&item?);

cloudevents-sdk-actix-web/src/server_request.rs

@@ -4,8 +4,7 @@ use actix_web::http::{HeaderName, HeaderValue};
 use actix_web::HttpResponse;
 use cloudevents::event::SpecVersion;
 use cloudevents::message::{
-    BinaryDeserializer, BinarySerializer, Error, MessageAttributeValue, SerializationResult,
-    StructuredSerializer,
+    BinaryDeserializer, BinarySerializer, MessageAttributeValue, Result, StructuredSerializer,
 };
 use cloudevents::Event;
 use std::str::FromStr;
@@ -21,41 +20,41 @@ impl HttpResponseSerializer {
 }
 
 impl BinarySerializer<HttpResponse> for HttpResponseSerializer {
-    fn set_spec_version(&mut self, spec_version: SpecVersion) -> SerializationResult {
+    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())?,
         );
-        SerializationResult::Ok(())
+        Ok(self)
     }
 
-    fn set_attribute(&mut self, name: &str, value: MessageAttributeValue) -> SerializationResult {
+    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())?,
         );
-        SerializationResult::Ok(())
+        Ok(self)
     }
 
-    fn set_extension(&mut self, name: &str, value: MessageAttributeValue) -> SerializationResult {
+    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())?,
         );
-        SerializationResult::Ok(())
+        Ok(self)
     }
 
-    fn end_with_data(mut self, bytes: Vec<u8>) -> Result<HttpResponse, Error> {
+    fn end_with_data(mut self, bytes: Vec<u8>) -> Result<HttpResponse> {
         Ok(self.builder.body(bytes))
     }
 
-    fn end(mut self) -> Result<HttpResponse, Error> {
+    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, Error> {
+    fn set_structured_event(mut self, bytes: Vec<u8>) -> Result<HttpResponse> {
         Ok(self
             .builder
             .set_header(
@@ -70,7 +69,7 @@ impl StructuredSerializer<HttpResponse> for HttpResponseSerializer {
 pub async fn event_to_response(
     event: Event,
     response: HttpResponseBuilder,
-) -> Result<HttpResponse, actix_web::error::Error> {
+) -> std::result::Result<HttpResponse, actix_web::error::Error> {
     BinaryDeserializer::deserialize_binary(event, HttpResponseSerializer::new(response))
         .map_err(actix_web::error::ErrorBadRequest)
 }

cloudevents-sdk-actix-web/src/server_response.rs

@@ -0,0 +1,19 @@
+[package]
+name = "cloudevents-sdk-reqwest"
+version = "0.1.0"
+authors = ["Francesco Guardiani <[email protected]>"]
+edition = "2018"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+cloudevents-sdk = { path = ".." }
+reqwest = { version = "0.10" }
+lazy_static = "1.4.0"
+bytes = "^0.5"
+serde_json = "^1.0"
+url = { version = "^2.1", features = ["serde"] }
+
+[dev-dependencies]
+mockito = "0.25.1"
+tokio = { version = "^0.2", features = ["full"] }

cloudevents-sdk-reqwest/Cargo.toml

@@ -0,0 +1,172 @@
+use super::headers;
+use cloudevents::event::SpecVersion;
+use cloudevents::message::{
+    BinaryDeserializer, BinarySerializer, MessageAttributeValue, Result, StructuredSerializer,
+};
+use cloudevents::Event;
+use reqwest::RequestBuilder;
+use std::str::FromStr;
+
+/// Wrapper for [`RequestBuilder`] that implements [`StructuredSerializer`] & [`BinarySerializer`] traits
+pub struct RequestSerializer {
+    req: RequestBuilder,
+}
+
+impl RequestSerializer {
+    pub fn new(req: RequestBuilder) -> RequestSerializer {
+        RequestSerializer { req }
+    }
+}
+
+impl BinarySerializer<RequestBuilder> for RequestSerializer {
+    fn set_spec_version(mut self, spec_version: SpecVersion) -> Result<Self> {
+        self.req = self
+            .req
+            .header(headers::SPEC_VERSION_HEADER.clone(), spec_version.as_str());
+        Ok(self)
+    }
+
+    fn set_attribute(mut self, name: &str, value: MessageAttributeValue) -> Result<Self> {
+        self.req = self.req.header(
+            headers::ATTRIBUTES_TO_HEADERS.get(name).unwrap().clone(),
+            value.to_string(),
+        );
+        Ok(self)
+    }
+
+    fn set_extension(mut self, name: &str, value: MessageAttributeValue) -> Result<Self> {
+        self.req = self
+            .req
+            .header(attribute_name_to_header!(name)?, value.to_string());
+        Ok(self)
+    }
+
+    fn end_with_data(self, bytes: Vec<u8>) -> Result<RequestBuilder> {
+        Ok(self.req.body(bytes))
+    }
+
+    fn end(self) -> Result<RequestBuilder> {
+        Ok(self.req)
+    }
+}
+
+impl StructuredSerializer<RequestBuilder> for RequestSerializer {
+    fn set_structured_event(self, bytes: Vec<u8>) -> Result<RequestBuilder> {
+        Ok(self
+            .req
+            .header(
+                reqwest::header::CONTENT_TYPE,
+                headers::CLOUDEVENTS_JSON_HEADER.clone(),
+            )
+            .body(bytes))
+    }
+}
+
+/// Method to transform an incoming [`HttpRequest`] to [`Event`]
+pub fn event_to_request(event: Event, request_builder: RequestBuilder) -> Result<RequestBuilder> {
+    BinaryDeserializer::deserialize_binary(event, RequestSerializer::new(request_builder))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use mockito::{mock, Matcher};
+
+    use cloudevents::message::StructuredDeserializer;
+    use cloudevents::EventBuilder;
+    use serde_json::json;
+    use url::Url;
+
+    #[tokio::test]
+    async fn test_request() {
+        let url = mockito::server_url();
+        let m = mock("POST", "/")
+            .match_header("ce-specversion", "1.0")
+            .match_header("ce-id", "0001")
+            .match_header("ce-type", "example.test")
+            .match_header("ce-source", "http://localhost/")
+            .match_header("ce-someint", "10")
+            .match_body(Matcher::Missing)
+            .create();
+
+        let input = EventBuilder::new()
+            .id("0001")
+            .ty("example.test")
+            .source(Url::from_str("http://localhost/").unwrap())
+            .extension("someint", "10")
+            .build();
+
+        let client = reqwest::Client::new();
+        event_to_request(input, client.post(&url))
+            .unwrap()
+            .send()
+            .await
+            .unwrap();
+
+        m.assert();
+    }
+
+    #[tokio::test]
+    async fn test_request_with_full_data() {
+        let j = json!({"hello": "world"});
+
+        let url = mockito::server_url();
+        let m = mock("POST", "/")
+            .match_header("ce-specversion", "1.0")
+            .match_header("ce-id", "0001")
+            .match_header("ce-type", "example.test")
+            .match_header("ce-source", "http://localhost/")
+            .match_header("content-type", "application/json")
+            .match_header("ce-someint", "10")
+            .match_body(Matcher::Exact(j.to_string()))
+            .create();
+
+        let input = EventBuilder::new()
+            .id("0001")
+            .ty("example.test")
+            .source(Url::from_str("http://localhost").unwrap())
+            .data("application/json", j.clone())
+            .extension("someint", "10")
+            .build();
+
+        let client = reqwest::Client::new();
+        event_to_request(input, client.post(&url))
+            .unwrap()
+            .send()
+            .await
+            .unwrap();
+
+        m.assert();
+    }
+
+    #[tokio::test]
+    async fn test_structured_request_with_full_data() {
+        let j = json!({"hello": "world"});
+
+        let input = EventBuilder::new()
+            .id("0001")
+            .ty("example.test")
+            .source(Url::from_str("http://localhost").unwrap())
+            .data("application/json", j.clone())
+            .extension("someint", "10")
+            .build();
+
+        let url = mockito::server_url();
+        let m = mock("POST", "/")
+            .match_header("content-type", "application/cloudevents+json")
+            .match_body(Matcher::Exact(serde_json::to_string(&input).unwrap()))
+            .create();
+
+        let client = reqwest::Client::new();
+        StructuredDeserializer::deserialize_structured(
+            input,
+            RequestSerializer::new(client.post(&url)),
+        )
+        .unwrap()
+        .send()
+        .await
+        .unwrap();
+
+        m.assert();
+    }
+}