Skip to content

Commit

Permalink
Bump actix dep to version 4
Browse files Browse the repository at this point in the history 
Biggest change is Responder is no longer async:
actix/actix-web#1891

Signed-off-by: Jim Crossley <jim@crossleys.org>
  • Loading branch information
@jcrossley3
jcrossley3 committed Jun 30, 2021
1 parent f6b45d1 commit d7aeb21
Show file tree
Hide file tree
Showing 5 changed files with 48 additions and 76 deletions.
2 changes: 1 addition & 1 deletion Cargo.toml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,7 @@ snafu = "^0.6"
bitflags = "^1.2"

# runtime optional deps
actix-web = { version = "^3", default-features = false, optional = true }
actix-web = { version = "4.0.0-beta.8", default-features = false, optional = true }
reqwest = { version = "^0.11", default-features = false, features = ["rustls-tls"], optional = true }
rdkafka = { version = "^0.25", features = ["cmake-build"], optional = true }
warp = { version = "^0.3", optional = true }
Expand Down
12 changes: 2 additions & 10 deletions src/actix/headers.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -5,14 +5,6 @@ use lazy_static::lazy_static;
use std::collections::HashMap;
use std::str::FromStr;

macro_rules! unwrap_optional_header {
($headers:expr, $name:expr) => {
$headers
.get::<&'static HeaderName>(&$name)
.map(|a| header_value_to_str!(a))
};
}

