Add actix-todo example

actix_todo/src/api.rs

@@ -0,0 +1,174 @@
+use actix::prelude::Addr;
+use actix_web::middleware::Response;
+use actix_web::{
+    error, fs::NamedFile, http, AsyncResponder, Form, FutureResponse, HttpRequest,
+    HttpResponse, Path, Responder, Result,
+};
+use futures::{future, Future};
+use tera::{Context, Tera};
+
+use db::{AllTasks, CreateTask, DbExecutor, DeleteTask, ToggleTask};
+use session::{self, FlashMessage};
+
+pub struct AppState {
+    pub template: Tera,
+    pub db: Addr<DbExecutor>,
+}
+
+pub fn index(req: HttpRequest<AppState>) -> FutureResponse<HttpResponse> {
+    req.state()
+        .db
+        .send(AllTasks)
+        .from_err()
+        .and_then(move |res| match res {
+            Ok(tasks) => {
+                let mut context = Context::new();
+                context.add("tasks", &tasks);
+
+                if let Some(flash) = session::get_flash(&req)? {
+                    context.add("msg", &(flash.kind, flash.message));
+                    session::clear_flash(&req);
+                }
+
+                let rendered = req.state()
+                    .template
+                    .render("index.html.tera", &context)
+                    .map_err(|e| {
+                        error::ErrorInternalServerError(e.description().to_owned())
+                    })?;
+
+                Ok(HttpResponse::Ok().body(rendered))
+            }
+            Err(e) => Err(e),
+        })
+        .responder()
+}
+
+#[derive(Deserialize)]
+pub struct CreateForm {
+    description: String,
+}
+
+pub fn create(
+    (req, params): (HttpRequest<AppState>, Form<CreateForm>),
+) -> FutureResponse<HttpResponse> {
+    if params.description.is_empty() {
+        future::lazy(move || {
+            session::set_flash(
+                &req,
+                FlashMessage::error("Description cannot be empty"),
+            )?;
+            Ok(redirect_to("/"))
+        }).responder()
+    } else {
+        req.state()
+            .db
+            .send(CreateTask {
+                description: params.description.clone(),
+            })
+            .from_err()
+            .and_then(move |res| match res {
+                Ok(_) => {
+                    session::set_flash(
+                        &req,
+                        FlashMessage::success("Task successfully added"),
+                    )?;
+                    Ok(redirect_to("/"))
+                }
+                Err(e) => Err(e),
+            })
+            .responder()
+    }
+}
+
+#[derive(Deserialize)]
+pub struct UpdateParams {
+    id: i32,
+}
+
+#[derive(Deserialize)]
+pub struct UpdateForm {
+    _method: String,
+}
+
+pub fn update(
+    (req, params, form): (HttpRequest<AppState>, Path<UpdateParams>, Form<UpdateForm>),
+) -> FutureResponse<HttpResponse> {
+    match form._method.as_ref() {
+        "put" => toggle(req, params),
+        "delete" => delete(req, params),
+        unsupported_method => {
+            let msg = format!("Unsupported HTTP method: {}", unsupported_method);
+            future::err(error::ErrorBadRequest(msg)).responder()
+        }
+    }
+}
+
+fn toggle(
+    req: HttpRequest<AppState>,
+    params: Path<UpdateParams>,
+) -> FutureResponse<HttpResponse> {
+    req.state()
+        .db
+        .send(ToggleTask { id: params.id })
+        .from_err()
+        .and_then(move |res| match res {
+            Ok(_) => Ok(redirect_to("/")),
+            Err(e) => Err(e),
+        })
+        .responder()
+}
+
+fn delete(
+    req: HttpRequest<AppState>,
+    params: Path<UpdateParams>,
+) -> FutureResponse<HttpResponse> {
+    req.state()
+        .db
+        .send(DeleteTask { id: params.id })
+        .from_err()
+        .and_then(move |res| match res {
+            Ok(_) => {
+                session::set_flash(&req, FlashMessage::success("Task was deleted."))?;
+                Ok(redirect_to("/"))
+            }
+            Err(e) => Err(e),
+        })
+        .responder()
+}
+
+fn redirect_to(location: &str) -> HttpResponse {
+    HttpResponse::Found()
+        .header(http::header::LOCATION, location)
+        .finish()
+}
+
+pub fn bad_request<S: 'static>(
+    req: &HttpRequest<S>,
+    resp: HttpResponse,
+) -> Result<Response> {
+    let new_resp = NamedFile::open("static/errors/400.html")?
+        .set_status_code(resp.status())
+        .respond_to(req)?;
+    Ok(Response::Done(new_resp))
+}
+
+pub fn not_found<S: 'static>(
+    req: &HttpRequest<S>,
+    resp: HttpResponse,
+) -> Result<Response> {
+    let new_resp = NamedFile::open("static/errors/404.html")?
+        .set_status_code(resp.status())
+        .respond_to(req)?;
+    Ok(Response::Done(new_resp))
+}
+
+pub fn internal_server_error<S: 'static>(
+    req: &HttpRequest<S>,
+    resp: HttpResponse,
+) -> Result<Response> {
+    let new_resp = NamedFile::open("static/errors/500.html")?
+        .set_status_code(resp.status())
+        .respond_to(req)?;
+    Ok(Response::Done(new_resp))
+}

