actix/actix-extras
1
0
Fork 0
mirror of https://github.com/actix/actix-extras.git synced 2024-11-23 15:51:06 +01:00
Code Issues Releases Wiki Activity

Merge pull request #8 from JohnTitor/actix-web-2

Browse Source 
Migrate to actix-web 2.0.0 and std::future
This commit is contained in:
Yuki Okushi 2020-01-03 16:59:29 +09:00 committed by GitHub
commit 2225642c6d
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 101 additions and 103 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
.travis.yml
View File

@ -4,8 +4,8 @@ rust:
- beta
- nightly
sudo: required
dist: trusty
os: linux
dist: xenial
env:
global:
@ -26,7 +26,7 @@ addons:
before_script:
- |
if [[ "$TRAVIS_RUST_VERSION" == "nightly" ]]; then
( ( cargo install clippy && export CLIPPY=true ) || export CLIPPY=false );
( ( rustup component add clippy && export CLIPPY=true ) || export CLIPPY=false );
fi
- export PATH=$PATH:~/.cargo/bin

4
CHANGES.md
View File

@ -1,5 +1,9 @@
# Changes
## 0.5.0 (in the future)
* Migrate to actix-web 2.0.0 and std::future
## 0.4.1 (2019-10-03)
* Upgrade prost and prost-derive to 0.5.0

13
Cargo.toml
View File

@ -1,7 +1,8 @@
[package]
name = "actix-protobuf"
version = "0.4.1"
authors = ["kingxsp <jin.hb.zh@outlook.com>"]
edition = "2018"
authors = ["kingxsp <jin.hb.zh@outlook.com>, Yuki Okushi <huyuumi.dev@gmail.com>"]
description = "Protobuf support for actix-web framework."
readme = "README.md"
keywords = ["actix"]
@ -20,16 +21,16 @@ path = "src/lib.rs"
[dependencies]
bytes = "0.4"
futures = "0.1"
derive_more = "0.14"
futures = "0.3.1"
derive_more = "0.99"
actix = "0.8.1"
actix-web = "1.0.0-rc"
actix = "0.9"
actix-rt = "1"
actix-web = "2"
prost = "0.5.0"
[dev-dependencies]
http = "^0.1"
prost-derive = "0.5.0"
[workspace]

10
examples/prost-example/Cargo.toml
View File

@ -1,7 +1,8 @@
[package]
name = "prost-example"
version = "0.3.0"
authors = ["kingxsp <jin.hb.zh@outlook.com>"]
version = "0.4.0"
edition = "2018"
authors = ["kingxsp <jin.hb.zh@outlook.com>, Yuki Okushi <huyuumi.dev@gmail.com>"]
[dependencies]
bytes = "0.4"
@ -10,6 +11,7 @@ env_logger = "*"
prost = "0.5.0"
prost-derive = "0.5.0"
actix = "0.8.1"
actix-web = "1.0.0-rc"
actix = "0.9"
actix-rt = "1"
actix-web = "2"
actix-protobuf = { path="../../" }

24
examples/prost-example/src/main.rs
View File

@ -1,9 +1,3 @@
extern crate actix;
extern crate actix_protobuf;
extern crate actix_web;
extern crate bytes;
extern crate env_logger;
extern crate prost;
#[macro_use]
extern crate prost_derive;
@ -18,25 +12,23 @@ pub struct MyObj {
pub name: String,
}
fn index(msg: ProtoBuf<MyObj>) -> Result<HttpResponse> {
async fn index(msg: ProtoBuf<MyObj>) -> Result<HttpResponse> {
println!("model: {:?}", msg);
HttpResponse::Ok().protobuf(msg.0) // <- send response
}
fn main() {
::std::env::set_var("RUST_LOG", "actix_web=info");
#[actix_rt::main]
async fn main() -> std::io::Result<()> {
std::env::set_var("RUST_LOG", "actix_web=debug,actix_server=info");
env_logger::init();
let sys = actix::System::new("prost-example");
HttpServer::new(|| {
App::new()
.wrap(middleware::Logger::default())
.service(web::resource("/").route(web::post().to(index)))
}).bind("127.0.0.1:8081")
.unwrap()
})
.bind("127.0.0.1:8081")?
.shutdown_timeout(1)
.start();
println!("Started http server: 127.0.0.1:8081");
let _ = sys.run();
.run()
.await
}

