Merge pull request #159 from sunli829/master

New feature: cloudevents-poem

Author: jcrossley3

.github/workflows/rust_tests.yml

@@ -137,3 +137,12 @@ jobs:
         command: build
         toolchain: ${{ matrix.toolchain }}
         args: --target ${{ matrix.target }} --manifest-path ./example-projects/axum-example/Cargo.toml
+
+    - uses: actions-rs/cargo@v1
+      name: "Build poem-example"
+      if: matrix.target == 'x86_64-unknown-linux-gnu' && matrix.toolchain == 'stable'
+      with:
+        command: build
+        toolchain: ${{ matrix.toolchain }}
+        args: --target ${{ matrix.target }} --manifest-path ./example-projects/poem-example/Cargo.toml
+

Cargo.toml

@@ -22,7 +22,8 @@ actix = ["actix-web", "async-trait", "bytes", "futures", "http"]
 reqwest = ["reqwest-lib", "async-trait", "bytes", "http"]
 rdkafka = ["rdkafka-lib", "bytes", "futures"]
 warp = ["warp-lib", "bytes", "http", "hyper"]
-axum = ["bytes", "http", "hyper", "axum-lib", "http-body",  "async-trait"]
+axum = ["bytes", "http", "hyper", "axum-lib", "http-body", "async-trait"]
+poem = ["bytes", "http", "poem-lib", "hyper", "async-trait"]
 
 [dependencies]
 serde = { version = "^1.0", features = ["derive"] }
@@ -44,8 +45,9 @@ bytes = { version = "^1.0", optional = true }
 futures = { version = "^0.3", optional = true }
 http = { version = "0.2", optional = true }
 hyper = { version = "^0.14", optional = true }
-axum-lib = { version = "^0.2", optional = true, package="axum"}
-http-body = { version = "^0.4", optional = true}
+axum-lib = { version = "^0.2", optional = true, package = "axum" }
+http-body = { version = "^0.4", optional = true }
+poem-lib = { version = "1.0.23", optional = true, package = "poem" }
 
 [target."cfg(not(target_arch = \"wasm32\"))".dependencies]
 hostname = "^0.3"

example-projects/poem-example/Cargo.toml

@@ -0,0 +1,16 @@
+[package]
+name = "poem-example"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+cloudevents-sdk = { path = "../..", features = ["poem"] }
+tokio = { version = "1.13", features = ["macros", "rt-multi-thread"] }
+tracing = "0.1"
+poem = { version = "1.0.23" }
+tracing-subscriber = "0.2"
+serde_json = "1.0"
+
+[dev-dependencies]
+chrono = { version = "0.4", features = ["serde"] }
+

example-projects/poem-example/README.md

@@ -0,0 +1,23 @@
+To run the server:
+
+```console
+cargo run
+```
+
+To test a GET:
+
+```console
+curl http://localhost:8080
+```
+
+To test a POST:
+
+```console
+curl -d '{"hello": "world"}' \
+  -H'content-type: application/json' \
+  -H'ce-specversion: 1.0' \
+  -H'ce-id: 1' \
+  -H'ce-source: http://cloudevents.io' \
+  -H'ce-type: dev.knative.example' \
+  http://localhost:8080
+```

example-projects/poem-example/src/main.rs

