Cook a Microservice

cargo install mdbook
mdbook serve

cd ratings
PORT=7878 RUST_LOG=info cargo run

Vérifier qu'il marche:

curl localhost:7878/ratings/0
curl localhost:7878/ratings/0 -d '{"reviewer":"me","rating":3}'
curl localhost:7878/ratings/0

cargo install cargo-watch
cd reviews
RUST_LOG=warn RATINGS_URL=http://127.0.0.1:7878 cargo watch --ignore db.sqlite -x 'run'

Dans lib.rs

extern crate actix;
extern crate actix_web;
extern crate futures;

extern crate serde;
#[macro_use]
extern crate serde_derive;

use actix_web::middleware::Logger;
use actix_web::{server, App};

pub struct AppState {
}

Dans la fonction run

let sys = actix::System::new("reviews");

server::new(move || {
    App::with_state(AppState {
    }).middleware(Logger::default())
}).bind(addr)
    .unwrap()
    .start();

let _ = sys.run();

Un serveur http qui écoute, mais ne répond que des 404

curl localhost:9080 -i
HTTP/1.1 404 Not Found
content-length: 0
date: Thu, 19 Apr 2018 21:36:42 GMT

Dans reviews.rs

use std::collections::HashMap;

use actix_web::{error, AsyncResponder, HttpResponse, Json, Path, State};
use futures::Future;
use futures;

Définir comment extraire le product_id du path

#[derive(Deserialize)]
pub struct ProductId {
    product_id: i32,
}

pub fn reviews(
    product_id: Path<ProductId>,
    state: State<super::AppState>,
) -> Box<Future<Item = HttpResponse, Error = error::Error>> {
    let product_id = product_id.product_id;

    unimplemented!()
}

POST d'une review

#[derive(Debug, Deserialize, Serialize)]
pub struct NewReview {}

pub fn create_review(
    product_id: Path<ProductId>,
    review: Json<NewReview>,
    state: State<super::AppState>,
) -> Box<Future<Item = HttpResponse, Error = error::Error>> {
    let product_id = product_id.product_id;

    unimplemented!()
}

Dans lib.rs

use actix_web::http;

mod reviews;

server::new(move || {
    App::with_state(AppState {
    }).middleware(Logger::default())
        .resource("/reviews/{product_id}", |r| {
            r.method(http::Method::GET).with2(reviews::reviews);
            r.method(http::Method::POST).with3(reviews::create_review);
        })
}).bind(addr)
    .unwrap()
    .start();

Les routes définies ne répondent pas, et un log apparaît quand on les appelle

curl localhost:9080/reviews/0 -i
curl: (52) Empty reply from server

thread 'arbiter:"85373381-235b-47af-bde8-4d85e5b778d8":"actor"' panicked at 'not yet implemented', src/reviews.rs:18:5

Dans models.rs

#[derive(Serialize, Debug)]
pub struct Review {
    pub product_id: i32,
    pub reviewer: String,
    pub review: String,
}

Dans lib.rs

mod models;

Dans reviews.rs

use models;

#[derive(Debug, Serialize)]
pub struct Product {
    pub id: i32,
    pub reviews: Vec<Review>,
}

#[derive(Debug, Serialize)]
pub struct Review {
    pub reviewer: String,
    pub text: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub rating: Option<Rating>,
}

#[derive(Debug, Serialize)]
pub struct Rating {
    pub stars: i32,
    pub color: Color,
}

#[derive(Debug, Serialize)]
#[serde(rename_all = "lowercase")]
pub enum Color {
    Blue,
    Yellow,
    Red,
}

pub fn rating_nb_to_rating(rating: i32) -> Rating {
    Rating {
        stars: rating,
        color: match rating {
            1 => Color::Red,
            2 | 3 => Color::Yellow,
            _ => Color::Blue,
        },
    }
}

pub fn reviews_with_ratings(
    reviews: Vec<models::Review>,
    ratings: HashMap<String, i32>,
) -> Vec<Review> {
    reviews
        .iter()
        .map(|review| {
            let reviewer = review.reviewer.clone();
            Review {
                rating: ratings.get(&reviewer).map(|&r| rating_nb_to_rating(r)),
                reviewer: reviewer,
                text: review.review.clone(),
            }
        })
        .collect()
}

