Merge pull request #19 from fiji-flo/async

Async

Author: fiji-flo

Cargo.lock

@@ -1,25 +1,27 @@
 [package]
 name = "dino-park-lookout"
-version = "0.2.0"
+version = "0.3.0"
 authors = ["Florian Merz <flomerz@gmail.com>"]
 edition = "2018"
 
 [dependencies]
-dino_park_gate = { git = "https://github.com/fiji-flo/dino-park-gate", branch = "0.2.0", version = "0.2.0" }
-cis_client = { git = "https://github.com/fiji-flo/cis_client-rust", branch = "0.2.0", version = "0.2.0" }
-cis_profile = { git = "https://github.com/fiji-flo/cis_profile-rust", branch = "0.2.0", version = "0.2.0" }
-actix-web = "1.0"
-actix-cors = "0.1"
-futures = "0.1"
+dino_park_gate = { git = "https://github.com/fiji-flo/dino-park-gate", branch = "0.3.0", version = "0.3.0" }
+cis_client = { git = "https://github.com/fiji-flo/cis_client-rust", branch = "0.4.0", version = "0.4.0", features = ["sync"] }
+cis_profile = { git = "https://github.com/fiji-flo/cis_profile-rust", branch = "0.3.1", version = "0.3.1" }
+actix-web = "2.0"
+actix-rt = "1.0"
+actix-cors = "0.2"
+futures = "0.3"
 log = "0.4"
 env_logger = "0.7"
 serde = "1.0.80"
 serde_json = "1.0.32"
 serde_derive = "1.0.80"
-reqwest = "0.9"
+reqwest = { version = "0.10", features = ["json"] }
 chrono = "0.4"
 config = "0.10"
 failure = "0.1"
 failure_derive = "0.1"
 biscuit = "0.4"
 url = "2.1"
+tokio = "0.2"

Cargo.toml

@@ -1,22 +1,23 @@
 export RUSTUP_HOME=/usr/local/rustup
 export CARGO_HOME=/usr/local/cargo
 export PATH=/usr/local/cargo/bin:$PATH
-export RUST_VERSION=1.38.0
+export RUST_VERSION=1.40.0
 
 set -eux
+
 dpkgArch="$(dpkg --print-architecture)"
 case "${dpkgArch##*-}" in
-    amd64) rustArch='x86_64-unknown-linux-gnu'; rustupSha256='a46fe67199b7bcbbde2dcbc23ae08db6f29883e260e23899a88b9073effc9076' ;;
-    armhf) rustArch='armv7-unknown-linux-gnueabihf'; rustupSha256='6af5abbbae02e13a9acae29593ec58116ab0e3eb893fa0381991e8b0934caea1' ;;
-    arm64) rustArch='aarch64-unknown-linux-gnu'; rustupSha256='51862e576f064d859546cca5f3d32297092a850861e567327422e65b60877a1b' ;;
-    i386) rustArch='i686-unknown-linux-gnu'; rustupSha256='91456c3e6b2a3067914b3327f07bc182e2a27c44bff473263ba81174884182be' ;;
+    amd64) rustArch='x86_64-unknown-linux-gnu'; rustupSha256='e68f193542c68ce83c449809d2cad262cc2bbb99640eb47c58fc1dc58cc30add' ;;
+    armhf) rustArch='armv7-unknown-linux-gnueabihf'; rustupSha256='7c1c329a676e50c287d8183b88f30cd6afd0be140826a9fbbc0e3d717fab34d7' ;;
+    arm64) rustArch='aarch64-unknown-linux-gnu'; rustupSha256='d861cc86594776414de001b96964be645c4bfa27024052704f0976dc3aed1b18' ;;
+    i386) rustArch='i686-unknown-linux-gnu'; rustupSha256='89f1f797dca2e5c1d75790c3c6b7be0ee473a7f4eca9663e623a41272a358da0' ;;
     *) echo >&2 "unsupported architecture: ${dpkgArch}"; exit 1 ;;
 esac
-url="https://static.rust-lang.org/rustup/archive/1.18.3/${rustArch}/rustup-init"
+url="https://static.rust-lang.org/rustup/archive/1.20.2/${rustArch}/rustup-init"
 wget "$url"
 echo "${rustupSha256} *rustup-init" | sha256sum -c -
 chmod +x rustup-init