@@ -0,0 +1,121 @@
+use cloudevents::{Event, EventBuilder, EventBuilderV10};
+use poem::error::InternalServerError;
+use poem::listener::TcpListener;
+use poem::middleware::Tracing;
+use poem::{get, handler, Endpoint, EndpointExt, Response, Result, Route, Server};
+use serde_json::json;
+
+#[handler]
+async fn get_event() -> Result<Event> {
+    let event = EventBuilderV10::new()
+        .id("1")
+        .source("url://example_response/")
+        .ty("example.ce")
+        .data(
+            "application/json",
+            json!({
+                "name": "John Doe",
+                "age": 43,
+                "phones": [
+                    "+44 1234567",
+                    "+44 2345678"
+                ]
+            }),
+        )
+        .build()
+        .map_err(InternalServerError)?;
+    Ok(event)
+}
+
+#[handler]
+async fn post_event(event: Event) -> Event {
+    tracing::debug!("received cloudevent {}", &event);
+    event
+}
+
+fn echo_app() -> impl Endpoint<Output = Response> {
+    Route::new()
+        .at("/", get(get_event).post(post_event))
+        .with(Tracing)
+}
+
+#[tokio::main]
+async fn main() -> Result<(), std::io::Error> {
+    if std::env::var("RUST_LOG").is_err() {
+        std::env::set_var("RUST_LOG", "poem=debug")
+    }
+    tracing_subscriber::fmt::init();
+
+    let server = Server::new(TcpListener::bind("127.0.0.1:8080")).await?;
+    server.run(echo_app()).await
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use chrono::Utc;
+    use poem::http::Method;
+    use poem::{Body, Request};
+    use serde_json::json;
+
+    #[tokio::test]
+    async fn poem_test() {
+        if std::env::var("RUST_LOG").is_err() {
+            std::env::set_var("RUST_LOG", "poem_example=debug")
+        }
+        tracing_subscriber::fmt::init();
+
+        let app = echo_app();
+        let time = Utc::now();
+        let j = json!({"hello": "world"});
+        let request = Request::builder()
+            .method(Method::POST)
+            .header("ce-specversion", "1.0")
+            .header("ce-id", "0001")
+            .header("ce-type", "example.test")
+            .header("ce-source", "http://localhost/")
+            .header("ce-someint", "10")
+            .header("ce-time", time.to_rfc3339())
+            .header("content-type", "application/json")
+            .body(Body::from_json(&j).unwrap());
+
+        let resp: Response = app.call(request).await;
+        assert_eq!(
+            resp.headers()
+                .get("ce-specversion")
+                .unwrap()
+                .to_str()
+                .unwrap(),
+            "1.0"
+        );
+        assert_eq!(
+            resp.headers().get("ce-id").unwrap().to_str().unwrap(),
+            "0001"
+        );
+        assert_eq!(
+            resp.headers().get("ce-type").unwrap().to_str().unwrap(),
+            "example.test"
+        );
+        assert_eq!(
+            resp.headers().get("ce-source").unwrap().to_str().unwrap(),
+            "http://localhost/"
+        );
+        assert_eq!(
+            resp.headers()
+                .get("content-type")
+                .unwrap()
+                .to_str()
+                .unwrap(),
+            "application/json"
+        );
+        assert_eq!(
+            resp.headers().get("ce-someint").unwrap().to_str().unwrap(),
+            "10"
+        );
+
+        assert_eq!(
+            j.to_string().as_bytes(),
+            resp.into_body().into_vec().await.unwrap()
+        );
+    }
+}

src/binding/mod.rs

@@ -9,9 +9,12 @@ pub mod axum;
     feature = "actix",
     feature = "warp",
     feature = "reqwest",
-    feature = "axum"
+    feature = "axum",
+    feature = "poem"
 ))]
 pub mod http;
+#[cfg(feature = "poem")]
+pub mod poem;
 #[cfg(feature = "rdkafka")]
 pub mod rdkafka;
 #[cfg(feature = "reqwest")]

src/binding/poem/extractor.rs

