From 7ccde8c2d32156d73599bc073b78a3d37d083f68 Mon Sep 17 00:00:00 2001 From: dowwie Date: Sat, 11 Aug 2018 06:58:29 -0400 Subject: [PATCH 1/2] added actix_redis example --- .travis.yml | 1 + Cargo.toml | 1 + actix_redis/Cargo.toml | 14 +++++++ actix_redis/README.md | 14 +++++++ actix_redis/src/main.rs | 86 +++++++++++++++++++++++++++++++++++++++++ 5 files changed, 116 insertions(+) create mode 100644 actix_redis/Cargo.toml create mode 100644 actix_redis/README.md create mode 100644 actix_redis/src/main.rs diff --git a/.travis.yml b/.travis.yml index 9bd4f006..6c4b00a4 100644 --- a/.travis.yml +++ b/.travis.yml @@ -36,6 +36,7 @@ script: - | cd async_db && cargo check && cd .. cd async_ex1 && cargo check && cd .. + cd actix_redis && cargo check && cd .. cd basics && cargo check && cd .. cd cookie-auth && cargo check && cd .. cd cookie-auth-full && cargo check && cd .. diff --git a/Cargo.toml b/Cargo.toml index 17ed4eb5..b07ba0e6 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,7 @@ [workspace] members = [ "./", + "actix_redis", "async_db", "async_ex1", "basics", diff --git a/actix_redis/Cargo.toml b/actix_redis/Cargo.toml new file mode 100644 index 00000000..a716e624 --- /dev/null +++ b/actix_redis/Cargo.toml @@ -0,0 +1,14 @@ +[package] +name = "actix_redis" +version = "0.1.0" +authors = ["dowwie "] + +[dependencies] +actix = "0.7.3" +actix-web = "0.7.3" +actix-redis = "0.5.1" +futures = "0.1.23" +redis-async = "0.4.0" +serde = "1.0.71" +serde_derive = "1.0.71" +env_logger = "0.5.12" diff --git a/actix_redis/README.md b/actix_redis/README.md new file mode 100644 index 00000000..bd0c9879 --- /dev/null +++ b/actix_redis/README.md @@ -0,0 +1,14 @@ +This project illustrates how to send multiple cache requests to redis in bulk, asynchronously. +This asyncio approach resembles traditional redis pipelining. Details about how this +is so can be read at https://github.com/benashford/redis-async-rs/issues/19#issuecomment-412208018 + + + +To test the demo, POST a json object containing three strings to the /cache_stuff endpoint: + {"one": "first entry", + "two": "second entry", + "three": "third entry" } + + + +These three entries will cache to redis, keyed accordingly. diff --git a/actix_redis/src/main.rs b/actix_redis/src/main.rs new file mode 100644 index 00000000..9134666d --- /dev/null +++ b/actix_redis/src/main.rs @@ -0,0 +1,86 @@ +extern crate actix; +extern crate actix_redis; +extern crate actix_web; +extern crate env_logger; +extern crate futures; +#[macro_use] extern crate redis_async; +extern crate serde; +#[macro_use] extern crate serde_derive; + + +use std::sync::Arc; +use actix::prelude::*; +use actix_redis::{Command, RedisActor, Error as ARError}; +use actix_web::{middleware, server, App, HttpRequest, HttpResponse, Json, + AsyncResponder, http::Method, Error as AWError}; +use futures::future::{Future, join_all}; +use redis_async::resp::RespValue; + + +#[derive(Deserialize)] +pub struct CacheInfo { + one: String, + two: String, + three: String +} + + +fn cache_stuff((info, req): (Json, HttpRequest)) + -> impl Future { + let info = info.into_inner(); + let redis = req.state().redis_addr.clone(); + + let one = redis.send(Command(resp_array!["SET", "mydomain:one", info.one])); + let two = redis.send(Command(resp_array!["SET", "mydomain:two", info.two])); + let three = redis.send(Command(resp_array!["SET", "mydomain:three", info.three])); + + // Creates a future which represents a collection of the results of the futures + // given. The returned future will drive execution for all of its underlying futures, + // collecting the results into a destination `Vec` in the same order as they + // were provided. If any future returns an error then all other futures will be + // canceled and an error will be returned immediately. If all futures complete + // successfully, however, then the returned future will succeed with a `Vec` of + // all the successful results. + let info_set = join_all(vec![one, two, three].into_iter()); + + info_set + .map_err(AWError::from) + .and_then(|res: Vec>| + // successful operations return "OK", so confirm that all returned as so + if !res.iter().all(|res| match res { + Ok(RespValue::SimpleString(x)) if x=="OK" => true, + _ => false + }) { + Ok(HttpResponse::InternalServerError().finish()) + } else { + Ok(HttpResponse::Ok().body("successfully cached values")) + } + ) + .responder() +} + + +pub struct AppState { + pub redis_addr: Arc> +} + +fn main() { + ::std::env::set_var("RUST_LOG", "actix_web=info,actix_redis=info"); + env_logger::init(); + let sys = actix::System::new("actix_redis_ex"); + + server::new(|| { + let redis_addr = Arc::new(RedisActor::start("127.0.0.1:6379")); + let app_state = AppState{redis_addr}; + + App::with_state(app_state) + .middleware(middleware::Logger::default()) + .resource("/cache_stuff", |r| r.method(Method::POST) + .with_async(cache_stuff)) + }).bind("0.0.0.0:8080") + .unwrap() + .workers(1) + .start(); + + let _ = sys.run(); +} From aff38dc15bf598db821457910076ae130bca1030 Mon Sep 17 00:00:00 2001 From: dowwie Date: Sat, 11 Aug 2018 09:59:07 -0400 Subject: [PATCH 2/2] added delete operations --- actix_redis/README.md | 5 +++-- actix_redis/src/main.rs | 24 ++++++++++++++++++++++-- 2 files changed, 25 insertions(+), 4 deletions(-) diff --git a/actix_redis/README.md b/actix_redis/README.md index bd0c9879..da1e2bbf 100644 --- a/actix_redis/README.md +++ b/actix_redis/README.md @@ -4,11 +4,12 @@ is so can be read at https://github.com/benashford/redis-async-rs/issues/19#issu -To test the demo, POST a json object containing three strings to the /cache_stuff endpoint: +To test the demo, POST a json object containing three strings to the /stuff endpoint: {"one": "first entry", "two": "second entry", "three": "third entry" } - These three entries will cache to redis, keyed accordingly. + +to delete these, simply issue a DELETE http request to /stuff endpoint diff --git a/actix_redis/src/main.rs b/actix_redis/src/main.rs index 9134666d..d9ccfa8e 100644 --- a/actix_redis/src/main.rs +++ b/actix_redis/src/main.rs @@ -59,6 +59,22 @@ fn cache_stuff((info, req): (Json, HttpRequest)) .responder() } +fn del_stuff(req: HttpRequest) + -> impl Future { + let redis = req.state().redis_addr.clone(); + + redis.send(Command(resp_array!["DEL", "mydomain:one", "mydomain:two", "mydomain:three"])) + .map_err(AWError::from) + .and_then(|res: Result| + match &res { + Ok(RespValue::Integer(x)) if x==&3 => + Ok(HttpResponse::Ok().body("successfully deleted values")), + _ =>{println!("---->{:?}", res); + Ok(HttpResponse::InternalServerError().finish())} + }) + .responder() + +} pub struct AppState { pub redis_addr: Arc> @@ -75,8 +91,12 @@ fn main() { App::with_state(app_state) .middleware(middleware::Logger::default()) - .resource("/cache_stuff", |r| r.method(Method::POST) - .with_async(cache_stuff)) + .resource("/stuff", |r| { + r.method(Method::POST) + .with_async(cache_stuff); + r.method(Method::DELETE) + .with_async(del_stuff)}) + }).bind("0.0.0.0:8080") .unwrap() .workers(1)