-./rustup-init -y --no-modify-path --default-toolchain $RUST_VERSION
+./rustup-init -y --no-modify-path --profile minimal --default-toolchain $RUST_VERSION
 rm rustup-init
 chmod -R a+w $RUSTUP_HOME $CARGO_HOME
 rustup --version

ci/install_deps.sh

@@ -8,4 +8,6 @@ pub enum UpdateError {
     SearchUpdate(Error),
     #[fail(display = "error updating groups: {}", _0)]
     GroupsUpdate(Error),
+    #[fail(display = "error updating")]
+    Other,
 }

src/error.rs

@@ -10,7 +10,7 @@ use actix_web::HttpResponse;
 use actix_web::Result;
 use serde_json::json;
 
-fn update_event<U: UpdaterClient + Clone + 'static>(
+async fn update_event<U: UpdaterClient + Clone + 'static>(
     updater: Data<U>,
     n: Json<Notification>,
 ) -> Result<HttpResponse> {
@@ -27,7 +27,8 @@ pub fn update_app<U: UpdaterClient + Clone + Send + 'static>(
                 .allowed_methods(vec!["POST"])
                 .allowed_headers(vec![http::header::AUTHORIZATION, http::header::ACCEPT])
                 .allowed_header(http::header::CONTENT_TYPE)
-                .max_age(3600),
+                .max_age(3600)
+                .finish(),
         )
         .data(updater)
         .service(web::resource("").route(web::post().to(update_event::<U>)))

src/events/app.rs

@@ -16,7 +16,8 @@ pub fn healthz_app() -> impl HttpServiceFactory {
                 .allowed_methods(vec!["GET", "HEAD"])
                 .allowed_headers(vec![http::header::AUTHORIZATION, http::header::ACCEPT])
                 .allowed_header(http::header::CONTENT_TYPE)
-                .max_age(3600),
+                .max_age(3600)
+                .finish(),
         )
         .service(web::resource("").to(healthz))
 }

src/healthz.rs

@@ -1,8 +1,8 @@
 use crate::bulk::Bulk;
 use crate::settings::DinoParkSettings;
+use crate::updater::send_profile;
 use crate::updater::UpdaterClient;
 use actix_cors::Cors;
-use actix_web::client::Client;
 use actix_web::dev::HttpServiceFactory;
 use actix_web::error;
 use actix_web::error::Error;
@@ -13,61 +13,37 @@ use actix_web::web::Json;
 use actix_web::HttpResponse;
 use actix_web::Result;
 use cis_profile::schema::Profile;
-use futures::Future;
+use futures::future::TryFutureExt;
 use serde_json::json;
 use serde_json::Value;
 
