New feature: cloudevents-reqwest

Conditionally compile reqwest module when enabled

This resulted in a naming conflict between my desired feature name,
"reqwest", and the optional dependency itself. So I adopted the
convention of prefixing the features with "cloudevents-".

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

Author: jcrossley3

.github/workflows/rust_tests.yml

@@ -91,14 +91,13 @@ jobs:
         CC: musl-gcc
         CXX: g++
 
-    # If wasm, then we test only the main module and cloudevents-sdk-reqwest
     - uses: actions-rs/cargo@v1
       name: "Build"
       if: matrix.target == 'wasm32-unknown-unknown'
       with:
         command: build
         toolchain: ${{ matrix.toolchain }}
-        args: --target wasm32-unknown-unknown --package cloudevents-sdk --package cloudevents-sdk-reqwest
+        args: --target wasm32-unknown-unknown --package cloudevents-sdk
 
     # Build examples
     - uses: actions-rs/cargo@v1

Cargo.toml

@@ -18,6 +18,7 @@ name = "cloudevents"
 
 [features]
 cloudevents-actix = ["actix-web", "async-trait", "lazy_static", "bytes", "futures"]
+cloudevents-reqwest = ["reqwest", "async-trait", "lazy_static", "bytes"]
 
 [dependencies]
 serde = { version = "^1.0", features = ["derive"] }
@@ -29,8 +30,9 @@ url = { version = "^2.1", features = ["serde"] }
 snafu = "^0.6"
 bitflags = "^1.2"
 
-# actix optional deps
+# runtime optional deps
 actix-web = { version = "^3", default-features = false, optional = true }
+reqwest = { version = "^0.11", default-features = false, features = ["rustls-tls"], optional = true }
 async-trait = { version = "^0.1.33", optional = true }
 lazy_static = { version = "1.4.0", optional = true }
 bytes = { version = "^1.0", optional = true }
@@ -50,11 +52,13 @@ claim = "0.3.1"
 version-sync = "^0.9"
 serde_yaml = "0.8"
 
-# actix dev-deps
+# runtime dev-deps
 actix-rt = { version = "^1" }
 url = { version = "^2.1", features = ["serde"] }
 serde_json = { version = "^1.0" }
 chrono = { version = "^0.4", features = ["serde"] }
+mockito = "0.25.1"
+tokio = { version = "^1.0", features = ["full"] }
 
 [workspace]
 members = [

cloudevents-sdk-reqwest/Cargo.toml

@@ -14,7 +14,7 @@ categories = ["web-programming", "encoding", "web-programming::http-client"]
 
 [dependencies]
 async-trait = "^0.1.33"
-cloudevents-sdk = { path = ".." }
+cloudevents-sdk = { version = "0.3.0", path = ".." }
 lazy_static = "1.4.0"
 bytes = "^1.0"
 

example-projects/reqwest-wasm-example/Cargo.toml

@@ -11,8 +11,7 @@ crate-type = ["cdylib"]
 
 [dependencies]
 reqwest = "^0.11"
-cloudevents-sdk = { path = "../.." }
-cloudevents-sdk-reqwest = { path = "../../cloudevents-sdk-reqwest" }
+cloudevents-sdk = { path = "../..", features = ["cloudevents-reqwest"] }
 url = { version = "^2.1" }
 web-sys = { version = "0.3.39", features = ["Window", "Location"] }
 wasm-bindgen-futures = "0.4.12"

example-projects/reqwest-wasm-example/src/lib.rs

@@ -1,5 +1,5 @@
+use cloudevents::reqwest::RequestBuilderExt;
 use cloudevents::{EventBuilder, EventBuilderV10};
-use cloudevents_sdk_reqwest::RequestBuilderExt;
 use wasm_bindgen::prelude::*;
 
 #[wasm_bindgen]

src/actix/server_request.rs

@@ -1,14 +1,14 @@
 use super::headers;
-use actix_web::http::HeaderName;
-use actix_web::web::{Bytes, BytesMut};
-use actix_web::{web, HttpMessage, HttpRequest};
-use async_trait::async_trait;
 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 actix_web::web::{Bytes, BytesMut};
+use actix_web::{web, HttpMessage, HttpRequest};
+use async_trait::async_trait;
 use futures::StreamExt;
 use std::convert::TryFrom;
 
@@ -145,8 +145,8 @@ mod tests {
     use super::*;
     use actix_web::test;
 
-    use chrono::Utc;
     use crate::{EventBuilder, EventBuilderV10};
+    use chrono::Utc;
     use serde_json::json;
 
     #[actix_rt::test]

src/lib.rs

@@ -42,6 +42,8 @@
 
 #[cfg(feature = "cloudevents-actix")]
 pub mod actix;
+#[cfg(feature = "cloudevents-reqwest")]
+pub mod reqwest;
 
 pub mod event;
 pub mod message;

src/reqwest/client_request.rs

@@ -0,0 +1,199 @@
+use super::headers;
+use crate::event::SpecVersion;
+use crate::message::{
+    BinaryDeserializer, BinarySerializer, MessageAttributeValue, Result, StructuredSerializer,
+};
+use crate::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 fill a [`RequestBuilder`] with an [`Event`].
+pub fn event_to_request(event: Event, request_builder: RequestBuilder) -> Result<RequestBuilder> {
+    BinaryDeserializer::deserialize_binary(event, RequestSerializer::new(request_builder))
+}
+
+/// Extension Trait for [`RequestBuilder`] which acts as a wrapper for the function [`event_to_request()`].
+///
+/// This trait is sealed and cannot be implemented for types outside of this crate.
+pub trait RequestBuilderExt: private::Sealed {
+    /// Write in this [`RequestBuilder`] the provided [`Event`]. Similar to invoking [`Event`].
+    fn event(self, event: Event) -> Result<RequestBuilder>;
+}
+
+impl RequestBuilderExt for RequestBuilder {
+    fn event(self, event: Event) -> Result<RequestBuilder> {
+        event_to_request(event, self)
+    }
+}
+
+// Sealing the RequestBuilderExt
+mod private {
+    pub trait Sealed {}
+    impl Sealed for reqwest::RequestBuilder {}
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use mockito::{mock, Matcher};
+
+    use crate::message::StructuredDeserializer;
+    use crate::{EventBuilder, EventBuilderV10};
+    use serde_json::json;
+
+    #[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 = EventBuilderV10::new()
+            .id("0001")
+            .ty("example.test")
+            .source("http://localhost/")
+            .extension("someint", "10")
+            .build()
+            .unwrap();
+
+        let client = reqwest::Client::new();
+        client
+            .post(&url)
+            .event(input)
+            .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 = EventBuilderV10::new()
+            .id("0001")
+            .ty("example.test")
+            .source("http://localhost/")
+            .data("application/json", j.clone())
+            .extension("someint", "10")
+            .build()
+            .unwrap();
+
+        let client = reqwest::Client::new();
+
+        client
+            .post(&url)
+            .event(input)
+            .unwrap()
+            .send()
+            .await
+            .unwrap();
+
+        m.assert();
+    }
+
+    #[tokio::test]
+    async fn test_structured_request_with_full_data() {
+        let j = json!({"hello": "world"});
+
+        let input = EventBuilderV10::new()
+            .id("0001")
+            .ty("example.test")
+            .source("http://localhost")
+            .data("application/json", j.clone())
+            .extension("someint", "10")
+            .build()
+            .unwrap();
+
+        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();
+    }
+}