actix_todo/src/db.rs

@@ -0,0 +1,100 @@
+use std::ops::Deref;
+
+use actix::prelude::{Actor, Handler, Message, SyncContext};
+use actix_web::{error, Error};
+use diesel::pg::PgConnection;
+use diesel::r2d2::{ConnectionManager, Pool, PoolError, PooledConnection};
+
+use model::{NewTask, Task};
+
+type PgPool = Pool<ConnectionManager<PgConnection>>;
+type PgPooledConnection = PooledConnection<ConnectionManager<PgConnection>>;
+
+pub fn init_pool(database_url: &str) -> Result<PgPool, PoolError> {
+    let manager = ConnectionManager::<PgConnection>::new(database_url);
+    Pool::builder().build(manager)
+}
+
+pub struct DbExecutor(pub PgPool);
+
+impl DbExecutor {
+    pub fn get_conn(&self) -> Result<PgPooledConnection, Error> {
+        self.0.get().map_err(|e| error::ErrorInternalServerError(e))
+    }
+}
+
+impl Actor for DbExecutor {
+    type Context = SyncContext<Self>;
+}
+
+pub struct AllTasks;
+
+impl Message for AllTasks {
+    type Result = Result<Vec<Task>, Error>;
+}
+
+impl Handler<AllTasks> for DbExecutor {
+    type Result = Result<Vec<Task>, Error>;
+
+    fn handle(&mut self, _: AllTasks, _: &mut Self::Context) -> Self::Result {
+        Task::all(self.get_conn()?.deref())
+            .map_err(|_| error::ErrorInternalServerError("Error inserting task"))
+    }
+}
+
+pub struct CreateTask {
+    pub description: String,
+}
+
+impl Message for CreateTask {
+    type Result = Result<(), Error>;
+}
+
+impl Handler<CreateTask> for DbExecutor {
+    type Result = Result<(), Error>;
+
+    fn handle(&mut self, todo: CreateTask, _: &mut Self::Context) -> Self::Result {
+        let new_task = NewTask {
+            description: todo.description,
+        };
+        Task::insert(new_task, self.get_conn()?.deref())
+            .map(|_| ())
+            .map_err(|_| error::ErrorInternalServerError("Error inserting task"))
+    }
+}
+
+pub struct ToggleTask {
+    pub id: i32,
+}
+
+impl Message for ToggleTask {
+    type Result = Result<(), Error>;
+}
+
+impl Handler<ToggleTask> for DbExecutor {
+    type Result = Result<(), Error>;
+
+    fn handle(&mut self, task: ToggleTask, _: &mut Self::Context) -> Self::Result {
+        Task::toggle_with_id(task.id, self.get_conn()?.deref())
+            .map(|_| ())
+            .map_err(|_| error::ErrorInternalServerError("Error inserting task"))
+    }
+}
+
+pub struct DeleteTask {
+    pub id: i32,
+}
+
+impl Message for DeleteTask {
+    type Result = Result<(), Error>;
+}
+
+impl Handler<DeleteTask> for DbExecutor {
+    type Result = Result<(), Error>;
+
+    fn handle(&mut self, task: DeleteTask, _: &mut Self::Context) -> Self::Result {
+        Task::delete_with_id(task.id, self.get_conn()?.deref())
+            .map(|_| ())
+            .map_err(|_| error::ErrorInternalServerError("Error inserting task"))
+    }
+}

actix_todo/src/main.rs

