cloudevents
/
sdk-rust
mirror of https://github.com/cloudevents/sdk-rust.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Update to actix-web 4 (#167) 

* Compatiblity with actix-web 4

Fixes #114

Main challenge here was no longer being able to to construct
web::Payload instances from dev::Payload instances. Instead, we now
invoke the web::Payload FromRequest impl directly.

Also adapted to the latest upstream test and body redesign.

Macros are featurized now so enabled default features for the doc
tests to pass.

Signed-off-by: Jim Crossley <jim@crossleys.org>
This commit is contained in:
Jim Crossley 2022-02-28 13:02:00 -05:00 committed by GitHub
parent 6af7d1e8ec
commit 5a9f64868d
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 97 additions and 73 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
Cargo.toml
View File

@ -18,7 +18,7 @@ name = "cloudevents"
[features]
http-binding = ["async-trait", "bytes", "futures", "http"]
actix = ["actix-web", "async-trait", "bytes", "futures", "http"]
actix = ["actix-web", "actix-http", "async-trait", "bytes", "futures", "http"]
reqwest = ["reqwest-lib", "async-trait", "bytes", "http"]
rdkafka = ["rdkafka-lib", "bytes", "futures"]
warp = ["warp-lib", "bytes", "http", "hyper"]
@ -36,7 +36,8 @@ snafu = "^0.6"
bitflags = "^1.2"
# runtime optional deps
actix-web = { version = "^3", default-features = false, optional = true }
actix-web = { version = "4", optional = true }
actix-http = { version = "3", optional = true }
reqwest-lib = { version = "^0.11", default-features = false, features = ["rustls-tls"], optional = true, package = "reqwest" }
rdkafka-lib = { version = "^0.28", features = ["cmake-build"], optional = true, package = "rdkafka" }
warp-lib = { version = "^0.3", optional = true, package = "warp" }
@ -67,7 +68,7 @@ version-sync = "0.9.2"
serde_yaml = "0.8"
# runtime dev-deps
actix-rt = { version = "^1" }
actix-rt = { version = "^2" }
url = { version = "^2.1", features = ["serde"] }
serde_json = { version = "^1.0" }
chrono = { version = "^0.4", features = ["serde"] }

4
example-projects/actix-web-example/Cargo.toml
View File

@ -6,8 +6,8 @@ edition = "2018"
[dependencies]
cloudevents-sdk = { path = "../..", features = ["actix"] }
actix-web = "^3"
actix-cors = "^0.5"
actix-web = "4"
actix-cors = "0.6.0-beta.8"
serde_json = "^1.0"
url = { version = "^2.1" }
env_logger = "0.7.1"

2
example-projects/actix-web-example/src/main.rs
View File

