To warp 0.2

Author: Mark-Simulacrum

Cargo.lock

@@ -10,7 +10,7 @@ opt-level = 0
 
 [dependencies]
 base64 = "0.20.0"
-bytes = "0.4.9"
+bytes = "0.5"
 bytes_1 = { version = "1", package = "bytes" }
 chrono = { version = "0.4", features = ["serde"] }
 crates-index = "0.18"
@@ -21,8 +21,8 @@ docsrs-metadata = { git = "https://github.com/rust-lang/docs.rs/" }
 dotenv = "0.15"
 failure = "0.1.3"
 flate2 = "1"
-http = "0.1.10"
-hyper = "0.12.8"
+http = "0.2"
+hyper = "0.13"
 lazy_static = "1.0"
 mime = "0.3.1"
 minifier = { version = "0.2", features = ["html"] }
@@ -43,7 +43,7 @@ tera = "0.11.7"
 toml = "0.4.6"
 url = "2"
 walkdir = "2"
-warp = "0.1.9"
+warp = "0.2"
 log = "0.4.6"
 env_logger = "0.10.0"
 hmac = "0.12"

Cargo.toml

@@ -82,9 +82,12 @@ pub fn auth_filter(
     token_type: TokenType,
 ) -> impl Filter<Extract = (AuthDetails,), Error = Rejection> + Clone {
     warp::header::headers_cloned().and_then(move |headers| {
-        match check_auth(&data, &headers, token_type) {
-            Some(details) => Ok(details),
-            None => Err(warp::reject::custom(HttpError::Forbidden.compat())),
+        let data = data.clone();
+        async move {
+            match check_auth(&data, &headers, token_type) {
+                Some(details) => Ok(details),
+                None => Err(warp::reject::custom(HttpError::Forbidden)),
+            }
         }
     })
 }

src/server/auth.rs

@@ -37,6 +37,8 @@ pub enum HttpError {
     Forbidden,
 }
 
+impl warp::reject::Reject for HttpError {}
+
 #[derive(Clone)]
 pub struct Data {
     pub config: Config,
@@ -135,7 +137,12 @@ pub fn run(config: Config, bind: SocketAddr) -> Fallible<()> {
             },
         );
 
-    warp::serve(routes).run(bind);
+    let rt = tokio::runtime::Builder::new_multi_thread()
+        .enable_all()
+        .build()?;
+    rt.block_on(async move {
+        warp::serve(routes).run(bind).await;
+    });
 
     Ok(())
 }

src/server/mod.rs

@@ -8,7 +8,7 @@ use crate::server::messages::Message;
 use crate::server::{Data, GithubData, HttpError};
 use crossbeam_channel::Sender;
 use failure::Compat;
-use http::{Response, StatusCode};
+use http::Response;
 use hyper::Body;
 use std::collections::HashMap;
 use std::sync::{Arc, Condvar, Mutex};
@@ -32,46 +32,46 @@ pub fn routes(
     let mutex_filter = warp::any().map(move || mutex.clone());
     let github_data_filter = warp::any().map(move || github_data.clone());
 
-    let config = warp::post2()
+    let config = warp::post()
         .and(warp::path("config"))
         .and(warp::path::end())
         .and(warp::body::json())
         .and(data_filter.clone())
         .and(auth_filter(data.clone(), TokenType::Agent))
         .map(endpoint_config);
 
-    let next_experiment = warp::post2()
+    let next_experiment = warp::post()
         .and(warp::path("next-experiment"))
         .and(warp::path::end())
         .and(mutex_filter.clone())
         .and(github_data_filter)
         .and(auth_filter(data.clone(), TokenType::Agent))
         .map(endpoint_next_experiment);
 
-    let next_crate = warp::post2()
+    let next_crate = warp::post()
         .and(warp::path("next-crate"))
         .and(warp::path::end())
         .and(warp::body::json())
         .and(data_filter.clone())
         .and(auth_filter(data.clone(), TokenType::Agent))
         .map(endpoint_next_crate);
 
-    let record_progress = warp::post2()
+    let record_progress = warp::post()
         .and(warp::path("record-progress"))
         .and(warp::path::end())
         .and(warp::body::json())
         .and(data_filter.clone())
         .and(auth_filter(data.clone(), TokenType::Agent))
         .map(endpoint_record_progress);
 
-    let heartbeat = warp::post2()
+    let heartbeat = warp::post()
         .and(warp::path("heartbeat"))
         .and(warp::path::end())
         .and(data_filter)
         .and(auth_filter(data.clone(), TokenType::Agent))
         .map(endpoint_heartbeat);
 
-    let error = warp::post2()
+    let error = warp::post()
         .and(warp::path("error"))
         .and(warp::path::end())
         .and(warp::body::json())
@@ -355,10 +355,10 @@ fn handle_results(resp: Fallible<Response<Body>>) -> Response<Body> {
     }
 }
 
