add request builder

backend/src/routers/user.rs

@@ -9,6 +9,7 @@ use flowy_net::response::*;
 use flowy_user::protobuf::SignUpParams;
 
 use crate::user_service::sign_up;
+use flowy_net::errors::ServerError;
 use sqlx::PgPool;
 use std::sync::Arc;
 

backend/src/user_service/auth.rs

@@ -1,6 +1,6 @@
 use anyhow::Context;
 use chrono::Utc;
-use flowy_net::response::{Code, FlowyResponse, ServerError};
+use flowy_net::{errors::ServerError, response::FlowyResponse};
 use flowy_user::{entities::SignUpResponse, protobuf::SignUpParams};
 use sqlx::{Error, PgPool, Postgres, Transaction};
 use std::sync::Arc;

backend/src/ws_service/entities/connect.rs

@@ -1,6 +1,6 @@
 use crate::ws_service::ClientMessage;
 use actix::{Message, Recipient};
-use flowy_net::response::ServerError;
+use flowy_net::errors::ServerError;
 use serde::{Deserialize, Serialize};
 use std::fmt::Formatter;
 

backend/src/ws_service/ws_server.rs

@@ -4,7 +4,7 @@ use crate::ws_service::{
 };
 use actix::{Actor, Context, Handler};
 use dashmap::DashMap;
-use flowy_net::response::ServerError;
+use flowy_net::errors::ServerError;
 
 pub struct WSServer {
     sessions: DashMap<SessionId, Session>,

rust-lib/flowy-net/Cargo.toml

@@ -15,7 +15,7 @@ serde_repr = "0.1"
 pin-project = "1.0.0"
 futures-core = { version = "0.3", default-features = false }
 log = "0.4"
-bytes = "1.0"
+bytes = { version = "1.0", features = ["serde"]}
 lazy_static = "1.4.0"
 tokio = { version = "1", features = ["full"] }
 actix-web = {version = "4.0.0-beta.8", optional = true}

rust-lib/flowy-net/src/errors.rs

@@ -0,0 +1,70 @@
+use bytes::Bytes;
+use serde::{Deserialize, Serialize, __private::Formatter};
+use serde_repr::*;
+use std::{fmt, fmt::Debug};
+
+use crate::response::FlowyResponse;
+
+#[derive(thiserror::Error, Debug, Serialize, Deserialize, Clone)]
+pub struct ServerError {
+    pub code: Code,
+    pub msg: String,
+}
+
+macro_rules! static_error {
+    ($name:ident, $status:expr) => {
+        #[allow(non_snake_case, missing_docs)]
+        pub fn $name<T: Debug>(error: T) -> ServerError {
+            let msg = format!("{:?}", error);
+            ServerError { code: $status, msg }
+        }
+    };
+}
+
+impl ServerError {
+    static_error!(internal, Code::InternalError);
+    static_error!(http, Code::HttpError);
+    static_error!(payload_none, Code::PayloadUnexpectedNone);
+}
+
+impl std::fmt::Display for ServerError {
+    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
+        let msg = format!("{:?}:{}", self.code, self.msg);
+        f.write_str(&msg)
+    }
+}
+
+impl std::convert::From<&ServerError> for FlowyResponse {
+    fn from(error: &ServerError) -> Self {
+        FlowyResponse {
+            data: Bytes::from(vec![]),
+            error: Some(error.clone()),
+        }
+    }
+}
+
+#[derive(Serialize_repr, Deserialize_repr, PartialEq, Debug, Clone)]
+#[repr(u16)]
+pub enum Code {
+    InvalidToken       = 1,
+    Unauthorized       = 3,
+    PayloadOverflow    = 4,
+    PayloadSerdeFail   = 5,
+    PayloadUnexpectedNone = 6,
+
+    ProtobufError      = 10,
+    SerdeError         = 11,
+
+    EmailAlreadyExists = 50,
+
+    ConnectRefused     = 100,
+    ConnectTimeout     = 101,
+    ConnectClose       = 102,
+    ConnectCancel      = 103,
+
+    SqlError           = 200,
+
+    HttpError          = 300,
+
+    InternalError      = 1000,
+}