let ratings = HashMap::new();
let reviews = vec![];

let build_response_from_ratings = move |ratings| {
    futures::future::result({
        // build the response
        let product = Product {
            id: product_id,
            reviews: reviews_with_ratings(reviews, ratings),
        };
        // return a 200 response with the reviews as json
        Ok(HttpResponse::Ok().json(product))
    })
};

build_response_from_ratings(ratings).responder()

#[derive(Debug, Deserialize, Serialize, Clone)]
pub struct NewReview {
    reviewer: String,
    text: String,
    rating: i32,
}

let review_to_save = models::Review {
    product_id: product_id as i32,
    reviewer: review.reviewer.clone(),
    review: review.text.clone(),
};

futures::future::result({
    Ok(HttpResponse::Ok().json(review.clone()))
}).responder()

Récupérer les reviews d'un product donne une liste vide

curl localhost:9080/reviews/0 -i
HTTP/1.1 200 OK
content-length: 21
content-type: application/json
date: Thu, 19 Apr 2018 21:59:21 GMT

{"id":0,"reviews":[]}

Poster une review la renvoie telle quelle

curl localhost:9080/reviews/0 -i -H 'Content-Type: application/json' -d '{"reviewer":"moi","rating":3,"text":"mon avis"}'
HTTP/1.1 200 OK
content-length: 47
content-type: application/json
date: Thu, 19 Apr 2018 22:00:57 GMT

{"reviewer":"moi","text":"mon avis","rating":3}

Poster un JSON invalide répond une erreur 400 et un message dans les logs

curl localhost:9080/reviews/0 -i -H 'Content-Type: application/json' -d '{"reviewer":"moi","rating":3}'
HTTP/1.1 400 Bad Request
content-length: 0
date: Thu, 19 Apr 2018 22:01:54 GMT

{"msg":"Error occured during request handling: Json deserialize error: missing field `text` at line 1 column 29","level":"WARN","ts":"2018-04-20T00:01:55.287447+02:00","logger":"app"}

Dans db.rs

use actix::prelude::*;

pub struct DbExecutor();

impl Actor for DbExecutor {
    type Context = SyncContext<Self>;
}

Définir le message `GetReviews`

use models;

#[derive(Debug)]
pub struct GetReviews {
    pub product_id: i32,
}

impl Message for GetReviews {
    type Result = Result<Vec<models::Review>, ()>;
}

impl Handler<GetReviews> for DbExecutor {
    type Result = Result<Vec<models::Review>, ()>;

    fn handle(&mut self, msg: GetReviews, _: &mut Self::Context) -> Self::Result {
        warn!("getting reviews for product {}", msg.product_id);

        Ok(vec![
            models::Review {
                product_id: msg.product_id,
                review: "great!".to_string(),
                reviewer: "user1".to_string(),
            },
        ])
    }
}

Définir le message `SaveReview`

#[derive(Debug)]
pub struct SaveReview {
    pub review: models::Review,
}

impl Message for SaveReview {
    type Result = Result<models::Review, ()>;
}

impl Handler<SaveReview> for DbExecutor {
    type Result = Result<models::Review, ()>;

    fn handle(&mut self, msg: SaveReview, _: &mut Self::Context) -> Self::Result {
        warn!("saving review {:?}", msg.review);

        Ok(msg.review)
    }
}

Dans lib.rs

use actix::prelude::*;

mod db;

pub struct AppState {
    db: Addr<Syn, db::DbExecutor>,
}

let db_addr = SyncArbiter::start(3, move || db::DbExecutor());