147
src/lib.rs
View File

@ -1,20 +1,10 @@
extern crate actix;
extern crate actix_web;
extern crate bytes;
extern crate derive_more;
extern crate futures;
#[cfg(test)]
extern crate http;
extern crate prost;
#[cfg(test)]
#[macro_use]
extern crate prost_derive;
use derive_more::Display;
use std::fmt;
use std::future::Future;
use std::ops::{Deref, DerefMut};
use std::pin::Pin;
use std::task;
use std::task::Poll;
use bytes::{BytesMut, IntoBuf};
use prost::DecodeError as ProtoBufDecodeError;
@ -25,7 +15,8 @@ use actix_web::dev::{HttpResponseBuilder, Payload};
use actix_web::error::{Error, PayloadError, ResponseError};
use actix_web::http::header::{CONTENT_LENGTH, CONTENT_TYPE};
use actix_web::{FromRequest, HttpMessage, HttpRequest, HttpResponse, Responder};
use futures::{Future, Poll, Stream};
use futures::future::{ready, FutureExt, LocalBoxFuture, Ready};
use futures::StreamExt;
#[derive(Debug, Display)]
pub enum ProtoBufPayloadError {
@ -125,7 +116,7 @@ where
{
type Config = ProtoBufConfig;
type Error = Error;
type Future = Box<dyn Future<Item = Self, Error = Error>>;
type Future = LocalBoxFuture<'static, Result<Self, Error>>;
#[inline]
fn from_request(req: &HttpRequest, payload: &mut Payload) -> Self::Future {
@ -133,29 +124,32 @@ where
.app_data::<ProtoBufConfig>()
.map(|c| c.limit)
.unwrap_or(262_144);
Box::new(
ProtoBufMessage::new(req, payload)
.limit(limit)
.map_err(move |e| e.into())
.map(ProtoBuf),
)
ProtoBufMessage::new(req, payload)
.limit(limit)
.map(move |res| match res {
Err(e) => Err(e.into()),
Ok(item) => Ok(ProtoBuf(item)),
})
.boxed_local()
}
}
impl<T: Message + Default> Responder for ProtoBuf<T> {
type Error = Error;
type Future = Result<HttpResponse, Error>;
type Future = Ready<Result<HttpResponse, Error>>;
fn respond_to(self, _: &HttpRequest) -> Self::Future {
let mut buf = Vec::new();
self.0
.encode(&mut buf)
.map_err(|e| Error::from(ProtoBufPayloadError::Serialize(e)))
.and_then(|()| {
Ok(HttpResponse::Ok()
.content_type("application/protobuf")
.body(buf))
})
ready(
self.0
.encode(&mut buf)
.map_err(|e| Error::from(ProtoBufPayloadError::Serialize(e)))
.and_then(|()| {
Ok(HttpResponse::Ok()
.content_type("application/protobuf")
.body(buf))
}),
)
}
}
@ -164,7 +158,7 @@ pub struct ProtoBufMessage<T: Message + Default> {
length: Option<usize>,
stream: Option<Payload>,
err: Option<ProtoBufPayloadError>,
fut: Option<Box<dyn Future<Item = T, Error = ProtoBufPayloadError>>>,
fut: Option<LocalBoxFuture<'static, Result<T, ProtoBufPayloadError>>>,
}
impl<T: Message + Default> ProtoBufMessage<T> {
@ -206,40 +200,50 @@ impl<T: Message + Default> ProtoBufMessage<T> {
}
impl<T: Message + Default + 'static> Future for ProtoBufMessage<T> {
type Item = T;
type Error = ProtoBufPayloadError;
type Output = Result<T, ProtoBufPayloadError>;
fn poll(&mut self) -> Poll<T, ProtoBufPayloadError> {
fn poll(
mut self: Pin<&mut Self>,
task: &mut task::Context<'_>,
) -> Poll<Self::Output> {
if let Some(ref mut fut) = self.fut {
return fut.poll();
return Pin::new(fut).poll(task);
}
if let Some(err) = self.err.take() {
return Err(err);
return Poll::Ready(Err(err));
}
let limit = self.limit;
if let Some(len) = self.length.take() {
if len > limit {
return Err(ProtoBufPayloadError::Overflow);
return Poll::Ready(Err(ProtoBufPayloadError::Overflow));
}
}
let fut = self
let mut stream = self
.stream
.take()
.expect("ProtoBufMessage could not be used second time")
.from_err()
.fold(BytesMut::with_capacity(8192), move |mut body, chunk| {
if (body.len() + chunk.len()) > limit {
Err(ProtoBufPayloadError::Overflow)
} else {
body.extend_from_slice(&chunk);
Ok(body)
.expect("ProtoBufMessage could not be used second time");
self.fut = Some(
async move {
let mut body = BytesMut::with_capacity(8192);
while let Some(item) = stream.next().await {
let chunk = item?;
if (body.len() + chunk.len()) > limit {
return Err(ProtoBufPayloadError::Overflow);
} else {
body.extend_from_slice(&chunk);
}
}
}).and_then(|body| Ok(<T>::decode(&mut body.into_buf())?));
self.fut = Some(Box::new(fut));
self.poll()
return Ok(<T>::decode(&mut body.into_buf())?);
}
.boxed_local(),
);
self.poll(task)
}
}
@ -262,8 +266,8 @@ impl ProtoBufResponseBuilder for HttpResponseBuilder {
#[cfg(test)]
mod tests {
use super::*;
use actix_web::test::{block_on, TestRequest};
use http::header;
use actix_web::http::header;
use actix_web::test::TestRequest;
impl PartialEq for ProtoBufPayloadError {
fn eq(&self, other: &ProtoBufPayloadError) -> bool {
@ -289,44 +293,39 @@ mod tests {
pub name: String,
}
#[test]
fn test_protobuf() {
#[actix_rt::test]
async fn test_protobuf() {
let protobuf = ProtoBuf(MyObject {
number: 9,
name: "test".to_owned(),
});
let req = TestRequest::default().to_http_request();
let resp = protobuf.respond_to(&req).unwrap();
let resp = protobuf.respond_to(&req).await.unwrap();
assert_eq!(
resp.headers().get(header::CONTENT_TYPE).unwrap(),
"application/protobuf"
);
}
#[test]
fn test_protobuf_message() {
#[actix_rt::test]
async fn test_protobuf_message() {
let (req, mut pl) = TestRequest::default().to_http_parts();
let protobuf = block_on(ProtoBufMessage::<MyObject>::new(&req, &mut pl));
let protobuf = ProtoBufMessage::<MyObject>::new(&req, &mut pl).await;
assert_eq!(protobuf.err().unwrap(), ProtoBufPayloadError::ContentType);
let (req, mut pl) = TestRequest::default()
.header(
header::CONTENT_TYPE,
header::HeaderValue::from_static("application/text"),
).to_http_parts();
let protobuf = block_on(ProtoBufMessage::<MyObject>::new(&req, &mut pl));
let (req, mut pl) =
TestRequest::with_header(header::CONTENT_TYPE, "application/text")
.to_http_parts();
let protobuf = ProtoBufMessage::<MyObject>::new(&req, &mut pl).await;
assert_eq!(protobuf.err().unwrap(), ProtoBufPayloadError::ContentType);
let (req, mut pl) = TestRequest::default()
.header(
header::CONTENT_TYPE,
header::HeaderValue::from_static("application/protobuf"),
).header(
header::CONTENT_LENGTH,
header::HeaderValue::from_static("10000"),
).to_http_parts();
let protobuf =
block_on(ProtoBufMessage::<MyObject>::new(&req, &mut pl).limit(100));
let (req, mut pl) =
TestRequest::with_header(header::CONTENT_TYPE, "application/protobuf")
.header(header::CONTENT_LENGTH, "10000")
.to_http_parts();
let protobuf = ProtoBufMessage::<MyObject>::new(&req, &mut pl)
.limit(100)
.await;
assert_eq!(protobuf.err().unwrap(), ProtoBufPayloadError::Overflow);
}
}