-fn handle_errors(err: Rejection) -> Result<Response<Body>, Rejection> {
-    let error = if let Some(compat) = err.find_cause::<Compat<HttpError>>() {
+async fn handle_errors(err: Rejection) -> Result<Response<Body>, Rejection> {
+    let error = if let Some(compat) = err.find::<Compat<HttpError>>() {
         Some(*compat.get_ref())
-    } else if let StatusCode::NOT_FOUND | StatusCode::METHOD_NOT_ALLOWED = err.status() {
+    } else if err.is_not_found() {
         Some(HttpError::NotFound)
     } else {
         None

src/server/routes/agent.rs

@@ -12,7 +12,7 @@ pub fn routes(
 ) -> impl Filter<Extract = (Response<Body>,), Error = Rejection> + Clone {
     let data_filter = warp::any().map(move || data.clone());
 
-    warp::get2()
+    warp::get()
         .and(warp::path::end())
         .and(data_filter)
         .map(|data| match endpoint_metrics(data) {

src/server/routes/metrics.rs

@@ -29,25 +29,25 @@ pub fn routes(
 ) -> impl Filter<Extract = (Response<Body>,), Error = Rejection> + Clone {
     let data_filter = warp::any().map(move || data.clone());
 
-    let queue = warp::get2()
+    let queue = warp::get()
         .and(warp::path::end())
         .and(data_filter.clone())
         .map(experiments::endpoint_queue);
 
-    let experiment = warp::get2()
+    let experiment = warp::get()
         .and(warp::path("ex"))
         .and(warp::path::param())
         .and(warp::path::end())
         .and(data_filter.clone())
         .map(experiments::endpoint_experiment);
 
-    let agents = warp::get2()
+    let agents = warp::get()
         .and(warp::path("agents"))
         .and(warp::path::end())
         .and(data_filter)
         .map(agents::endpoint_list);
 
-    let assets = warp::get2()
+    let assets = warp::get()
         .and(warp::path("assets"))
         .and(warp::path::param())
         .and(warp::path::end())
@@ -144,18 +144,19 @@ fn handle_results(resp: Fallible<Response<Body>>) -> Response<Body> {
     }
 }
 
-fn handle_errors(err: Rejection) -> Result<Response<Body>, Rejection> {
-    match err.status() {
-        StatusCode::NOT_FOUND | StatusCode::METHOD_NOT_ALLOWED => match error_404() {
-            Ok(resp) => Ok(resp),
+async fn handle_errors(err: Rejection) -> Result<Response<Body>, Rejection> {
+    if err.is_not_found() {
+        match error_404() {
+            Ok(resp) => return Ok(resp),
             Err(err) => {
                 error!("failed to render 404 page!");
                 crate::utils::report_failure(&err);
-                Ok(error_500())
+                return Ok(error_500());
             }
-        },
-        _ => Err(err),
+        }
     }
+
+    Err(err)
 }
 
 fn render_template<C: Serialize>(name: &str, context: &C) -> Fallible<Response<Body>> {

src/server/routes/ui/mod.rs

@@ -6,13 +6,13 @@ use crate::server::github::{EventIssueComment, Issue, Repository};
 use crate::server::messages::Message;
 use crate::server::routes::webhooks::args::Command;
 use crate::server::{Data, GithubData};
-use bytes::buf::Buf;
+use bytes::Bytes;
 use hmac::{Hmac, Mac};
 use http::{HeaderMap, Response, StatusCode};
 use hyper::Body;
 use std::str::FromStr;
 use std::sync::Arc;
-use warp::{self, filters::body::FullBody, Filter, Rejection};
+use warp::{self, Filter, Rejection};
 
 fn process_webhook(
     payload: &[u8],
@@ -194,7 +194,7 @@ fn receive_endpoint(
     data: Arc<Data>,
     github_data: Arc<GithubData>,
     headers: HeaderMap,
-    body: FullBody,
+    body: Bytes,
 ) -> Fallible<()> {
     let signature = headers
         .get("X-Hub-Signature")
@@ -209,27 +209,32 @@ fn receive_endpoint(
         .and_then(|h| h.to_str().ok())
         .ok_or_else(|| err_msg("missing header Host\n"))?;
 
-    process_webhook(body.bytes(), host, signature, event, &data, &github_data)
+    process_webhook(&body[..], host, signature, event, &data, &github_data)
 }
 
 pub fn routes(
     data: Arc<Data>,
     github_data: Option<Arc<GithubData>>,
 ) -> impl Filter<Extract = (Response<Body>,), Error = Rejection> + Clone {
     let data_filter = warp::any().map(move || data.clone());
-    let github_data_filter = warp::any().and_then(move || match github_data.clone() {
-        Some(github_data) => Ok(github_data),
-        None => Err(warp::reject::not_found()),
+    let github_data_filter = warp::any().and_then(move || {
+        let g = github_data.clone();
+        async move {
+            match g {
+                Some(github_data) => Ok(github_data),
+                None => Err(warp::reject::not_found()),
+            }
+        }
     });
 
-    warp::post2()
+    warp::post()
         .and(warp::path::end())
         .and(data_filter)
         .and(github_data_filter)
         .and(warp::header::headers_cloned())
-        .and(warp::body::concat())
+        .and(warp::body::bytes())
         .map(
-            |data: Arc<Data>, github_data: Arc<GithubData>, headers: HeaderMap, body: FullBody| {
+            |data: Arc<Data>, github_data: Arc<GithubData>, headers: HeaderMap, body: Bytes| {
                 let mut resp: Response<Body>;
                 match receive_endpoint(data, github_data, headers, body) {
                     Ok(()) => resp = Response::new("OK\n".into()),