Merge branch 'proto'

actix-protobuf/.gitignore

@@ -0,0 +1,14 @@
+Cargo.lock
+target/
+guide/build/
+/gh-pages
+
+*.so
+*.out
+*.pyc
+*.pid
+*.sock
+*~
+
+# These are backup files generated by rustfmt
+**/*.rs.bk

actix-protobuf/.travis.yml

@@ -0,0 +1,53 @@
+language: rust
+rust:
+  - stable
+  - beta
+  - nightly
+
+os: linux
+dist: xenial
+
+env:
+  global:
+    - RUSTFLAGS="-C link-dead-code"
+
+addons:
+  apt:
+    packages:
+      - libcurl4-openssl-dev
+      - libelf-dev
+      - libdw-dev
+      - cmake
+      - gcc
+      - binutils-dev
+      - libiberty-dev
+
+# Add clippy
+before_script:
+  - |
+    if [[ "$TRAVIS_RUST_VERSION" == "nightly" ]]; then
+    ( ( rustup component add clippy && export CLIPPY=true ) || export CLIPPY=false );
+    fi
+  - export PATH=$PATH:~/.cargo/bin
+
+script:
+  - |
+    if [[ "$TRAVIS_RUST_VERSION" == "nightly" ]]; then
+        USE_SKEPTIC=1 cargo test
+    else
+        cargo test && cargo check --examples
+    fi
+  - |
+    if [[ "$TRAVIS_RUST_VERSION" == "nightly" && $CLIPPY ]]; then
+        cargo clippy
+    fi
+
+# Upload docs
+after_success:
+  - |
+    if [[ "$TRAVIS_OS_NAME" == "linux" && "$TRAVIS_RUST_VERSION" == "nightly" ]]; then
+      bash <(curl https://raw.githubusercontent.com/xd009642/tarpaulin/master/travis-install.sh)
+      USE_SKEPTIC=1 cargo tarpaulin --out Xml
+      bash <(curl -s https://codecov.io/bash)
+      echo "Uploaded code coverage"
+    fi

actix-protobuf/CHANGES.md

@@ -0,0 +1,28 @@
+# Changes
+
+## 0.5.0 (2019-01-24)
+
+* Migrate to actix-web 2.0.0 and std::future
+* Update prost to 0.6
+* Update bytes to 0.5
+
+## 0.4.1 (2019-10-03)
+
+* Upgrade prost and prost-derive to 0.5.0
+
+## 0.4.0 (2019-05-18)
+
+* Upgrade to actix-web 1.0.0-rc
+* Removed `protobuf` method for `HttpRequest` (use `ProtoBuf` extractor instead)
+
+## 0.3.0 (2019-03-07)
+
+* Upgrade to actix-web 0.7.18
+
+## 0.2.0 (2018-04-10)
+
+* Provide protobuf extractor
+
+## 0.1.0 (2018-03-21)
+
+* First release

actix-protobuf/Cargo.toml

@@ -0,0 +1,40 @@
+[package]
+name = "actix-protobuf"
+version = "0.5.0"
+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"]
+homepage = "https://github.com/actix/actix-protobuf"
+license = "MIT/Apache-2.0"
+exclude = [".gitignore", ".travis.yml", ".cargo/config",
+           "appveyor.yml", "/examples/**"]
+
+[badges]
+travis-ci = { repository = "actix/actix-protobuf", branch = "master" }
+codecov = { repository = "actix/actix-protobuf", branch = "master", service = "github" }
+
+[lib]
+name = "actix_protobuf"
+path = "src/lib.rs"
+
+[dependencies]
+bytes = "0.5"
+futures = "0.3.1"
+derive_more = "0.99"
+
+actix = "0.9"
+actix-rt = "1"
+actix-web = "2"
+
+prost = "0.6.0"
+
+[dev-dependencies]
+prost-derive = "0.6.0"
+
+[workspace]
+members = [
+  "./",
+  "examples/prost-example",
+]

actix-protobuf/README.md

@@ -0,0 +1,35 @@
+# Actix-web ProtoBuf [![Build Status](https://travis-ci.org/actix/actix-protobuf.svg?branch=master)](https://travis-ci.org/actix/actix-protobuf) [![codecov](https://codecov.io/gh/actix/actix-protobuf/branch/master/graph/badge.svg)](https://codecov.io/gh/actix/actix-protobuf) [![crates.io](http://meritbadge.herokuapp.com/actix-protobuf)](https://crates.io/crates/actix-protobuf) [![Join the chat at https://gitter.im/actix/actix](https://badges.gitter.im/actix/actix.svg)](https://gitter.im/actix/actix?utm_source=badge&utm_medium=badge&utm_campaign=pr-badge&utm_content=badge)
+
+Protobuf support for actix-web framework.
+
+
+## Example
+
+```rust,ignore
+use actix_protobuf::*;
+use actix_web::*;
+
+#[derive(Clone, PartialEq, Message)]
+pub struct MyObj {
+    #[prost(int32, tag = "1")]
+    pub number: i32,
+    #[prost(string, tag = "2")]
+    pub name: String,
+}
+
+async fn index(msg: ProtoBuf<MyObj>) -> Result<HttpResponse> {
+    println!("model: {:?}", msg);
+    HttpResponse::Ok().protobuf(msg.0) // <- send response
+}
+```
+
+See [here](https://github.com/actix/actix-protobuf/tree/master/examples/prost-example) for the complete example.
+
+## License
+
+This project is licensed under either of
+
+* Apache License, Version 2.0, ([LICENSE-APACHE](LICENSE-APACHE) or [http://www.apache.org/licenses/LICENSE-2.0](http://www.apache.org/licenses/LICENSE-2.0))
+* MIT license ([LICENSE-MIT](LICENSE-MIT) or [http://opensource.org/licenses/MIT](http://opensource.org/licenses/MIT))
+
+at your option.

actix-protobuf/examples/prost-example/Cargo.toml

@@ -0,0 +1,17 @@
+[package]
+name = "prost-example"
+version = "0.5.0"
+edition = "2018"
+authors = ["kingxsp <jin.hb.zh@outlook.com>, Yuki Okushi <huyuumi.dev@gmail.com>"]
+
+[dependencies]
+bytes = "0.5"
+env_logger = "*"
+
+prost = "0.6.0"
+prost-derive = "0.6.0"
+
+actix = "0.9"
+actix-rt = "1"
+actix-web = "2"
+actix-protobuf = { path="../../" }

actix-protobuf/examples/prost-example/client.py

@@ -0,0 +1,68 @@
+#!/usr/bin/env python3
+# just start server and run client.py
+
+# wget https://github.com/protocolbuffers/protobuf/releases/download/v3.11.2/protobuf-python-3.11.2.zip
+# unzip protobuf-python-3.11.2.zip.1
+# cd protobuf-3.11.2/python/
+# python3 setup.py install
+
+# pip3 install --upgrade pip
+# pip3 install aiohttp
+
+# python3 client.py
+
+import test_pb2
+import traceback
+import sys
+
+import asyncio
+import aiohttp
+
+def op():
+    try:
+        obj = test_pb2.MyObj()
+        obj.number = 9
+        obj.name = 'USB'
+
+        #Serialize
+        sendDataStr = obj.SerializeToString()
+        #print serialized string value
+        print('serialized string:', sendDataStr)
+        #------------------------#
+        #  message transmission  #
+        #------------------------#
+        receiveDataStr = sendDataStr
+        receiveData = test_pb2.MyObj()
+
+        #Deserialize
+        receiveData.ParseFromString(receiveDataStr)
+        print('pares serialize string, return: devId = ', receiveData.number, ', name = ', receiveData.name)
+    except(Exception, e):
+        print(Exception, ':', e)
+        print(traceback.print_exc())
+        errInfo = sys.exc_info()
+        print(errInfo[0], ':', errInfo[1])
+
+
+async def fetch(session):
+    obj = test_pb2.MyObj()
+    obj.number = 9
+    obj.name = 'USB'
+    async with session.post('http://127.0.0.1:8081/', data=obj.SerializeToString(),
+        headers={"content-type": "application/protobuf"}) as resp:
+        print(resp.status)
+        data = await resp.read()
+        receiveObj = test_pb2.MyObj()
+        receiveObj.ParseFromString(data)
+        print(receiveObj)
+
+async def go(loop):
+    obj = test_pb2.MyObj()
+    obj.number = 9
+    obj.name = 'USB'
+    async with aiohttp.ClientSession(loop=loop) as session:
+        await fetch(session)
+
+loop = asyncio.get_event_loop()
+loop.run_until_complete(go(loop))
+loop.close()

actix-protobuf/examples/prost-example/src/main.rs

@@ -0,0 +1,34 @@
+#[macro_use]
+extern crate prost_derive;
+
+use actix_protobuf::*;
+use actix_web::*;
+
+#[derive(Clone, PartialEq, Message)]
+pub struct MyObj {
+    #[prost(int32, tag = "1")]
+    pub number: i32,
+    #[prost(string, tag = "2")]
+    pub name: String,
+}
+
+async fn index(msg: ProtoBuf<MyObj>) -> Result<HttpResponse> {
+    println!("model: {:?}", msg);
+    HttpResponse::Ok().protobuf(msg.0) // <- send response
+}
+
+#[actix_rt::main]
+async fn main() -> std::io::Result<()> {
+    std::env::set_var("RUST_LOG", "actix_web=debug,actix_server=info");
+    env_logger::init();
+
+    HttpServer::new(|| {
+        App::new()
+            .wrap(middleware::Logger::default())
+            .service(web::resource("/").route(web::post().to(index)))
+    })
+    .bind("127.0.0.1:8081")?
+    .shutdown_timeout(1)
+    .run()
+    .await
+}

actix-protobuf/examples/prost-example/test.proto

@@ -0,0 +1,6 @@
+syntax = "proto3";
+
+message MyObj {
+    int32 number = 1;
+    string name = 2;
+}

actix-protobuf/examples/prost-example/test_pb2.py

@@ -0,0 +1,75 @@
+# -*- coding: utf-8 -*-
+# Generated by the protocol buffer compiler.  DO NOT EDIT!
+# source: test.proto
+
+from google.protobuf import descriptor as _descriptor
+from google.protobuf import message as _message
+from google.protobuf import reflection as _reflection
+from google.protobuf import symbol_database as _symbol_database
+# @@protoc_insertion_point(imports)
+
+_sym_db = _symbol_database.Default()
+
+
+
+
+DESCRIPTOR = _descriptor.FileDescriptor(
+  name='test.proto',
+  package='',
+  syntax='proto3',
+  serialized_options=None,
+  serialized_pb=b'\n\ntest.proto\"%\n\x05MyObj\x12\x0e\n\x06number\x18\x01 \x01(\x05\x12\x0c\n\x04name\x18\x02 \x01(\tb\x06proto3'
+)
+
+
+
+
+_MYOBJ = _descriptor.Descriptor(
+  name='MyObj',
+  full_name='MyObj',
+  filename=None,
+  file=DESCRIPTOR,
+  containing_type=None,
+  fields=[
+    _descriptor.FieldDescriptor(
+      name='number', full_name='MyObj.number', index=0,
+      number=1, type=5, cpp_type=1, label=1,
+      has_default_value=False, default_value=0,
+      message_type=None, enum_type=None, containing_type=None,
+      is_extension=False, extension_scope=None,
+      serialized_options=None, file=DESCRIPTOR),
+    _descriptor.FieldDescriptor(
+      name='name', full_name='MyObj.name', index=1,
+      number=2, type=9, cpp_type=9, label=1,
+      has_default_value=False, default_value=b"".decode('utf-8'),
+      message_type=None, enum_type=None, containing_type=None,
+      is_extension=False, extension_scope=None,
+      serialized_options=None, file=DESCRIPTOR),
+  ],
+  extensions=[
+  ],
+  nested_types=[],
+  enum_types=[
+  ],
+  serialized_options=None,
+  is_extendable=False,
+  syntax='proto3',
+  extension_ranges=[],
+  oneofs=[
+  ],
+  serialized_start=14,
+  serialized_end=51,
+)
+
+DESCRIPTOR.message_types_by_name['MyObj'] = _MYOBJ
+_sym_db.RegisterFileDescriptor(DESCRIPTOR)
+
+MyObj = _reflection.GeneratedProtocolMessageType('MyObj', (_message.Message,), {
+  'DESCRIPTOR' : _MYOBJ,
+  '__module__' : 'test_pb2'
+  # @@protoc_insertion_point(class_scope:MyObj)
+  })
+_sym_db.RegisterMessage(MyObj)
+
+
+# @@protoc_insertion_point(module_scope)

actix-protobuf/rustfmt.toml

@@ -0,0 +1,2 @@
+max_width = 89
+reorder_imports = true

actix-protobuf/src/lib.rs

@@ -0,0 +1,331 @@
+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;
+use prost::DecodeError as ProtoBufDecodeError;
+use prost::EncodeError as ProtoBufEncodeError;
+use prost::Message;
+
+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::{ready, FutureExt, LocalBoxFuture, Ready};
+use futures::StreamExt;
+
+#[derive(Debug, Display)]
+pub enum ProtoBufPayloadError {
+    /// Payload size is bigger than 256k
+    #[display(fmt = "Payload size is bigger than 256k")]
+    Overflow,
+    /// Content type error
+    #[display(fmt = "Content type error")]
+    ContentType,
+    /// Serialize error
+    #[display(fmt = "ProtoBuf serialize error: {}", _0)]
+    Serialize(ProtoBufEncodeError),
+    /// Deserialize error
+    #[display(fmt = "ProtoBuf deserialize error: {}", _0)]
+    Deserialize(ProtoBufDecodeError),
+    /// Payload error
+    #[display(fmt = "Error that occur during reading payload: {}", _0)]
+    Payload(PayloadError),
+}
+
+impl ResponseError for ProtoBufPayloadError {
+    fn error_response(&self) -> HttpResponse {
+        match *self {
+            ProtoBufPayloadError::Overflow => HttpResponse::PayloadTooLarge().into(),
+            _ => HttpResponse::BadRequest().into(),
+        }
+    }
+}
+
+impl From<PayloadError> for ProtoBufPayloadError {
+    fn from(err: PayloadError) -> ProtoBufPayloadError {
+        ProtoBufPayloadError::Payload(err)
+    }
+}
+
+impl From<ProtoBufDecodeError> for ProtoBufPayloadError {
+    fn from(err: ProtoBufDecodeError) -> ProtoBufPayloadError {
+        ProtoBufPayloadError::Deserialize(err)
+    }
+}
+
+pub struct ProtoBuf<T: Message>(pub T);
+
+impl<T: Message> Deref for ProtoBuf<T> {
+    type Target = T;
+
+    fn deref(&self) -> &T {
+        &self.0
+    }
+}
+
+impl<T: Message> DerefMut for ProtoBuf<T> {
+    fn deref_mut(&mut self) -> &mut T {
+        &mut self.0
+    }
+}
+
+impl<T: Message> fmt::Debug for ProtoBuf<T>
+where
+    T: fmt::Debug,
+{
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        write!(f, "ProtoBuf: {:?}", self.0)
+    }
+}
+
+impl<T: Message> fmt::Display for ProtoBuf<T>
+where
+    T: fmt::Display,
+{
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        fmt::Display::fmt(&self.0, f)
+    }
+}
+
+pub struct ProtoBufConfig {
+    limit: usize,
+}
+
+impl ProtoBufConfig {
+    /// Change max size of payload. By default max size is 256Kb
+    pub fn limit(&mut self, limit: usize) -> &mut Self {
+        self.limit = limit;
+        self
+    }
+}
+
+impl Default for ProtoBufConfig {
+    fn default() -> Self {
+        ProtoBufConfig { limit: 262_144 }
+    }
+}
+
+impl<T> FromRequest for ProtoBuf<T>
+where
+    T: Message + Default + 'static,
+{
+    type Config = ProtoBufConfig;
+    type Error = Error;
+    type Future = LocalBoxFuture<'static, Result<Self, Error>>;
+
+    #[inline]
+    fn from_request(req: &HttpRequest, payload: &mut Payload) -> Self::Future {
+        let limit = req
+            .app_data::<ProtoBufConfig>()
+            .map(|c| c.limit)
+            .unwrap_or(262_144);
+        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 = Ready<Result<HttpResponse, Error>>;
+
+    fn respond_to(self, _: &HttpRequest) -> Self::Future {
+        let mut buf = Vec::new();
+        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))
+                }),
+        )
+    }
+}
+
+pub struct ProtoBufMessage<T: Message + Default> {
+    limit: usize,
+    length: Option<usize>,
+    stream: Option<Payload>,
+    err: Option<ProtoBufPayloadError>,
+    fut: Option<LocalBoxFuture<'static, Result<T, ProtoBufPayloadError>>>,
+}
+
+impl<T: Message + Default> ProtoBufMessage<T> {
+    /// Create `ProtoBufMessage` for request.
+    pub fn new(req: &HttpRequest, payload: &mut Payload) -> Self {
+        if req.content_type() != "application/protobuf" {
+            return ProtoBufMessage {
+                limit: 262_144,
+                length: None,
+                stream: None,
+                fut: None,
+                err: Some(ProtoBufPayloadError::ContentType),
+            };
+        }
+
+        let mut len = None;
+        if let Some(l) = req.headers().get(CONTENT_LENGTH) {
+            if let Ok(s) = l.to_str() {
+                if let Ok(l) = s.parse::<usize>() {
+                    len = Some(l)
+                }
+            }
+        }
+
+        ProtoBufMessage {
+            limit: 262_144,
+            length: len,
+            stream: Some(payload.take()),
+            fut: None,
+            err: None,
+        }
+    }
+
+    /// Change max size of payload. By default max size is 256Kb
+    pub fn limit(mut self, limit: usize) -> Self {
+        self.limit = limit;
+        self
+    }
+}
+
+impl<T: Message + Default + 'static> Future for ProtoBufMessage<T> {
+    type Output = Result<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 Pin::new(fut).poll(task);
+        }
+
+        if let Some(err) = self.err.take() {
+            return Poll::Ready(Err(err));
+        }
+
+        let limit = self.limit;
+        if let Some(len) = self.length.take() {
+            if len > limit {
+                return Poll::Ready(Err(ProtoBufPayloadError::Overflow));
+            }
+        }
+
+        let mut stream = self
+            .stream
+            .take()
+            .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);
+                    }
+                }
+
+                return Ok(<T>::decode(&mut body)?);
+            }
+            .boxed_local(),
+        );
+        self.poll(task)
+    }
+}
+
+pub trait ProtoBufResponseBuilder {
+    fn protobuf<T: Message>(&mut self, value: T) -> Result<HttpResponse, Error>;
+}
+
+impl ProtoBufResponseBuilder for HttpResponseBuilder {
+    fn protobuf<T: Message>(&mut self, value: T) -> Result<HttpResponse, Error> {
+        self.header(CONTENT_TYPE, "application/protobuf");
+
+        let mut body = Vec::new();
+        value
+            .encode(&mut body)
+            .map_err(ProtoBufPayloadError::Serialize)?;
+        Ok(self.body(body))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use actix_web::http::header;
+    use actix_web::test::TestRequest;
+
+    impl PartialEq for ProtoBufPayloadError {
+        fn eq(&self, other: &ProtoBufPayloadError) -> bool {
+            match *self {
+                ProtoBufPayloadError::Overflow => match *other {
+                    ProtoBufPayloadError::Overflow => true,
+                    _ => false,
+                },
+                ProtoBufPayloadError::ContentType => match *other {
+                    ProtoBufPayloadError::ContentType => true,
+                    _ => false,
+                },
+                _ => false,
+            }
+        }
+    }
+
+    #[derive(Clone, PartialEq, Message)]
+    pub struct MyObject {
+        #[prost(int32, tag = "1")]
+        pub number: i32,
+        #[prost(string, tag = "2")]
+        pub name: String,
+    }
+
+    #[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).await.unwrap();
+        assert_eq!(
+            resp.headers().get(header::CONTENT_TYPE).unwrap(),
+            "application/protobuf"
+        );
+    }
+
+    #[actix_rt::test]
+    async fn test_protobuf_message() {
+        let (req, mut pl) = TestRequest::default().to_http_parts();
+        let protobuf = ProtoBufMessage::<MyObject>::new(&req, &mut pl).await;
+        assert_eq!(protobuf.err().unwrap(), ProtoBufPayloadError::ContentType);
+
+        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::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);
+    }
+}