@@ -0,0 +1,87 @@
+extern crate actix;
+extern crate actix_web;
+extern crate dotenv;
+extern crate env_logger;
+extern crate futures;
+#[macro_use]
+extern crate diesel;
+#[macro_use]
+extern crate log;
+#[macro_use]
+extern crate serde_derive;
+#[macro_use]
+extern crate tera;
+
+use actix::prelude::SyncArbiter;
+use actix_web::middleware::session::{CookieSessionBackend, SessionStorage};
+use actix_web::middleware::{ErrorHandlers, Logger};
+use actix_web::{dev::Resource, fs, http, server, App};
+use dotenv::dotenv;
+use std::env;
+use tera::Tera;
+
+mod api;
+mod db;
+mod model;
+mod schema;
+mod session;
+
+static SESSION_SIGNING_KEY: &[u8] = &[0; 32];
+const NUM_DB_THREADS: usize = 3;
+
+fn main() {
+    dotenv().ok();
+
+    std::env::set_var("RUST_LOG", "actix_todo=debug,actix_web=info");
+    env_logger::init();
+
+    // Start the Actix system
+    let system = actix::System::new("todo-app");
+
+    let database_url = env::var("DATABASE_URL").expect("DATABASE_URL must be set");
+    let pool = db::init_pool(&database_url).expect("Failed to create pool");
+    let addr = SyncArbiter::start(NUM_DB_THREADS, move || db::DbExecutor(pool.clone()));
+
+    let app = move || {
+        debug!("Constructing the App");
+
+        let templates: Tera = compile_templates!("templates/**/*");
+
+        let session_store = SessionStorage::new(
+            CookieSessionBackend::signed(SESSION_SIGNING_KEY).secure(false),
+        );
+
+        let error_handlers = ErrorHandlers::new()
+            .handler(
+                http::StatusCode::INTERNAL_SERVER_ERROR,
+                api::internal_server_error,
+            )
+            .handler(http::StatusCode::BAD_REQUEST, api::bad_request)
+            .handler(http::StatusCode::NOT_FOUND, api::not_found);
+
+        let static_files = fs::StaticFiles::new("static/")
+            .expect("failed constructing static files handler");
+
+        let state = api::AppState {
+            template: templates,
+            db: addr.clone(),
+        };
+
+        App::with_state(state)
+            .middleware(Logger::default())
+            .middleware(session_store)
+            .middleware(error_handlers)
+            .route("/", http::Method::GET, api::index)
+            .route("/todo", http::Method::POST, api::create)
+            .resource("/todo/{id}", |r: &mut Resource<_>| {
+                r.post().with(api::update)
+            })
+            .handler("/static", static_files)
+    };
+
+    debug!("Starting server");
+    server::new(app).bind("localhost:8088").unwrap().start();
+
+    // Run actix system, this method actually starts all async processes
+    let _ = system.run();
+}

actix_todo/src/model.rs

@@ -0,0 +1,46 @@
+use diesel;
+use diesel::pg::PgConnection;
+use diesel::prelude::*;
+
+use schema::{
+    tasks, tasks::dsl::{completed as task_completed, tasks as all_tasks},
+};
+
+#[derive(Debug, Insertable)]
+#[table_name = "tasks"]
+pub struct NewTask {
+    pub description: String,
+}
+
+#[derive(Debug, Queryable, Serialize)]
+pub struct Task {
+    pub id: i32,
+    pub description: String,
+    pub completed: bool,
+}
+
+impl Task {
+    pub fn all(conn: &PgConnection) -> QueryResult<Vec<Task>> {
+        all_tasks.order(tasks::id.desc()).load::<Task>(conn)
+    }
+
+    pub fn insert(todo: NewTask, conn: &PgConnection) -> QueryResult<usize> {
+        diesel::insert_into(tasks::table)
+            .values(&todo)
+            .execute(conn)
+    }
+
+    pub fn toggle_with_id(id: i32, conn: &PgConnection) -> QueryResult<usize> {
+        let task = all_tasks.find(id).get_result::<Task>(conn)?;
+
+        let new_status = !task.completed;
+        let updated_task = diesel::update(all_tasks.find(id));
+        updated_task
+            .set(task_completed.eq(new_status))
+            .execute(conn)
+    }
+
+    pub fn delete_with_id(id: i32, conn: &PgConnection) -> QueryResult<usize> {
+        diesel::delete(all_tasks.find(id)).execute(conn)
+    }
+}

actix_todo/src/schema.rs

@@ -0,0 +1,7 @@
+table! {
+    tasks (id) {
+        id -> Int4,
+        description -> Varchar,
+        completed -> Bool,
+    }
+}

actix_todo/src/session.rs

@@ -0,0 +1,39 @@
+use actix_web::error::Result;
+use actix_web::middleware::session::RequestSession;
+use actix_web::HttpRequest;
+
+const FLASH_KEY: &str = "flash";
+
+pub fn set_flash<T>(request: &HttpRequest<T>, flash: FlashMessage) -> Result<()> {
+    request.session().set(FLASH_KEY, flash)
+}
+
+pub fn get_flash<T>(req: &HttpRequest<T>) -> Result<Option<FlashMessage>> {
+    req.session().get::<FlashMessage>(FLASH_KEY)
+}
+
+pub fn clear_flash<T>(req: &HttpRequest<T>) {
+    req.session().remove(FLASH_KEY);
+}
+
+#[derive(Deserialize, Serialize)]
+pub struct FlashMessage {
+    pub kind: String,
+    pub message: String,
+}
+
+impl FlashMessage {
+    pub fn success(message: &str) -> Self {
+        Self {
+            kind: "success".to_owned(),
+            message: message.to_owned(),
+        }
+    }
+
+    pub fn error(message: &str) -> Self {
+        Self {
+            kind: "error".to_owned(),
+            message: message.to_owned(),
+        }
+    }
+}