@ -29,8 +29,8 @@ async fn main() -> std::io::Result<()> {
HttpServer::new(|| {
App::new()
.wrap(actix_web::middleware::Logger::default())
.wrap(actix_cors::Cors::permissive())
.wrap(actix_web::middleware::Logger::default())
.service(post_event)
.service(get_event)
})

41
src/binding/actix/mod.rs
View File

@ -3,6 +3,19 @@
//! To deserialize an HTTP request as CloudEvent:
//!
//! ```
//! use cloudevents::Event;
//! use actix_web::post;
//!
//! #[post("/")]
//! async fn post_event(event: Event) -> Result<String, actix_web::Error> {
//! println!("Received Event: {:?}", event);
//! Ok(format!("{:?}", event))
//! }
//! ```
//!
//! For more complex applications, access the Payload directly:
//!
//! ```
//! use cloudevents::binding::actix::HttpRequestExt;
//! use actix_web::{HttpRequest, web, post};
//!
@ -17,14 +30,36 @@
//! To serialize a CloudEvent to an HTTP response:
//!
//! ```
//! use actix_web::get;
//! use cloudevents::{Event, EventBuilderV10, EventBuilder};
//! use serde_json::json;
//!
//! #[get("/")]
//! async fn get_event() -> Event {
//! let payload = json!({"hello": "world"});
//!
//! EventBuilderV10::new()
//! .id("0001")
//! .ty("example.test")
//! .source("http://localhost/")
//! .data("application/json", payload)
//! .extension("someint", "10")
//! .build()
//! .unwrap()
//! }
//! ```
//!
//! For more complex applications, use the HTTP response builder extension:
//!
//! ```
//! use cloudevents::binding::actix::HttpResponseBuilderExt;
//! use actix_web::{HttpRequest, web, get, HttpResponse};
//! use actix_web::{get, HttpResponse};
//! use cloudevents::{EventBuilderV10, EventBuilder};
//! use serde_json::json;
//!
//! #[get("/")]
//! async fn get_event() -> Result<HttpResponse, actix_web::Error> {
//! Ok(HttpResponse::Ok()
//! HttpResponse::Ok()
//! .event(
//! EventBuilderV10::new()
//! .id("0001")
@ -34,8 +69,6 @@
//! .build()
//! .expect("No error while building the event"),
//! )
//! .await?
//! )
//! }
//! ```

69
src/binding/actix/server_request.rs
View File

@ -1,14 +1,14 @@
use crate::binding::http::{to_event, Headers};
use crate::Event;
use actix_web::dev::Payload;
use actix_web::web::BytesMut;
use actix_web::{web, HttpRequest};
use async_trait::async_trait;
use futures::future::LocalBoxFuture;
use futures::{FutureExt, StreamExt};
use futures::{future::LocalBoxFuture, FutureExt, StreamExt};
use http::header::{AsHeaderName, HeaderName, HeaderValue};
/// Implement Headers for the actix HeaderMap
impl<'a> Headers<'a> for actix_web::http::HeaderMap {
impl<'a> Headers<'a> for actix_http::header::HeaderMap {
type Iterator = Box<dyn Iterator<Item = (&'a HeaderName, &'a HeaderValue)> + 'a>;
fn get<K: AsHeaderName>(&self, key: K) -> Option<&HeaderValue> {
self.get(key.as_str())
@ -32,14 +32,18 @@ pub async fn request_to_event(
/// So that an actix-web handler may take an Event parameter
impl actix_web::FromRequest for Event {
type Config = ();
type Error = actix_web::Error;
type Future = LocalBoxFuture<'static, std::result::Result<Self, Self::Error>>;
fn from_request(r: &HttpRequest, p: &mut actix_web::dev::Payload) -> Self::Future {
let payload = web::Payload(p.take());
fn from_request(r: &HttpRequest, p: &mut Payload) -> Self::Future {
let request = r.to_owned();
async move { request_to_event(&request, payload).await }.boxed_local()
bytes::Bytes::from_request(&request, p)
.map(move |bytes| match bytes {
Ok(b) => to_event(request.headers(), b.to_vec())
.map_err(actix_web::error::ErrorBadRequest),
Err(e) => Err(e),
})
.boxed_local()
}
}
@ -74,24 +78,31 @@ mod private {
#[cfg(test)]
mod tests {
use super::*;
use actix_web::test;
use actix_web::{test, FromRequest};
use crate::test::fixtures;
use serde_json::json;
async fn to_event(req: &HttpRequest, mut payload: Payload) -> Event {
web::Payload::from_request(&req, &mut payload)
.then(|p| req.to_event(p.unwrap()))
.await
.unwrap()
}
#[actix_rt::test]
async fn test_request() {
let expected = fixtures::v10::minimal_string_extension();
let (req, payload) = test::TestRequest::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")
.insert_header(("ce-specversion", "1.0"))
.insert_header(("ce-id", "0001"))
.insert_header(("ce-type", "test_event.test_application"))
.insert_header(("ce-source", "http://localhost/"))
.insert_header(("ce-someint", "10"))
.to_http_parts();
let resp = req.to_event(web::Payload(payload)).await.unwrap();
assert_eq!(expected, resp);
assert_eq!(expected, to_event(&req, payload).await);
}
#[actix_rt::test]
@ -99,21 +110,20 @@ mod tests {
let expected = fixtures::v10::full_binary_json_data_string_extension();
let (req, payload) = test::TestRequest::post()
.header("ce-specversion", "1.0")
.header("ce-id", "0001")
.header("ce-type", "test_event.test_application")
.header("ce-subject", "cloudevents-sdk")
.header("ce-source", "http://localhost/")
.header("ce-time", fixtures::time().to_rfc3339())
.header("ce-string_ex", "val")
.header("ce-int_ex", "10")
.header("ce-bool_ex", "true")
.header("content-type", "application/json")
.insert_header(("ce-specversion", "1.0"))
.insert_header(("ce-id", "0001"))
.insert_header(("ce-type", "test_event.test_application"))
.insert_header(("ce-subject", "cloudevents-sdk"))
.insert_header(("ce-source", "http://localhost/"))
.insert_header(("ce-time", fixtures::time().to_rfc3339()))
.insert_header(("ce-string_ex", "val"))
.insert_header(("ce-int_ex", "10"))
.insert_header(("ce-bool_ex", "true"))
.insert_header(("content-type", "application/json"))
.set_json(&fixtures::json_data())
.to_http_parts();
let resp = req.to_event(web::Payload(payload)).await.unwrap();
assert_eq!(expected, resp);
assert_eq!(expected, to_event(&req, payload).await);
}
#[actix_rt::test]
@ -136,11 +146,10 @@ mod tests {
let expected = fixtures::v10::full_json_data_string_extension();
let (req, payload) = test::TestRequest::post()
.header("content-type", "application/cloudevents+json")
.insert_header(("content-type", "application/cloudevents+json"))
.set_payload(bytes)
.to_http_parts();
let resp = req.to_event(web::Payload(payload)).await.unwrap();
assert_eq!(expected, resp);
assert_eq!(expected, to_event(&req, payload).await);
}
}

45
src/binding/actix/server_response.rs
View File

@ -1,16 +1,12 @@
use crate::binding::http::{Builder, Serializer};
use crate::message::{BinaryDeserializer, Result};
use crate::Event;
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;
use actix_web::{HttpRequest, HttpResponse, HttpResponseBuilder};
impl Builder<HttpResponse> for HttpResponseBuilder {
fn header(&mut self, key: &str, value: http::header::HeaderValue) {
self.set_header(key, value);
self.insert_header((key, value));
}
fn body(&mut self, bytes: Vec<u8>) -> Result<HttpResponse> {
Ok(HttpResponseBuilder::body(self, bytes))
@ -21,7 +17,7 @@ impl Builder<HttpResponse> for HttpResponseBuilder {
}
/// Method to fill an [`HttpResponseBuilder`] with an [`Event`].
pub async fn event_to_response<T: Builder<HttpResponse> + 'static>(
pub fn event_to_response<T: Builder<HttpResponse> + 'static>(
event: Event,
response: T,
) -> std::result::Result<HttpResponse, actix_web::error::Error> {
@ -31,40 +27,30 @@ pub async fn event_to_response<T: Builder<HttpResponse> + 'static>(
/// So that an actix-web handler may return an Event
impl actix_web::Responder for Event {
type Error = actix_web::Error;
type Future = LocalBoxFuture<'static, std::result::Result<HttpResponse, Self::Error>>;
fn respond_to(self, _: &actix_web::HttpRequest) -> Self::Future {
async { HttpResponse::build(StatusCode::OK).event(self).await }.boxed_local()
type Body = actix_web::body::BoxBody;
fn respond_to(self, _: &HttpRequest) -> HttpResponse {
HttpResponse::build(StatusCode::OK).event(self).unwrap()
}
}
/// Extension Trait for [`HttpResponseBuilder`] which acts as a wrapper for the function [`event_to_response()`].
///
/// This trait is sealed and cannot be implemented for types outside of this crate.
#[async_trait(?Send)]
pub trait HttpResponseBuilderExt: private::Sealed {
/// Fill this [`HttpResponseBuilder`] with an [`Event`].
async fn event(
self,
event: Event,
) -> std::result::Result<HttpResponse, actix_web::error::Error>;
fn event(self, event: Event) -> std::result::Result<HttpResponse, actix_web::Error>;
}
#[async_trait(?Send)]
impl HttpResponseBuilderExt for HttpResponseBuilder {
async fn event(
self,
event: Event,
) -> std::result::Result<HttpResponse, actix_web::error::Error> {
event_to_response(event, self).await
fn event(self, event: Event) -> std::result::Result<HttpResponse, actix_web::Error> {
event_to_response(event, self)
}
}
// Sealing the HttpResponseBuilderExt
mod private {
pub trait Sealed {}
impl Sealed for actix_web::dev::HttpResponseBuilder {}
impl Sealed for actix_web::HttpResponseBuilder {}
}
#[cfg(test)]
@ -74,7 +60,6 @@ mod tests {
use crate::test::fixtures;
use actix_web::http::StatusCode;
use actix_web::test;
use futures::TryStreamExt;
#[actix_rt::test]
async fn test_response() {
@ -82,7 +67,6 @@ mod tests {
let resp = HttpResponseBuilder::new(StatusCode::OK)
.event(input)
.await
.unwrap();
assert_eq!(
@ -115,9 +99,8 @@ mod tests {
async fn test_response_with_full_data() {
let input = fixtures::v10::full_binary_json_data_string_extension();
let mut resp = HttpResponseBuilder::new(StatusCode::OK)
let resp = HttpResponseBuilder::new(StatusCode::OK)
.event(input)
.await
.unwrap();
assert_eq!(
@ -153,9 +136,7 @@ mod tests {
"10"
);
let bytes = test::load_stream(resp.take_body().into_stream())
.await
.unwrap();
assert_eq!(fixtures::json_data_binary(), bytes.as_ref())
let sr = test::TestRequest::default().to_srv_response(resp);
assert_eq!(fixtures::json_data_binary(), test::read_body(sr).await);
}
}

2
src/lib.rs
View File

@ -39,7 +39,7 @@
//! - `actix`: Enables the [`binding::actix`] protocol binding module. This
//! extends the [`actix_web::HttpRequest`] with a
//! [`to_event`](binding::actix::HttpRequestExt::to_event) function, the
//! [`actix_web::dev::HttpResponseBuilder`] with an
//! [`actix_web::HttpResponseBuilder`] with an
//! [`event`](binding::actix::HttpResponseBuilderExt::event) function,
//! and implementations for [`actix_web::FromRequest`] and
//! [`actix_web::Responder`] in order to take advantage of actix-web's