@@ -0,0 +1,112 @@
+use async_trait::async_trait;
+use poem_lib::error::ReadBodyError;
+use poem_lib::http::StatusCode;
+use poem_lib::{FromRequest, IntoResponse, Request, RequestBody, Response};
+
+use crate::binding::http::to_event;
+use crate::Event;
+
+#[derive(Debug)]
+pub enum ParseEventError {
+    ReadBody(ReadBodyError),
+    ParseEvent(crate::message::Error),
+}
+
+impl From<ReadBodyError> for ParseEventError {
+    fn from(err: ReadBodyError) -> Self {
+        ParseEventError::ReadBody(err)
+    }
+}
+
+impl From<crate::message::Error> for ParseEventError {
+    fn from(err: crate::message::Error) -> Self {
+        ParseEventError::ParseEvent(err)
+    }
+}
+
+impl IntoResponse for ParseEventError {
+    fn into_response(self) -> Response {
+        match self {
+            ParseEventError::ReadBody(err) => err.into_response(),
+            ParseEventError::ParseEvent(err) => (StatusCode::BAD_REQUEST, err.to_string()).into(),
+        }
+    }
+}
+
+#[async_trait]
+impl<'a> FromRequest<'a> for Event {
+    type Error = ParseEventError;
+
+    async fn from_request(req: &'a Request, body: &mut RequestBody) -> Result<Self, Self::Error> {
+        Ok(to_event(req.headers(), body.take()?.into_vec().await?)?)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::test::fixtures;
+    use poem_lib::http::Method;
+
+    #[tokio::test]
+    async fn test_request() {
+        let expected = fixtures::v10::minimal_string_extension();
+
+        let req = Request::builder()
+            .method(Method::POST)
+            .header("ce-specversion", "1.0")
+            .header("ce-id", "0001")
+            .header("ce-type", "test_event.test_application")
+            .header("ce-source", "http://localhost/")
+            .header("ce-someint", "10")
+            .finish();
+        let (req, mut body) = req.split();
+        let result = Event::from_request(&req, &mut body).await.unwrap();
+
+        assert_eq!(expected, result);
+    }
+
+    #[tokio::test]
+    async fn test_bad_request() {
+        let req = Request::builder()
+            .method(Method::POST)
+            .header("ce-specversion", "BAD SPECIFICATION")
+            .header("ce-id", "0001")
+            .header("ce-type", "example.test")
+            .header("ce-source", "http://localhost/")
+            .header("ce-someint", "10")
+            .header("ce-time", fixtures::time().to_rfc3339())
+            .finish();
+
+        let (req, mut body) = req.split();
+        let resp = Event::from_request(&req, &mut body).await.into_response();
+        assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
+        assert_eq!(
+            resp.into_body().into_string().await.unwrap(),
+            "Invalid specversion BAD SPECIFICATION"
+        );
+    }
+
+    #[tokio::test]
+    async fn test_request_with_full_data() {
+        let expected = fixtures::v10::full_binary_json_data_string_extension();
+
+        let req = Request::builder()
+            .method(Method::POST)
+            .header("ce-specversion", "1.0")
+            .header("ce-id", "0001")
+            .header("ce-type", "test_event.test_application")
+            .header("ce-source", "http://localhost/")
+            .header("ce-subject", "cloudevents-sdk")
+            .header("content-type", "application/json")
+            .header("ce-string_ex", "val")
+            .header("ce-int_ex", "10")
+            .header("ce-bool_ex", "true")
+            .header("ce-time", &fixtures::time().to_rfc3339())
+            .body(fixtures::json_data_binary());
+        let (req, mut body) = req.split();
+        let result = Event::from_request(&req, &mut body).await.unwrap();
+
+        assert_eq!(expected, result);
+    }
+}

src/binding/poem/mod.rs

@@ -0,0 +1,59 @@
+//! This module integrates the [cloudevents-sdk](https://docs.rs/cloudevents-sdk) with
+//! [Poem](https://docs.rs/poem/) to easily send and receive CloudEvents.
+//!
+//! To deserialize an HTTP request as CloudEvent
+//!
+//! To echo events:
+//!
+//! ```rust
+//! use cloudevents::Event;
+//! use poem_lib as poem;
+//! use poem::{handler, Route, post};
+//!
+//! #[handler]
+//! async fn index(event: Event) -> Event {
+//!     println!("received cloudevent {}", &event);
+//!     event
+//! }
+//!
+//! let app = Route::new().at("/", post(index));
+//! ```
+//!
+//! To create event inside request handlers and send them as responses:
+//!
+//! ```rust
+//! use cloudevents::{Event, EventBuilder, EventBuilderV10};
+//! use poem_lib as poem;
+//! use poem::{handler, Route, post, Result};
+//! use poem::error::InternalServerError;
+//! use serde_json::json;
+//!
+//! #[handler]
+//! async fn index() -> Result<Event> {
+//!     let event = EventBuilderV10::new()
+//!         .id("1")
+//!         .source("url://example_response/")
+//!         .ty("example.ce")
+//!         .data(
+//!             mime::APPLICATION_JSON.to_string(),
+//!             json!({
+//!                 "name": "John Doe",
+//!                 "age": 43,
+//!                 "phones": [
+//!                     "+44 1234567",
+//!                     "+44 2345678"
+//!                 ]
+//!             }),
+//!         )
+//!         .build()
+//!         .map_err(InternalServerError)?;
+//!     Ok(event)
+//! }
+//!
+//! let app = Route::new().at("/", post(index));
+//! ```
+
+mod extractor;
+mod response;
+
+pub use extractor::ParseEventError;