macro_rules! header_value_to_str {
($header_value:expr) => {
$header_value
Expand Down Expand Up @@ -61,8 +53,8 @@ fn attributes_to_headers(
lazy_static! {
pub(crate) static ref ATTRIBUTES_TO_HEADERS: HashMap<&'static str, HeaderName> =
attributes_to_headers(SpecVersion::all_attribute_names());
pub(crate) static ref SPEC_VERSION_HEADER: HeaderName =
HeaderName::from_static("ce-specversion");
pub(crate) static ref CLOUDEVENTS_JSON_HEADER: HeaderValue =
HeaderValue::from_static("application/cloudevents+json");
}

pub(crate) static SPEC_VERSION_HEADER: &str = "ce-specversion";
4 changes: 1 addition & 3 deletions src/actix/mod.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@
//!
//! #[get("/")]
//! async fn get_event() -> Result<HttpResponse, actix_web::Error> {
//! Ok(HttpResponse::Ok()
//! HttpResponse::Ok()
//! .event(
//! EventBuilderV10::new()
//! .id("0001")
Expand All @@ -34,8 +34,6 @@
//! .build()
//! .expect("No error while building the event"),
//! )
//! .await?
//! )
//! }
//! ```
Expand Down
46 changes: 24 additions & 22 deletions src/actix/server_request.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,6 @@ use crate::message::{
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;
Expand All @@ -32,18 +31,22 @@ impl<'a> BinaryDeserializer for HttpRequestDeserializer<'a> {
}

let spec_version = SpecVersion::try_from(
unwrap_optional_header!(self.req.headers(), headers::SPEC_VERSION_HEADER).unwrap()?,
self.req
.headers()
.get(headers::SPEC_VERSION_HEADER)
.unwrap()
.to_str()
.unwrap(),
)?;

visitor = visitor.set_spec_version(spec_version.clone())?;

let attributes = spec_version.attribute_names();

for (hn, hv) in
self.req.headers().iter().filter(|(hn, _)| {
headers::SPEC_VERSION_HEADER.ne(hn) && hn.as_str().starts_with("ce-")
})
{
for (hn, hv) in self.req.headers().iter().filter(|(hn, _)| {
let key = hn.as_str();
headers::SPEC_VERSION_HEADER.ne(key) && key.starts_with("ce-")
}) {
let name = &hn.as_str()["ce-".len()..];

if attributes.contains(&name) {
Expand Down Expand Up @@ -90,8 +93,7 @@ impl<'a> MessageDeserializer for HttpRequestDeserializer<'a> {
} else if self
.req
.headers()
.get::<&'static HeaderName>(&super::headers::SPEC_VERSION_HEADER)
.is_some()
.contains_key(super::headers::SPEC_VERSION_HEADER)
{
Encoding::BINARY
} else {
Expand Down Expand Up @@ -178,12 +180,12 @@ mod tests {
.unwrap();

let (req, payload) = test::TestRequest::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())
.insert_header(("ce-specversion", "1.0"))
.insert_header(("ce-id", "0001"))
.insert_header(("ce-type", "example.test"))
.insert_header(("ce-source", "http://localhost/"))
.insert_header(("ce-someint", "10"))
.insert_header(("ce-time", time.to_rfc3339()))
.to_http_parts();

let resp = req.to_event(web::Payload(payload)).await.unwrap();
Expand All @@ -208,13 +210,13 @@ mod tests {
.unwrap();

let (req, payload) = test::TestRequest::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")
.insert_header(("ce-specversion", "1.0"))
.insert_header(("ce-id", "0001"))
.insert_header(("ce-type", "example.test"))
.insert_header(("ce-source", "http://localhost"))
.insert_header(("ce-someint", "10"))
.insert_header(("ce-time", time.to_rfc3339()))
.insert_header(("content-type", "application/json"))
.set_json(&j)
.to_http_parts();

Expand Down
60 changes: 20 additions & 40 deletions src/actix/server_response.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4,12 +4,9 @@ use crate::message::{
BinaryDeserializer, BinarySerializer, MessageAttributeValue, Result, StructuredSerializer,
};
use crate::Event;
use actix_web::dev::HttpResponseBuilder;
use actix_web::http::{HeaderName, HeaderValue, StatusCode};
use actix_web::HttpResponse;
use async_trait::async_trait;
use futures::future::LocalBoxFuture;
use futures::FutureExt;
use actix_web::HttpResponseBuilder;
use std::str::FromStr;

/// Wrapper for [`HttpResponseBuilder`] that implements [`StructuredSerializer`] and [`BinarySerializer`].
Expand All @@ -25,26 +22,26 @@ impl HttpResponseSerializer {

impl BinarySerializer<HttpResponse> for HttpResponseSerializer {
fn set_spec_version(mut self, spec_version: SpecVersion) -> Result<Self> {
self.builder.set_header(
headers::SPEC_VERSION_HEADER.clone(),
self.builder.insert_header((
headers::SPEC_VERSION_HEADER,
str_to_header_value!(spec_version.as_str())?,
);
));
Ok(self)
}

fn set_attribute(mut self, name: &str, value: MessageAttributeValue) -> Result<Self> {
self.builder.set_header(
self.builder.insert_header((
headers::ATTRIBUTES_TO_HEADERS.get(name).unwrap().clone(),
str_to_header_value!(value.to_string().as_str())?,
);
));
Ok(self)
}

fn set_extension(mut self, name: &str, value: MessageAttributeValue) -> Result<Self> {
self.builder.set_header(
self.builder.insert_header((
attribute_name_to_header!(name)?,
str_to_header_value!(value.to_string().as_str())?,
);
));
Ok(self)
}

Expand All @@ -61,16 +58,16 @@ impl StructuredSerializer<HttpResponse> for HttpResponseSerializer {
fn set_structured_event(mut self, bytes: Vec<u8>) -> Result<HttpResponse> {
Ok(self
.builder
.set_header(
.insert_header((
actix_web::http::header::CONTENT_TYPE,
headers::CLOUDEVENTS_JSON_HEADER.clone(),
)
))
.body(bytes))
}
}

/// Method to fill an [`HttpResponseBuilder`] with an [`Event`].
pub async fn event_to_response(
pub fn event_to_response(
event: Event,
response: HttpResponseBuilder,
) -> std::result::Result<HttpResponse, actix_web::error::Error> {
Expand All @@ -80,50 +77,37 @@ pub async fn event_to_response(

/// 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()
fn respond_to(self, _: &actix_web::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::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::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)]
mod tests {
use super::*;

use crate::{EventBuilder, EventBuilderV10};
use actix_web::http::StatusCode;
use actix_web::test;
use futures::TryStreamExt;
use actix_web::{http::StatusCode, test};
use serde_json::json;

#[actix_rt::test]
Expand All @@ -138,7 +122,6 @@ mod tests {

let resp = HttpResponseBuilder::new(StatusCode::OK)
.event(input)
.await
.unwrap();

assert_eq!(
Expand Down Expand Up @@ -180,9 +163,8 @@ mod tests {
.build()
.unwrap();

let mut resp = HttpResponseBuilder::new(StatusCode::OK)
let resp = HttpResponseBuilder::new(StatusCode::OK)
.event(input)
.await
.unwrap();

assert_eq!(
Expand Down Expand Up @@ -218,9 +200,7 @@ mod tests {
"10"
);

let bytes = test::load_stream(resp.take_body().into_stream())
.await
.unwrap();
let bytes = test::load_body(resp.into_body()).await.unwrap();
assert_eq!(j.to_string().as_bytes(), bytes.as_ref())
}
}

0 comments on commit d7aeb21

Please sign in to comment.