server::new(move || {
    App::with_state(AppState {
        db: db_addr.clone(),
    }).middleware(Logger::default())

Dans reviews.rs

use db;

Enlever

let reviews = vec![];

Et remplacer

futures::future::result({

par:

state
    .db
    .send(db::GetReviews {
        product_id: product_id,
    })
    .map(|result| match result {
        Ok(reviews) => reviews,
        Err(err) => {
            error!("{:?}", err);
            vec![]
        }
    })
    .from_err()
    .and_then(move |reviews| {

Remplacer

futures::future::result({ Ok(HttpResponse::Ok().json(review.clone())) }).responder()

par:

state
    .db
    .send(db::SaveReview {
        review: review_to_save,
    })
    .from_err()
    .and_then(move |_| { Ok(HttpResponse::Ok().json(review.clone())) }).responder()

Notre avis s'affiche !

curl localhost:9080/reviews/0 -i
HTTP/1.1 200 OK
content-length: 57
content-type: application/json
date: Thu, 19 Apr 2018 22:13:43 GMT

{"id":0,"reviews":[{"reviewer":"user1","text":"great!"}]}

À la sauvegarde d'un avis, un log s'affiche:

curl localhost:9080/reviews/0 -i -H 'Content-Type: application/json' -d '{"reviewer":"moi","rating":3,"text":"mon avis"}'
HTTP/1.1 200 OK
content-length: 47
content-type: application/json
date: Thu, 19 Apr 2018 22:23:06 GMT

{"reviewer":"moi","text":"mon avis","rating":3}

{"msg":"saving review Review { product_id: 0, reviewer: \"moi\", review: \"mon avis\" }","level":"WARN","ts":"2018-04-20T00:23:06.973686+02:00","logger":"app"}

use std::env;

#[derive(Debug)]
pub struct Config {
    pub host: String,
    pub port: String,
    pub database_url: String,
    pub ratings_url: String,
}

impl Config {
    pub fn new() -> Self {
        Self {
            host: env::var("HOST").unwrap_or_else(|_| "0.0.0.0".to_string()),
            port: env::var("PORT").unwrap_or_else(|_| "9081".to_string()),
            database_url: env::var("DATABASE_URL").unwrap_or_else(|_| "db.sqlite".to_string()),
            ratings_url: env::var("RATINGS_URL")
                .unwrap_or_else(|_| "http://ratings:9080".to_string()),
        }
    }
}

Dans bin.rs

mod config;

Remplacer l'assignation existante d'addr par:

let config = config::Config::new();
let addr = format!("{}:{}", config.host, config.port);

Dans lib.rs

#[macro_use]
extern crate lazy_static;

mod config;

lazy_static! {
    static ref CONFIG: config::Config = config::Config::new();
}

Notre service écoute maintenant sur le port 9081

curl localhost:9081/reviews/0 -i
HTTP/1.1 200 OK
content-length: 57
content-type: application/json
date: Thu, 19 Apr 2018 22:27:26 GMT

{"id":0,"reviews":[{"reviewer":"user1","text":"great!"}]}

Dans lib.rs

#[macro_use]
extern crate diesel;
extern crate r2d2;

use diesel::prelude::*;
use diesel::r2d2::ConnectionManager;

Dans lib.rs

mod schema;

Dans model.rs

use super::schema::reviews;

#[derive(Serialize, Debug, Queryable, Insertable)]
#[table_name = "reviews"]
pub struct Review {
    pub product_id: i32,
    pub reviewer: String,
    pub review: String,
}

Dans lib.rs, à la place de l'assignation à db_addr:

let manager = ConnectionManager::<SqliteConnection>::new(CONFIG.database_url.clone());
let pool = r2d2::Pool::builder().build(manager).expect("Failed to create pool.");
let db_addr = SyncArbiter::start(3, move || db::DbExecutor(pool.clone()));

use diesel;
use diesel::prelude::*;
use diesel::r2d2::{ConnectionManager, Pool};

use schema;

Changer la définition de DbExecutor par:

pub struct DbExecutor(pub Pool<ConnectionManager<SqliteConnection>>);

Changer le message GetReviews par:

#[derive(Debug)]
pub struct GetReviews {
    pub product_id: i32,
}

impl Message for GetReviews {
    type Result = Result<Vec<models::Review>, diesel::result::Error>;
}

impl Handler<GetReviews> for DbExecutor {
    type Result = Result<Vec<models::Review>, diesel::result::Error>;

    fn handle(&mut self, msg: GetReviews, _: &mut Self::Context) -> Self::Result {
        warn!("getting reviews for product {}", msg.product_id);

        use self::schema::reviews::dsl::*;

        let conn: &SqliteConnection = &self.0.get().unwrap();

        let items = reviews
            .filter(product_id.eq(msg.product_id))
            .load::<models::Review>(conn)?;

        Ok(items)
    }
}

Changer le message SaveReview par:

#[derive(Debug)]
pub struct SaveReview {
    pub review: models::Review,
}

impl Message for SaveReview {
    type Result = Result<models::Review, diesel::result::Error>;
}

impl Handler<SaveReview> for DbExecutor {
    type Result = Result<models::Review, diesel::result::Error>;

    fn handle(&mut self, msg: SaveReview, _: &mut Self::Context) -> Self::Result {
        warn!("saving review {:?}", msg.review);

        use self::schema::reviews::dsl::*;

        let conn: &SqliteConnection = &self.0.get().unwrap();

        diesel::insert_into(reviews)
            .values(&msg.review)
            .execute(conn)?;

        Ok(msg.review)
    }
}

On récupère les reviews de la base de données:

curl localhost:9081/reviews/0
{
   "reviews" : [
      {
         "text" : "An extremely entertaining play by Shakespeare. The slapstick humour is refreshing!",
         "reviewer" : "Reviewer1"
      },
      {
         "reviewer" : "Reviewer2",
         "text" : "Absolutely fun and entertaining. The play lacks thematic depth when compared to other plays by Shakespeare."
      }
   ],
   "id" : 0
}

Les avis envoyés sont sauvegardés:

curl localhost:9081/reviews/1 -H 'Content-Type: application/json' -d '{"reviewer":"moi","rating":3,"text":"mon avis"}'
{"reviewer":"moi","text":"mon avis","rating":3}

curl localhost:9081/reviews/1
{
   "id" : 1,
   "reviews" : [
      {
         "text" : "mon avis",
         "reviewer" : "moi"
      }
   ]
}

Dans reviews.rs

use actix_web::client::ClientRequest;
use actix_web::HttpMessage;

#[derive(Debug, Deserialize)]
pub struct RatingsResponse {
    id: i32,
    pub ratings: HashMap<String, i32>,
}

Remplacer l'assignation de ratings par:

let ratings = ClientRequest::get(&format!("{}/ratings/{}", ::CONFIG.ratings_url, product_id))
    .finish()
    .unwrap()
    .send()
    .map_err(error::Error::from)
    .and_then(move |resp| {
        resp.json()
            .from_err()
            .and_then(|ratings: RatingsResponse| Ok(ratings.ratings))
    })
    .or_else(|err| {
        // in case of error, log it and continue with an empty list of ratings
        error!("{:?}", err);
        Ok(HashMap::new())
    });

Et remplacer

build_response_from_ratings(ratings).responder()

par

ratings.and_then(build_response_from_ratings).responder()

Remplacer

state
    .db
    .send(db::SaveReview {
        review: review_to_save,
    })
    .from_err()
    .and_then(move |_| Ok(HttpResponse::Ok().json(review.clone())))
    .responder()

par:

ClientRequest::post(&format!("{}/ratings/{}", ::CONFIG.ratings_url, product_id))
    .json(review.clone())
    .unwrap()
    .send()
    .map(|_| ())
    .or_else(|err| {
        // in case of error, log it and ignore it
        error!("{:?}", err);
        Ok(())
    })
    .and_then(move |_| {
        state
            .db
            .send(db::SaveReview {
                review: review_to_save,
            })
            .from_err()
            .and_then(move |_| Ok(HttpResponse::Ok().json(review.clone())))
    })
    .responder()

Les ratings sont maintenant dispo quand on demande des avis, et l'application ratings a loggé qu'elle a été appellée

curl localhost:9081/reviews/0
{
   "reviews" : [
      {
         "rating" : {
            "stars" : 5,
            "color" : "blue"
         },
         "text" : "An extremely entertaining play by Shakespeare. The slapstick humour is refreshing!",
         "reviewer" : "Reviewer1"
      },
      {
         "reviewer" : "Reviewer2",
         "rating" : {
            "color" : "blue",
            "stars" : 4
         },
         "text" : "Absolutely fun and entertaining. The play lacks thematic depth when compared to other plays by Shakespeare."
      }
   ],
   "id" : 0
}

{"msg":"requesting ratings for product 0","level":"INFO","ts":"2018-04-20T00:52:35.348631+02:00","logger":"app"}
{"msg":"[RESPONSE][d65d721f-b0d4-4689-a1fd-7b455fedd76d][HTTP/1.1][200 OK][755µs]","level":"INFO","ts":"2018-04-20T00:52:35.348900+02:00","logger":"app"}

À la sauvegarde d'un avis, la note est sauvegardée dans l'application ratings

curl localhost:9081/reviews/3 -H 'Content-Type: application/json' -d '{"reviewer":"moi","rating":3,"text":"mon avis"}'
{"reviewer":"moi","text":"mon avis","rating":3}

curl localhost:9081/reviews/3
{
   "reviews" : [
      {
         "reviewer" : "moi",
         "text" : "mon avis",
         "rating" : {
            "color" : "yellow",
            "stars" : 3
         }
      }
   ],
   "id" : 3
}

{"msg":"saving new rating NewRating { reviewer: \"moi\", rating: 3 } for product 3","level":"INFO","ts":"2018-04-20T01:00:22.008725+02:00","logger":"app"}
{"msg":"[RESPONSE][4c0951b3-f302-4759-9458-506a6b1a9f97][HTTP/1.1][200 OK][822µs]","level":"INFO","ts":"2018-04-20T01:00:22.009032+02:00","logger":"app"}

Dans lib.rs

extern crate time;

Dans health.rs

use actix_web::{HttpRequest, HttpResponse};

use time;

#[derive(Serialize)]
pub struct Healthcheck {
    now: i64,
    version: &'static str,
    status: &'static str,
}

pub fn healthcheck(_: HttpRequest<super::AppState>) -> HttpResponse {
    HttpResponse::Ok().json(Healthcheck {
        now: time::now_utc().to_timespec().sec,
        version: env!("CARGO_PKG_VERSION"),
        status: "Reviews is healthy",
    })
}

Dans lib.rs

mod health;

    .resource("/health", |r| {
        r.method(http::Method::GET).f(health::healthcheck)
    })

curl localhost:9081/health
{
   "status" : "Reviews is healthy",
   "now" : 1524179124,
   "version" : "0.1.0"
}

Cook a Microservice

Préparation

Avoir le hands on en local

Lancer le service ratings

Lancer le service avec rebuild automatique

Serveur HTTP

Démarrer une application et un système d'acteur

Résultat

Router les requêtes

Définir les méthodes retournant une réponse

Définir comment extraire le product_id du path

GET des reviews

POST d'une review

Définir les routes

Résultat

Renvoyer des données

Définir le modèle

Traduire avec les entrées / sorties

Définir la structure JSON de sortie

Convertir le modèle en sortie

Renvoyer la réponse

Définir la structure JSON en entrée

Convertir l'entrée en modèle

Renvoyer la réponse

Résultat

Acteur DB

Définir l'acteur

Définir le message GetReviews

Définir le message SaveReview

Ajouter l'acteur à l'AppState

Appeller l'acteur pendant les requêtes

Get des reviews

Save d'une review

Résultat

Configuration

module config.rs

Ecouter sur l'adresse de la configuration

Rendre la configuration accessible

Résultat

Utilisation de diesel

Nouvelles dépendances

Migrations et schémas

Ajout d'attributs au model

Configurer le pool de connection

Acteur db db.rs

Importer les nouvelles dépendances

Ajouter le pool de connection

Récupérer les reviews

Sauvegarder une review

Résultat

Communiquer avec le service ratings

Nouveaux imports

Récupérer les ratings

Définir la structure du json

Appeler le service

Sauvegarder un nouveau rating

Appeler le service

Résultat

Ajouter un healthcheck

Nouvelle dépendance

Module health

Ajouter les imports

Déclarer les données retournées

Retourner les données

Ajouter la route pour /GET healthcheck

Résultat

Définir le message `GetReviews`

Définir le message `SaveReview`

Ajouter l'acteur à l'`AppState`

module `config.rs`

Acteur db `db.rs`