rust-lib/flowy-net/src/lib.rs

@@ -1,4 +1,5 @@
 pub mod config;
+pub mod errors;
 pub mod future;
 pub mod request;
 pub mod response;

rust-lib/flowy-net/src/request/request.rs

@@ -1,26 +1,111 @@
-use crate::response::{Code, FlowyResponse, ServerError};
+use crate::{
+    errors::{Code, ServerError},
+    response::FlowyResponse,
+};
 use bytes::Bytes;
 use hyper::http;
-use protobuf::{Message, ProtobufError};
-use reqwest::{Client, Response};
+use protobuf::ProtobufError;
+use reqwest::{Client, Method, Response};
 use std::{
     convert::{TryFrom, TryInto},
     time::Duration,
 };
 use tokio::sync::oneshot;
 
+pub struct HttpRequestBuilder {
+    url: String,
+    body: Option<Bytes>,
+    response: Option<Bytes>,
+    method: Method,
+}
+
+impl HttpRequestBuilder {
+    fn new() -> Self {
+        Self {
+            url: "".to_owned(),
+            body: None,
+            response: None,
+            method: Method::GET,
+        }
+    }
+
+    pub fn get(url: &str) -> Self {
+        let mut builder = Self::new();
+        builder.url = url.to_owned();
+        builder.method = Method::GET;
+        builder
+    }
+
+    pub fn post(url: &str) -> Self {
+        let mut builder = Self::new();
+        builder.url = url.to_owned();
+        builder.method = Method::POST;
+        builder
+    }
+
+    pub fn protobuf<T1>(mut self, body: T1) -> Result<Self, ServerError>
+    where
+        T1: TryInto<Bytes, Error = ProtobufError>,
+    {
+        let body: Bytes = body.try_into()?;
+        self.body = Some(body);
+        Ok(self)
+    }
+
+    pub async fn send(mut self) -> Result<Self, ServerError> {
+        let (tx, rx) = oneshot::channel::<Result<Response, _>>();
+        // reqwest client is not 'Sync' by channel is.
+        let url = self.url.clone();
+        let body = self.body.take();
+        let method = self.method.clone();
+
+        tokio::spawn(async move {
+            let client = default_client();
+            let mut builder = client.request(method, url);
+
+            if let Some(body) = body {
+                builder = builder.body(body);
+            }
+
+            let response = builder.send().await;
+            tx.send(response);
+        });
+
+        let response = rx.await??;
+        let data = get_response_data(response).await?;
+        self.response = Some(data);
+        Ok(self)
+    }
+
+    pub fn response<T2>(mut self) -> Result<T2, ServerError>
+    where
+        T2: TryFrom<Bytes, Error = ProtobufError>,
+    {
+        let data = self.response.take();
+        match data {
+            None => {
+                let msg = format!("Request: {} receives unexpected empty body", self.url);
+                Err(ServerError::payload_none(msg))
+            },
+            Some(data) => Ok(T2::try_from(data)?),
+        }
+    }
+}
+
+#[allow(dead_code)]
 pub async fn http_post<T1, T2>(url: &str, data: T1) -> Result<T2, ServerError>
 where
     T1: TryInto<Bytes, Error = ProtobufError>,
     T2: TryFrom<Bytes, Error = ProtobufError>,
 {
-    let request_bytes: Bytes = data.try_into()?;
+    let body: Bytes = data.try_into()?;
     let url = url.to_owned();
     let (tx, rx) = oneshot::channel::<Result<Response, _>>();
 
+    // reqwest client is not 'Sync' by channel is.
     tokio::spawn(async move {
         let client = default_client();
-        let response = client.post(&url).body(request_bytes).send().await;
+        let response = client.post(&url).body(body).send().await;
         tx.send(response);
     });
 

rust-lib/flowy-net/src/response/response.rs

@@ -1,71 +1,9 @@
+use crate::errors::{Code, ServerError};
 use bytes::Bytes;
-use serde::{Deserialize, Serialize, __private::Formatter};
-use serde_repr::*;
-use std::{convert::TryInto, error::Error, fmt, fmt::Debug};
+use serde::{Deserialize, Serialize};
+use std::{convert::TryInto, error::Error, fmt::Debug};
 use tokio::sync::oneshot::error::RecvError;
 
-#[derive(thiserror::Error, Debug, Serialize, Deserialize, Clone)]
-pub struct ServerError {
-    pub code: Code,
-    pub msg: String,
-}
-
-macro_rules! static_error {
-    ($name:ident, $status:expr) => {
-        #[allow(non_snake_case, missing_docs)]
-        pub fn $name<T: Debug>(error: T) -> ServerError {
-            let msg = format!("{:?}", error);
-            ServerError { code: $status, msg }
-        }
-    };
-}
-
-impl ServerError {
-    static_error!(internal, Code::InternalError);
-    static_error!(http, Code::HttpError);
-}
-
-impl std::fmt::Display for ServerError {
-    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
-        let msg = format!("{:?}:{}", self.code, self.msg);
-        f.write_str(&msg)
-    }
-}
-
-impl std::convert::From<&ServerError> for FlowyResponse {
-    fn from(error: &ServerError) -> Self {
-        FlowyResponse {
-            data: Bytes::from(vec![]),
-            error: Some(error.clone()),
-        }
-    }
-}
-
-#[derive(Serialize_repr, Deserialize_repr, PartialEq, Debug, Clone)]
-#[repr(u16)]
-pub enum Code {
-    InvalidToken       = 1,
-    Unauthorized       = 3,
-    PayloadOverflow    = 4,
-    PayloadSerdeFail   = 5,
-
-    ProtobufError      = 6,
-    SerdeError         = 7,
-
-    EmailAlreadyExists = 50,
-
-    ConnectRefused     = 100,
-    ConnectTimeout     = 101,
-    ConnectClose       = 102,
-    ConnectCancel      = 103,
-
-    SqlError           = 200,
-
-    HttpError          = 300,
-
-    InternalError      = 1000,
-}
-
 #[derive(Debug, Serialize, Deserialize)]
 pub struct FlowyResponse {
     pub data: Bytes,

rust-lib/flowy-net/src/response/response_http.rs

@@ -1,7 +1,7 @@
 use crate::response::*;
-use actix_web::{body::Body, error::ResponseError, BaseHttpResponse, HttpResponse};
-use reqwest::StatusCode;
-use serde::Serialize;
+use actix_web::{error::ResponseError, HttpResponse};
+
+use crate::errors::ServerError;
 
 impl ResponseError for ServerError {
     fn error_response(&self) -> HttpResponse {

rust-lib/flowy-user/src/errors.rs

@@ -132,8 +132,8 @@ impl std::convert::From<flowy_sqlite::Error> for UserError {
     }
 }
 
-impl std::convert::From<flowy_net::response::ServerError> for UserError {
-    fn from(error: flowy_net::response::ServerError) -> Self {
+impl std::convert::From<flowy_net::errors::ServerError> for UserError {
+    fn from(error: flowy_net::errors::ServerError) -> Self {
         ErrorBuilder::new(UserErrCode::NetworkError)
             .error(error)
             .build()

rust-lib/flowy-user/src/services/user/user_server.rs

@@ -3,7 +3,11 @@ use crate::{
     errors::{ErrorBuilder, UserErrCode, UserError},
 };
 
-use flowy_net::{config::SIGN_UP_URL, future::ResultFuture, request::http_post};
+use flowy_net::{
+    config::SIGN_UP_URL,
+    future::ResultFuture,
+    request::{http_post, HttpRequestBuilder},
+};
 use std::sync::Arc;
 
 pub trait UserServer {
@@ -27,8 +31,12 @@ impl UserServerImpl {}
 impl UserServer for UserServerImpl {
     fn sign_up(&self, params: SignUpParams) -> ResultFuture<SignUpResponse, UserError> {
         ResultFuture::new(async move {
-            let resp = http_post(SIGN_UP_URL.as_ref(), params).await?;
-            Ok(resp)
+            let response = HttpRequestBuilder::post(SIGN_UP_URL.as_ref())
+                .protobuf(params)?
+                .send()
+                .await?
+                .response()?;
+            Ok(response)
         })
     }