mirror of
https://github.com/actix/examples
synced 2025-06-28 18:00:37 +02:00
added async_db project
This commit is contained in:
dowwie
136
async_db/src/db.rs
Normal file
136
async_db/src/db.rs
Normal file
@ -0,0 +1,136 @@
|
||||
use actix::prelude::*;
|
||||
use actix_web::*;
|
||||
use std::{time::Duration, thread::sleep};
|
||||
use failure::Error;
|
||||
use r2d2;
|
||||
use r2d2_sqlite;
|
||||
|
||||
|
||||
pub type Pool = r2d2::Pool<r2d2_sqlite::SqliteConnectionManager>;
|
||||
pub type Connection = r2d2::PooledConnection<r2d2_sqlite::SqliteConnectionManager>;
|
||||
|
||||
|
||||
pub struct DbExecutor(pub Pool);
|
||||
impl Actor for DbExecutor {
|
||||
type Context = SyncContext<Self>;
|
||||
}
|
||||
|
||||
|
||||
#[derive(Debug, Serialize, Deserialize)]
|
||||
pub enum WeatherAgg {
|
||||
AnnualAgg {year: i32, total: f64},
|
||||
MonthAgg {year: i32, month: i32, total: f64}
|
||||
}
|
||||
|
||||
pub enum Queries {
|
||||
GetTopTenHottestYears,
|
||||
GetTopTenColdestYears,
|
||||
GetTopTenHottestMonths,
|
||||
GetTopTenColdestMonths
|
||||
}
|
||||
|
||||
//pub struct GetTopTenHottestYears;
|
||||
impl Message for Queries {
|
||||
type Result = Result<Vec<WeatherAgg>, Error>;
|
||||
}
|
||||
impl Handler<Queries> for DbExecutor {
|
||||
type Result = Result<Vec<WeatherAgg>, Error>;
|
||||
|
||||
fn handle(&mut self, msg: Queries, _: &mut Self::Context) -> Self::Result {
|
||||
let conn: Connection = self.0.get()?;
|
||||
|
||||
match msg {
|
||||
Queries::GetTopTenHottestYears => get_hottest_years(conn),
|
||||
Queries::GetTopTenColdestYears => get_coldest_years(conn),
|
||||
Queries::GetTopTenHottestMonths =>get_hottest_months(conn),
|
||||
Queries::GetTopTenColdestMonths => get_coldest_months(conn),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn get_hottest_years(conn: Connection) -> Result<Vec<WeatherAgg>, Error> {
|
||||
let stmt = "
|
||||
SELECT cast(strftime('%Y', date) as int) as theyear,
|
||||
sum(tmax) as total
|
||||
FROM nyc_weather
|
||||
WHERE tmax <> 'TMAX'
|
||||
GROUP BY theyear
|
||||
ORDER BY total DESC LIMIT 10;";
|
||||
|
||||
|
||||
let mut prep_stmt = conn.prepare(stmt)?;
|
||||
let annuals = prep_stmt.query_map(&[], |row| {
|
||||
WeatherAgg::AnnualAgg{year: row.get(0),
|
||||
total: row.get(1)}})
|
||||
.and_then(|mapped_rows|
|
||||
Ok(mapped_rows.map(|row| row.unwrap()).collect::<Vec<WeatherAgg>>()))?;
|
||||
|
||||
sleep(Duration::from_secs(2));
|
||||
|
||||
Ok(annuals)
|
||||
}
|
||||
|
||||
fn get_coldest_years(conn: Connection) -> Result<Vec<WeatherAgg>, Error> {
|
||||
let stmt = "
|
||||
SELECT cast(strftime('%Y', date) as int) as theyear,
|
||||
sum(tmax) as total
|
||||
FROM nyc_weather
|
||||
WHERE tmax <> 'TMAX'
|
||||
GROUP BY theyear
|
||||
ORDER BY total ASC LIMIT 10;";
|
||||
|
||||
let mut prep_stmt = conn.prepare(stmt)?;
|
||||
let annuals = prep_stmt.query_map(&[], |row| {
|
||||
WeatherAgg::AnnualAgg{year: row.get(0),
|
||||
total: row.get(1)}})
|
||||
.and_then(|mapped_rows|
|
||||
Ok(mapped_rows.map(|row| row.unwrap()).collect::<Vec<WeatherAgg>>()))?;
|
||||
|
||||
sleep(Duration::from_secs(2));
|
||||
|
||||
Ok(annuals)
|
||||
}
|
||||
|
||||
fn get_hottest_months(conn: Connection) -> Result<Vec<WeatherAgg>, Error> {
|
||||
let stmt =
|
||||
"SELECT cast(strftime('%Y', date) as int) as theyear,
|
||||
cast(strftime('%m', date) as int) as themonth,
|
||||
sum(tmax) as total
|
||||
FROM nyc_weather
|
||||
WHERE tmax <> 'TMAX'
|
||||
GROUP BY theyear, themonth
|
||||
ORDER BY total DESC LIMIT 10;";
|
||||
|
||||
let mut prep_stmt = conn.prepare(stmt)?;
|
||||
let annuals = prep_stmt.query_map(&[], |row| {
|
||||
WeatherAgg::MonthAgg{year: row.get(0),
|
||||
month: row.get(1),
|
||||
total: row.get(2)}})
|
||||
.and_then(|mapped_rows|
|
||||
Ok(mapped_rows.map(|row| row.unwrap()).collect::<Vec<WeatherAgg>>()))?;
|
||||
|
||||
sleep(Duration::from_secs(2));
|
||||
Ok(annuals)
|
||||
}
|
||||
|
||||
fn get_coldest_months(conn: Connection) -> Result<Vec<WeatherAgg>, Error> {
|
||||
let stmt =
|
||||
"SELECT cast(strftime('%Y', date) as int) as theyear,
|
||||
cast(strftime('%m', date) as int) as themonth,
|
||||
sum(tmax) as total
|
||||
FROM nyc_weather
|
||||
WHERE tmax <> 'TMAX'
|
||||
GROUP BY theyear, themonth
|
||||
ORDER BY total ASC LIMIT 10;";
|
||||
|
||||
let mut prep_stmt = conn.prepare(stmt)?;
|
||||
let annuals = prep_stmt.query_map(&[], |row| {
|
||||
WeatherAgg::MonthAgg{year: row.get(0),
|
||||
month: row.get(1),
|
||||
total: row.get(2)}})
|
||||
.and_then(|mapped_rows|
|
||||
Ok(mapped_rows.map(|row| row.unwrap()).collect::<Vec<WeatherAgg>>()))?;
|
||||
|
||||
sleep(Duration::from_secs(2));
|
||||
Ok(annuals)
|
||||
}
|
||||
118
async_db/src/main.rs
Normal file
118
async_db/src/main.rs
Normal file
@ -0,0 +1,118 @@
|
||||
/* Actix-Web Asynchronous Database Example
|
||||
|
||||
This project illustrates two examples:
|
||||
|
||||
1. An asynchronous handler that executes 4 queries in *sequential order*,
|
||||
collecting the results and returning them as a single serialized json object
|
||||
|
||||
2. An asynchronous handler that executes 4 queries in *parallel*,
|
||||
collecting the results and returning them as a single serialized json object
|
||||
|
||||
*/
|
||||
|
||||
#[macro_use] extern crate actix;
|
||||
extern crate actix_web;
|
||||
extern crate env_logger;
|
||||
extern crate failure;
|
||||
extern crate futures;
|
||||
extern crate num_cpus;
|
||||
extern crate r2d2;
|
||||
extern crate r2d2_sqlite;
|
||||
extern crate serde;
|
||||
#[macro_use] extern crate serde_derive;
|
||||
extern crate serde_json;
|
||||
|
||||
use actix::prelude::*;
|
||||
use actix_web::{
|
||||
http, middleware, server, App, AsyncResponder, FutureResponse, HttpResponse,
|
||||
State, Error as AWError
|
||||
};
|
||||
use std::error::Error as StdError;
|
||||
use failure::Error;
|
||||
use futures::future::{Future, join_all, ok as fut_ok, err as fut_err};
|
||||
use r2d2_sqlite::SqliteConnectionManager;
|
||||
|
||||
mod db;
|
||||
use db::{DbExecutor, Queries, WeatherAgg, Pool};
|
||||
|
||||
/// State with DbExecutor address
|
||||
struct AppState {
|
||||
db: Addr<Syn, DbExecutor>,
|
||||
}
|
||||
|
||||
|
||||
/// Version 1: Calls 4 queries in sequential order, as an asynchronous handler
|
||||
fn asyncio_weather(state: State<AppState>) -> FutureResponse<HttpResponse> {
|
||||
let mut result: Vec<Vec<WeatherAgg>> = vec![];
|
||||
|
||||
state.db.send(Queries::GetTopTenHottestYears).from_err()
|
||||
.and_then(move |res| {
|
||||
result.push(res.unwrap());
|
||||
state.db.send(Queries::GetTopTenColdestYears).from_err()
|
||||
.and_then(move |res| {
|
||||
result.push(res.unwrap());
|
||||
state.db.send(Queries::GetTopTenHottestMonths).from_err()
|
||||
.and_then(move |res| {
|
||||
result.push(res.unwrap());
|
||||
state.db.send(Queries::GetTopTenColdestMonths).from_err()
|
||||
.and_then(move |res| {
|
||||
result.push(res.unwrap());
|
||||
fut_ok(result)
|
||||
})
|
||||
})
|
||||
})
|
||||
})
|
||||
.and_then(|res| Ok(HttpResponse::Ok().json(res)))
|
||||
.responder()
|
||||
}
|
||||
|
||||
/// Version 2: Calls 4 queries in parallel, as an asynchronous handler
|
||||
/// Returning Error types turn into None values in the response
|
||||
fn parallel_weather(state: State<AppState>) -> FutureResponse<HttpResponse> {
|
||||
let fut_result = vec![
|
||||
Box::new(state.db.send(Queries::GetTopTenHottestYears)),
|
||||
Box::new(state.db.send(Queries::GetTopTenColdestYears)),
|
||||
Box::new(state.db.send(Queries::GetTopTenHottestMonths)),
|
||||
Box::new(state.db.send(Queries::GetTopTenColdestMonths))];
|
||||
|
||||
join_all(fut_result)
|
||||
.map_err(AWError::from)
|
||||
.and_then(|result| {
|
||||
let res: Vec<Option<Vec<WeatherAgg>>> =
|
||||
result.into_iter().map(|x| x.ok()).collect();
|
||||
|
||||
Ok(HttpResponse::Ok().json(res))
|
||||
})
|
||||
.responder()
|
||||
}
|
||||
|
||||
fn main() {
|
||||
::std::env::set_var("RUST_LOG", "actix_web=info");
|
||||
env_logger::init();
|
||||
let sys = actix::System::new("parallel_db_example");
|
||||
|
||||
// Start N db executor actors (N = number of cores avail)
|
||||
|
||||
let manager = SqliteConnectionManager::file("weather.db");
|
||||
let pool = Pool::new(manager).unwrap();
|
||||
|
||||
let addr = SyncArbiter::start(num_cpus::get(), move || DbExecutor(pool.clone()));
|
||||
|
||||
// Start http server
|
||||
server::new(move || {
|
||||
App::with_state(AppState{db: addr.clone()})
|
||||
// enable logger
|
||||
.middleware(middleware::Logger::default())
|
||||
.resource("/asyncio_weather", |r|
|
||||
r.method(http::Method::GET)
|
||||
.with(asyncio_weather))
|
||||
.resource("/parallel_weather", |r|
|
||||
r.method(http::Method::GET)
|
||||
.with(parallel_weather))
|
||||
}).bind("127.0.0.1:8080")
|
||||
.unwrap()
|
||||
.start();
|
||||
|
||||
println!("Started http server: 127.0.0.1:8080");
|
||||
let _ = sys.run();
|
||||
}
|
||||
Reference in New Issue
Block a user