Add Axum binding

Signed-off-by: andrew webber (personal) <[email protected]>

Author: andrewwebber

.github/workflows/rust_tests.yml

@@ -129,3 +129,11 @@ jobs:
         command: build
         toolchain: ${{ matrix.toolchain }}
         args: --target ${{ matrix.target }} --manifest-path ./example-projects/warp-example/Cargo.toml
+
+    - uses: actions-rs/cargo@v1
+      name: "Build axum-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/axum-example/Cargo.toml

Cargo.toml

@@ -21,6 +21,7 @@ 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"]
 
 [dependencies]
 serde = { version = "^1.0", features = ["derive"] }
@@ -42,6 +43,8 @@ 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}
 
 [target."cfg(not(target_arch = \"wasm32\"))".dependencies]
 hostname = "^0.3"
@@ -65,3 +68,4 @@ chrono = { version = "^0.4", features = ["serde"] }
 mockito = "0.25.1"
 tokio = { version = "^1.0", features = ["full"] }
 mime = "0.3"
+tower = { version = "0.4", features = ["util"] }

example-projects/axum-example/Cargo.toml

@@ -0,0 +1,21 @@
+[package]
+name = "axum-example"
+version = "0.3.0"
+authors = ["Andrew Webber <[email protected]>"]
+edition = "2018"
+
+[dependencies]
+cloudevents-sdk = { path = "../..", features = ["axum"] }
+axum = "^0.2"
+http = "^0.2"
+tokio = { version = "^1", features = ["full"] }
+tracing = "^0.1"
+tracing-subscriber = "^0.2"
+tower-http = { version = "^0.1", features = ["trace"] }
+
+[dev-dependencies]
+tower = { version = "^0.4", features = ["util"] }
+serde = { version = "^1.0", features = ["derive"] }
+serde_json = "^1.0"
+chrono = { version = "^0.4", features = ["serde"] }
+hyper = { version = "^0.14" }

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

@@ -0,0 +1,115 @@
+use axum::{
+    handler::{get, post},
+    routing::BoxRoute,
+    Router,
+};
+use cloudevents::Event;
+use http::StatusCode;
+use std::net::SocketAddr;
+use tower_http::trace::TraceLayer;
+
+fn echo_app() -> Router<BoxRoute> {
+    Router::new()
+        .route("/", get(|| async { "hello from cloudevents server" }))
+        .route(
+            "/",
+            post(|event: Event| async move {
+                tracing::debug!("received cloudevent {}", &event);
+                (StatusCode::OK, event)
+            }),
+        )
+        .layer(TraceLayer::new_for_http())
+        .boxed()
+}
+
+#[tokio::main]
+async fn main() {
+    if std::env::var("RUST_LOG").is_err() {
+        std::env::set_var("RUST_LOG", "axum_example=debug,tower_http=debug")
+    }
+    tracing_subscriber::fmt::init();
+    let service = echo_app();
+    let addr = SocketAddr::from(([127, 0, 0, 1], 8080));
+    tracing::debug!("listening on {}", addr);
+    axum::Server::bind(&addr)
+        .serve(service.into_make_service())
+        .await
+        .unwrap();
+}
+
+#[cfg(test)]
+mod tests {
+
+    use super::echo_app;
+
+    use axum::{
+        body::Body,
+        http::{self, Request},
+    };
+    use chrono::Utc;
+    use hyper;
+    use serde_json::json;
+    use tower::ServiceExt; // for `app.oneshot()`
+
+    #[tokio::test]
+    async fn axum_mod_test() {
+        if std::env::var("RUST_LOG").is_err() {
+            std::env::set_var("RUST_LOG", "axum_example=debug,tower_http=debug")
+        }
+        tracing_subscriber::fmt::init();
+
+        let app = echo_app();
+        let time = Utc::now();
+        let j = json!({"hello": "world"});
+        let request = Request::builder()
+            .method(http::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(serde_json::to_vec(&j).unwrap()))
+            .unwrap();
+
+        let resp = app.oneshot(request).await.unwrap();
+        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"
+        );
+
+        let (_, body) = resp.into_parts();
+        let body = hyper::body::to_bytes(body).await.unwrap();
+
+        assert_eq!(j.to_string().as_bytes(), body);
+    }
+}