-pub fn internal_update(
-    dp: &DinoParkSettings,
-    profile: &Profile,
-) -> impl Future<Item = Value, Error = Error> {
-    let id = profile
-        .user_id
-        .value
-        .clone()
-        .unwrap_or_else(|| String::from("unknown"));
-    info!("internally updating profile for: {}", &id);
-    let id_c = id.clone();
-    let orgchart_update = Client::default()
-        .post(&dp.orgchart_update_endpoint)
-        .send_json(profile)
-        .map(move |_| info!("internally updated orgchart for: {}", id));
-
-    orgchart_update
-        .join(
-            Client::default()
-                .post(&dp.search_update_endpoint)
-                .send_json(profile)
-                .map(move |_| info!("internally updated search for: {}", id_c)),
-        )
-        .map(|_| json!({}))
-        .map_err(Into::into)
+pub async fn internal_update(dp: &DinoParkSettings, profile: Profile) -> Result<Value, Error> {
+    send_profile(dp, profile).map_err(Into::into).await
 }
 
-fn internal_update_event(
+async fn internal_update_event(
     dino_park_settings: Data<DinoParkSettings>,
     profile: Json<Profile>,
-) -> impl Future<Item = HttpResponse, Error = Error> {
+) -> Result<HttpResponse, Error> {
     let id = profile
         .user_id
         .value
         .clone()
         .unwrap_or_else(|| String::from("unknown"));
     info!("internally updating profile for: {}", &id);
     let id_c = id.clone();
-    internal_update(&dino_park_settings, &profile)
-        .map(move |res| {
-            info!("internally updated profile for {}", id);
-            HttpResponse::Ok().json(res)
-        })
-        .map_err(move |e| {
+    let res = internal_update(&dino_park_settings, profile.into_inner()).await;
+    info!("internally updated profile for {}", id);
+    match res {
+        Ok(j) => Ok(HttpResponse::Ok().json(j)),
+        Err(e) => {
             error!("failed to internally update profile for {}: {}", id_c, e);
-            error::ErrorInternalServerError(e)
-        })
-        .map_err(Into::into)
+            Err(error::ErrorInternalServerError(e))
+        }
+    }
 }
 
-fn bulk_update<U: UpdaterClient + Clone + 'static>(
+async fn bulk_update<U: UpdaterClient + Clone + 'static>(
     updater: Data<U>,
     bulk: Json<Bulk>,
 ) -> Result<HttpResponse> {
@@ -85,11 +61,12 @@ pub fn internal_app<U: UpdaterClient + Clone + Send + 'static>(
                 .allowed_methods(vec!["POST"])
                 .allowed_headers(vec![http::header::AUTHORIZATION, http::header::ACCEPT])
                 .allowed_header(http::header::CONTENT_TYPE)
-                .max_age(3600),
+                .max_age(3600)
+                .finish(),
         )
         .data(updater)
         .data(dino_park_settings)
-        .data(web::JsonConfig::default().limit(1_048_576))
+        .app_data(web::JsonConfig::default().limit(1_048_576))
         .service(web::resource("/bulk").route(web::post().to(bulk_update::<U>)))
-        .service(web::resource("/update").route(web::post().to_async(internal_update_event)))
+        .service(web::resource("/update").route(web::post().to(internal_update_event)))
 }

src/internal/app.rs

@@ -20,6 +20,7 @@ use crate::internal::app::internal_app;
 use crate::updater::InternalUpdater;
 use crate::updater::Updater;
 use crate::updater::UpdaterClient;
+use actix_rt::System;
 use actix_web::middleware::Logger;
 use actix_web::web;
 use actix_web::App;
@@ -34,7 +35,7 @@ use std::thread::spawn;
 fn main() -> Result<(), Error> {
     ::std::env::set_var(
         "RUST_LOG",
-        "actix_web=info,dino_park_lookout=info,dino_park_gate=info",
+        "actix_web=info,dino_park_lookout=info,dino_park_gate=info,cis_client=info,shared_expiry_get=info",
     );
     env_logger::init();
     info!("building the lookout");
@@ -43,9 +44,10 @@ fn main() -> Result<(), Error> {
         .map_err(|e| format_err!("unable to create cis_client: {}", e))?;
     let dino_park = s.dino_park.clone();
     let validation_settings = s.auth.validation.clone();
-    let provider = Provider::from_issuer(&s.auth.issuer)?;
+    let mut rt = tokio::runtime::Runtime::new()?;
+    let provider = rt.block_on(Provider::from_issuer(&s.auth.issuer))?;
     // Start http server
-    let updater = InternalUpdater::new(cis_client.clone(), dino_park.clone());
+    let updater = InternalUpdater::new(cis_client, dino_park.clone());
 
     let client = updater.client();
     let stop_client = updater.client();
@@ -54,7 +56,6 @@ fn main() -> Result<(), Error> {
             error!("unable to start updater: {}", e);
         }
     });
-
     let server = HttpServer::new(move || {
         let auth_middleware = SimpleAuth {
             checker: provider.clone(),
@@ -74,9 +75,9 @@ fn main() -> Result<(), Error> {
             .service(healthz_app())
     })
     .bind("0.0.0.0:8082")?;
-    server
-        .run()
-        .map_err(|_| format_err!("unable to run server!"))?;
+
+    System::new("lookout-actix-rt").block_on(async move { server.run().await })?;
+
     info!("Stopped http server");
     stop_client.stop();
     updater_thread