Merge branch 'dev' into issue-232-binary-release

elpiel · web-flow · commit 17a7acf5577e · 2020-01-08T12:36:20.000+02:00
diff --git a/primitives/src/sentry.rs b/primitives/src/sentry.rs
@@ -62,8 +62,6 @@ pub struct Earner {
     pub promilles: u64,
 }
 
-pub type EarnerAddress = ValidatorId;
-
 #[derive(Debug, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub struct EventAggregate {
diff --git a/sentry/src/chain.rs b/sentry/src/chain.rs
@@ -1,22 +1,24 @@
 use crate::{Application, ResponseError};
+use futures::future::BoxFuture;
 use hyper::{Body, Request};
 use primitives::adapter::Adapter;
-use std::future::Future;
 
 // chain middleware function calls
 //
 // function signature
 // fn middleware(mut req: Request) -> Result<Request, ResponseError>
 
-pub async fn chain<'a, A, M, MF>(
+pub async fn chain<'a, A: Adapter + 'static, M>(
     req: Request<Body>,
     app: &'a Application<A>,
     middlewares: Vec<M>,
 ) -> Result<Request<Body>, ResponseError>
 where
-    A: Adapter,
-    MF: Future<Output = Result<Request<Body>, ResponseError>> + Send,
-    M: FnMut(Request<Body>, &'a Application<A>) -> MF,
+    M: FnMut(
+            Request<Body>,
+            &'a Application<A>,
+        ) -> BoxFuture<'a, Result<Request<Body>, ResponseError>>
+        + 'static,
 {
     let mut req = Ok(req);
 
diff --git a/sentry/src/lib.rs b/sentry/src/lib.rs
@@ -9,6 +9,7 @@ use crate::middleware::cors::{cors, Cors};
 use crate::routes::channel::channel_status;
 use crate::routes::event_aggregate::list_channel_event_aggregates;
 use crate::routes::validator_message::{extract_params, list_validator_messages};
+use futures::future::{BoxFuture, FutureExt};
 use hyper::{Body, Method, Request, Response, StatusCode};
 use lazy_static::lazy_static;
 use primitives::adapter::Adapter;
@@ -20,12 +21,10 @@ use routes::cfg::config;
 use routes::channel::{channel_list, create_channel, last_approved};
 use slog::{error, Logger};
 use std::collections::HashMap;
-
 pub mod middleware {
     pub mod auth;
     pub mod channel;
     pub mod cors;
-    pub mod event_aggregate;
 }
 
 pub mod routes {
@@ -48,28 +47,23 @@ lazy_static! {
     static ref CHANNEL_STATUS_BY_CHANNEL_ID: Regex = Regex::new(r"^/channel/0x([a-zA-Z0-9]{64})/status/?$").expect("The regex should be valid");
     // Only the initial Regex to be matched.
     static ref CHANNEL_VALIDATOR_MESSAGES: Regex = Regex::new(r"^/channel/0x([a-zA-Z0-9]{64})/validator-messages(/.*)?$").expect("The regex should be valid");
-    static ref CHANNEL_EVENTS_AGGREGATES: Regex = Regex::new(r"^/channel/0x([a-zA-Z0-9]{64})/events-aggregates(/|/0x[a-zA-Z0-9]{40}/?)?$").expect("The regex should be valid");
-    // @TODO define other regex routes
+    static ref CHANNEL_EVENTS_AGGREGATES: Regex = Regex::new(r"^/channel/0x([a-zA-Z0-9]{64})/events-aggregates/?$").expect("The regex should be valid");
     static ref ANALYTICS_BY_CHANNEL_ID: Regex = Regex::new(r"^/analytics/0x([a-zA-Z0-9]{64})/?$").expect("The regex should be valid");
     static ref ADVERTISER_ANALYTICS_BY_CHANNEL_ID: Regex = Regex::new(r"^/analytics/for-advertiser/0x([a-zA-Z0-9]{64})/?$").expect("The regex should be valid");
 }
 
-async fn config_middleware<A: Adapter>(
-    req: Request<Body>,
-    _: &Application<A>,
-) -> Result<Request<Body>, ResponseError> {
-    Ok(req)
-}
-
-async fn auth_required_middleware<A: Adapter>(
+fn auth_required_middleware<'a, A: Adapter>(
     req: Request<Body>,
     _: &Application<A>,
-) -> Result<Request<Body>, ResponseError> {
-    if req.extensions().get::<Session>().is_some() {
-        Ok(req)
-    } else {
-        Err(ResponseError::Unauthorized)
+) -> BoxFuture<'a, Result<Request<Body>, ResponseError>> {
+    async move {
+        if req.extensions().get::<Session>().is_some() {
+            Ok(req)
+        } else {
+            Err(ResponseError::Unauthorized)
+        }
     }
+    .boxed()
 }
 
 #[derive(Debug)]
@@ -137,8 +131,7 @@ impl<A: Adapter + 'static> Application<A> {
 
             ("/analytics", &Method::GET) => analytics(req, &self).await,
             ("/analytics/for-advertiser", &Method::GET) => {
-                // @TODO get advertiser channels
-                let req = match chain(req, &self, vec![auth_required_middleware]).await {
+                let req = match chain(req, &self, vec![Box::new(auth_required_middleware)]).await {
                     Ok(req) => req,
                     Err(error) => {
                         return map_response_error(error);
@@ -147,7 +140,7 @@ impl<A: Adapter + 'static> Application<A> {
                 advertiser_analytics(req, &self).await
             }
             ("/analytics/for-publisher", &Method::GET) => {
-                let req = match chain(req, &self, vec![auth_required_middleware]).await {
+                let req = match chain(req, &self, vec![Box::new(auth_required_middleware)]).await {
                     Ok(req) => req,
                     Err(error) => {
                         return map_response_error(error);
@@ -170,7 +163,7 @@ impl<A: Adapter + 'static> Application<A> {
     }
 }
 
-async fn analytics_router<A: Adapter>(
+async fn analytics_router<A: Adapter + 'static>(
     mut req: Request<Body>,
     app: &Application<A>,
 ) -> Result<Response<Body>, ResponseError> {
@@ -184,15 +177,15 @@ async fn analytics_router<A: Adapter>(
                     .map_or("".to_string(), |m| m.as_str().to_string())]);
                 req.extensions_mut().insert(param);
 
-                let req = chain(req, app, vec![channel_load]).await?;
+                let req = chain(req, app, vec![Box::new(channel_load)]).await?;
                 analytics(req, app).await
             } else if let Some(caps) = ADVERTISER_ANALYTICS_BY_CHANNEL_ID.captures(route) {
                 let param = RouteParams(vec![caps
                     .get(1)
                     .map_or("".to_string(), |m| m.as_str().to_string())]);
                 req.extensions_mut().insert(param);
 
-                let req = chain(req, app, vec![auth_required_middleware]).await?;
+                let req = auth_required_middleware(req, app).await?;
                 advertiser_analytics(req, app).await
             } else {
                 Err(ResponseError::NotFound)
@@ -202,7 +195,7 @@ async fn analytics_router<A: Adapter>(
     }
 }
 
-async fn channels_router<A: Adapter>(
+async fn channels_router<A: Adapter + 'static>(
     mut req: Request<Body>,
     app: &Application<A>,
 ) -> Result<Response<Body>, ResponseError> {
@@ -217,15 +210,6 @@ async fn channels_router<A: Adapter>(
             .map_or("".to_string(), |m| m.as_str().to_string())]);
         req.extensions_mut().insert(param);
 
-        // example with middleware
-        // @TODO remove later
-        let req = match chain(req, app, vec![config_middleware]).await {
-            Ok(req) => req,
-            Err(error) => {
-                return Err(error);
-            }
-        };
-
         last_approved(req, app).await
     } else if let (Some(caps), &Method::GET) =
         (CHANNEL_STATUS_BY_CHANNEL_ID.captures(&path), method)
@@ -235,13 +219,7 @@ async fn channels_router<A: Adapter>(
             .map_or("".to_string(), |m| m.as_str().to_string())]);
         req.extensions_mut().insert(param);
 
-        let req = match chain(req, app, vec![channel_load]).await {
-            Ok(req) => req,
-            Err(error) => {
-                return Err(error);
-            }
-        };
-
+        let req = channel_load(req, app).await?;
         channel_status(req, app).await
     } else if let (Some(caps), &Method::GET) = (CHANNEL_VALIDATOR_MESSAGES.captures(&path), method)
     {
@@ -251,7 +229,7 @@ async fn channels_router<A: Adapter>(
 
         req.extensions_mut().insert(param);
 
-        let req = match chain(req, app, vec![channel_load]).await {
+        let req = match chain(req, app, vec![Box::new(channel_load)]).await {
             Ok(req) => req,
             Err(error) => {
                 return Err(error);
@@ -280,7 +258,7 @@ async fn channels_router<A: Adapter>(
         ]);
         req.extensions_mut().insert(param);
 
-        let req = chain(req, app, vec![channel_load]).await?;
+        let req = chain(req, app, vec![Box::new(channel_load)]).await?;
 
         list_channel_event_aggregates(req, app).await
     } else {
diff --git a/sentry/src/middleware/channel.rs b/sentry/src/middleware/channel.rs
@@ -1,61 +1,68 @@
 use crate::db::{get_channel_by_id, get_channel_by_id_and_validator};
 use crate::{Application, ResponseError, RouteParams};
+use futures::future::{BoxFuture, FutureExt};
 use hex::FromHex;
 use hyper::{Body, Request};
 use primitives::adapter::Adapter;
 use primitives::{ChannelId, ValidatorId};
 use std::convert::TryFrom;
 
 /// channel_load & channel_if_exist
-pub async fn channel_load<A: Adapter>(
+pub fn channel_load<'a, A: Adapter + 'static>(
     mut req: Request<Body>,
-    app: &Application<A>,
-) -> Result<Request<Body>, ResponseError> {
-    let id = req
-        .extensions()
-        .get::<RouteParams>()
-        .ok_or_else(|| ResponseError::BadRequest("Route params not found".to_string()))?
-        .get(0)
-        .ok_or_else(|| ResponseError::BadRequest("No id".to_string()))?;
-
-    let channel_id = ChannelId::from_hex(id)
-        .map_err(|_| ResponseError::BadRequest("Wrong Channel Id".to_string()))?;
-    let channel = get_channel_by_id(&app.pool, &channel_id)
-        .await?
-        .ok_or_else(|| ResponseError::NotFound)?;
-
-    req.extensions_mut().insert(channel);
-
-    Ok(req)
+    app: &'a Application<A>,
+) -> BoxFuture<'a, Result<Request<Body>, ResponseError>> {
+    async move {
+        let id = req
+            .extensions()
+            .get::<RouteParams>()
+            .ok_or_else(|| ResponseError::BadRequest("Route params not found".to_string()))?
+            .get(0)
+            .ok_or_else(|| ResponseError::BadRequest("No id".to_string()))?;
+
+        let channel_id = ChannelId::from_hex(id)
+            .map_err(|_| ResponseError::BadRequest("Wrong Channel Id".to_string()))?;
+        let channel = get_channel_by_id(&app.pool, &channel_id)
+            .await?
+            .ok_or_else(|| ResponseError::NotFound)?;
+
+        req.extensions_mut().insert(channel);
+
+        Ok(req)
+    }
+    .boxed()
 }
 
-pub async fn channel_if_active<A: Adapter>(
+pub fn channel_if_active<'a, A: Adapter + 'static>(
     mut req: Request<Body>,
-    app: &Application<A>,
-) -> Result<Request<Body>, ResponseError> {
-    let route_params = req
-        .extensions()
-        .get::<RouteParams>()
-        .ok_or_else(|| ResponseError::BadRequest("Route params not found".to_string()))?;
+    app: &'a Application<A>,
+) -> BoxFuture<'a, Result<Request<Body>, ResponseError>> {
+    async move {
+        let route_params = req
+            .extensions()
+            .get::<RouteParams>()
+            .ok_or_else(|| ResponseError::BadRequest("Route params not found".to_string()))?;
 
-    let id = route_params
-        .get(0)
-        .ok_or_else(|| ResponseError::BadRequest("No id".to_string()))?;
+        let id = route_params
+            .get(0)
+            .ok_or_else(|| ResponseError::BadRequest("No id".to_string()))?;
 
-    let channel_id = ChannelId::from_hex(id)
-        .map_err(|_| ResponseError::BadRequest("Wrong Channel Id".to_string()))?;
+        let channel_id = ChannelId::from_hex(id)
+            .map_err(|_| ResponseError::BadRequest("Wrong Channel Id".to_string()))?;
 
-    let validator_id = route_params
-        .get(1)
-        .ok_or_else(|| ResponseError::BadRequest("No Validator Id".to_string()))?;
-    let validator_id = ValidatorId::try_from(&validator_id)
-        .map_err(|_| ResponseError::BadRequest("Wrong Validator Id".to_string()))?;
+        let validator_id = route_params
+            .get(1)
+            .ok_or_else(|| ResponseError::BadRequest("No Validator Id".to_string()))?;
+        let validator_id = ValidatorId::try_from(&validator_id)
+            .map_err(|_| ResponseError::BadRequest("Wrong Validator Id".to_string()))?;
 
-    let channel = get_channel_by_id_and_validator(&app.pool, &channel_id, &validator_id)
-        .await?
-        .ok_or_else(|| ResponseError::NotFound)?;
+        let channel = get_channel_by_id_and_validator(&app.pool, &channel_id, &validator_id)
+            .await?
+            .ok_or_else(|| ResponseError::NotFound)?;
 
-    req.extensions_mut().insert(channel);
+        req.extensions_mut().insert(channel);
 
-    Ok(req)
+        Ok(req)
+    }
+    .boxed()
 }
diff --git a/sentry/src/middleware/event_aggregate.rs b/sentry/src/middleware/event_aggregate.rs

Original file line number	Diff line number	Diff line change
`@@ -62,8 +62,6 @@ pub struct Earner {`
`62`	`62`	`pub promilles: u64,`
`63`	`63`	`}`
`64`	`64`
`65`		`-pub type EarnerAddress = ValidatorId;`
`66`		`-`
`67`	`65`	`#[derive(Debug, Serialize, Deserialize)]`
`68`	`66`	`#[serde(rename_all = "camelCase")]`
`69`	`67`	`pub struct EventAggregate {`