src/binding/axum/extract.rs

@@ -0,0 +1,143 @@
+use axum_lib as axum;
+
+use async_trait::async_trait;
+use axum::extract::{FromRequest, RequestParts};
+use http::StatusCode;
+use http_body::Body;
+use hyper::body;
+
+use crate::binding::http::to_event;
+use crate::event::Event;
+
+type BoxError = Box<dyn std::error::Error + Send + Sync>;
+
+#[async_trait]
+impl<B> FromRequest<B> for Event
+where
+    B: Body + Send,
+    B::Data: Send,
+    B::Error: Into<BoxError>,
+{
+    type Rejection = (StatusCode, String);
+
+    async fn from_request(req: &mut RequestParts<B>) -> Result<Self, Self::Rejection> {
+        let headers = req.headers().cloned().ok_or(0).map_err(|_| {
+            (
+                StatusCode::BAD_REQUEST,
+                "unexpected empty headers".to_string(),
+            )
+        })?;
+
+        let req_body = req
+            .take_body()
+            .ok_or(0)
+            .map_err(|_| (StatusCode::BAD_REQUEST, "unexpected empty body".to_string()))?;
+
+        let buf = body::to_bytes(req_body)
+            .await
+            .map_err(|e| (StatusCode::BAD_REQUEST, format!("{}", e.into())))?;
+        to_event(&headers, buf.to_vec()).map_err(|e| (StatusCode::BAD_REQUEST, format!("{}", e)))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use axum_lib as axum;
+
+    use super::*;
+    use axum::body::Body;
+    use axum::http::{self, Request, StatusCode};
+    use chrono::Utc;
+    use serde_json::json;
+
+    use crate::{EventBuilder, EventBuilderV10};
+
+    #[tokio::test]
+    async fn axum_test_request() {
+        let time = Utc::now();
+        let expected = EventBuilderV10::new()
+            .id("0001")
+            .ty("example.test")
+            .source("http://localhost/")
+            .time(time)
+            .extension("someint", "10")
+            .build()
+            .unwrap();
+
+        let mut request = RequestParts::new(
+            Request::builder()
+                .method(http::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())
+                .body(Body::empty())
+                .unwrap(),
+        );
+
+        let result = Event::from_request(&mut request).await.unwrap();
+
+        assert_eq!(expected, result);
+    }
+
+    #[tokio::test]
+    async fn axum_test_bad_request() {
+        let time = Utc::now();
+
+        let mut request = RequestParts::new(
+            Request::builder()
+                .method(http::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", time.to_rfc3339())
+                .body(Body::empty())
+                .unwrap(),
+        );
+
+        let result = Event::from_request(&mut request).await;
+        assert!(result.is_err());
+        let rejection = result.unwrap_err();
+
+        let reason = rejection.0;
+        assert_eq!(reason, StatusCode::BAD_REQUEST)
+    }
+
+    #[tokio::test]
+    async fn axum_test_request_with_full_data() {
+        let time = Utc::now();
+        let j = json!({"hello": "world"});
+
+        let expected = EventBuilderV10::new()
+            .id("0001")
+            .ty("example.test")
+            .source("http://localhost")
+            .time(time)
+            .data("application/json", j.to_string().into_bytes())
+            .extension("someint", "10")
+            .build()
+            .unwrap();
+
+        let mut request = RequestParts::new(
+            Request::builder()
+                .method(http::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(serde_json::to_vec(&j).unwrap()))
+                .unwrap(),
+        );
+
+        let result = Event::from_request(&mut request).await.unwrap();
+
+        assert_eq!(expected, result);
+    }
+}