actix/examples
1
0
Fork 0
mirror of https://github.com/actix/examples synced 2025-02-17 15:23:31 +01:00
Code Issues Releases Wiki Activity

migrate readis and websockets examples

Browse Source
This commit is contained in:
Nikolay Kim 2019-12-16 11:23:36 +06:00
parent 22fd5f3869
commit ca3f11b59e
9 changed files with 163 additions and 158 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
Cargo.toml
View File

@ -1,6 +1,5 @@
[workspace] [workspace]
members = [ members = [
# "actix_redis",
"async_db", "async_db",
"async_ex1", "async_ex1",
"async_ex2", "async_ex2",
@ -22,7 +21,8 @@ members = [
"openssl", "openssl",
# "protobuf", # "protobuf",
"r2d2", "r2d2",
# "redis-session", "redis",
"redis-session",
"rustls", "rustls",
"server-sent-events", "server-sent-events",
"simple-auth-server", "simple-auth-server",
@ -36,7 +36,7 @@ members = [
# "udp-echo", # "udp-echo",
"unix-socket", "unix-socket",
# "web-cors/backend", # "web-cors/backend",
# "websocket", "websocket",
"websocket-chat", "websocket-chat",
# "websocket-chat-broker", # "websocket-chat-broker",
# "websocket-tcp-chat", # "websocket-tcp-chat",

15
redis-session/Cargo.toml
View File

@ -1,18 +1,17 @@
[package] [package]
name = "redis_session" name = "redis_session"
version = "0.1.0" version = "2.0.0"
authors = ["Nikolay Kim <fafhrd91@gmail.com>"] authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
workspace = ".."
edition = "2018" edition = "2018"
[dependencies] [dependencies]
actix-web = "1.0.3" actix-rt = "1.0.0"
actix-session = "0.2.0" actix-web = "2.0.0-alpha.6"
actix-redis = { version = "0.6.0", features = ["web"] } actix-session = "0.3.0-alpha.3"
actix-redis = { version = "0.8.0-alpha.1", features = ["web"] }
env_logger = "0.6" env_logger = "0.6"
serde = { version = "^1.0", features = ["derive"] } serde = { version = "^1.0", features = ["derive"] }
actix-service = "0.4.1" actix-service = "1.0.0"
actix-http-test = "0.2.2" actix-http = "1.0.0"
actix-http = "0.2.5"
serde_json = "1.0.40" serde_json = "1.0.40"
time = "0.1.42" time = "0.1.42"

84
redis-session/src/main.rs
View File

@ -19,7 +19,7 @@ pub struct IndexResponse {
counter: i32, counter: i32,
} }
fn index(session: Session) -> Result<HttpResponse> { async fn index(session: Session) -> Result<HttpResponse> {
let user_id: Option<String> = session.get::<String>("user_id").unwrap(); let user_id: Option<String> = session.get::<String>("user_id").unwrap();
let counter: i32 = session let counter: i32 = session
.get::<i32>("counter") .get::<i32>("counter")
@ -29,7 +29,7 @@ fn index(session: Session) -> Result<HttpResponse> {
Ok(HttpResponse::Ok().json(IndexResponse { user_id, counter })) Ok(HttpResponse::Ok().json(IndexResponse { user_id, counter }))
} }
fn do_something(session: Session) -> Result<HttpResponse> { async fn do_something(session: Session) -> Result<HttpResponse> {
let user_id: Option<String> = session.get::<String>("user_id").unwrap(); let user_id: Option<String> = session.get::<String>("user_id").unwrap();
let counter: i32 = session let counter: i32 = session
.get::<i32>("counter") .get::<i32>("counter")
@ -44,7 +44,8 @@ fn do_something(session: Session) -> Result<HttpResponse> {
struct Identity { struct Identity {
user_id: String, user_id: String,
} }
fn login(user_id: web::Json<Identity>, session: Session) -> Result<HttpResponse> {
async fn login(user_id: web::Json<Identity>, session: Session) -> Result<HttpResponse> {
let id = user_id.into_inner().user_id; let id = user_id.into_inner().user_id;
session.set("user_id", &id)?; session.set("user_id", &id)?;
session.renew(); session.renew();
@ -60,7 +61,7 @@ fn login(user_id: web::Json<Identity>, session: Session) -> Result<HttpResponse>
})) }))
} }
fn logout(session: Session) -> Result<HttpResponse> { async fn logout(session: Session) -> Result<HttpResponse> {
let id: Option<String> = session.get("user_id")?; let id: Option<String> = session.get("user_id")?;
if let Some(x) = id { if let Some(x) = id {
session.purge(); session.purge();
@ -70,7 +71,8 @@ fn logout(session: Session) -> Result<HttpResponse> {
} }
} }
fn main() -> std::io::Result<()> { #[actix_rt::main]
async fn main() -> std::io::Result<()> {
std::env::set_var("RUST_LOG", "actix_web=info,actix_redis=info"); std::env::set_var("RUST_LOG", "actix_web=info,actix_redis=info");
env_logger::init(); env_logger::init();
@ -86,24 +88,24 @@ fn main() -> std::io::Result<()> {
.service(resource("/logout").route(post().to(logout))) .service(resource("/logout").route(post().to(logout)))
}) })
.bind("127.0.0.1:8080")? .bind("127.0.0.1:8080")?
.run() .start()
.await
} }
#[cfg(test)] #[cfg(test)]
mod test { mod test {
use super::*; use super::*;
use actix_http::{httpmessage::HttpMessage, HttpService}; use actix_http::httpmessage::HttpMessage;
use actix_http_test::{block_on, TestServer};
use actix_web::{ use actix_web::{
middleware, middleware, test,
web::{get, post, resource}, web::{get, post, resource},
App, App,
}; };
use serde_json::json; use serde_json::json;
use time; use time;
#[test] #[actix_rt::test]
fn test_workflow() { async fn test_workflow() {
// Step 1: GET index // Step 1: GET index
// - set-cookie actix-session will be in response (session cookie #1) // - set-cookie actix-session will be in response (session cookie #1)
// - response should be: {"counter": 0, "user_id": None} // - response should be: {"counter": 0, "user_id": None}
@ -134,26 +136,24 @@ mod test {
// - set-cookie actix-session will be in response (session cookie #3) // - set-cookie actix-session will be in response (session cookie #3)
// - response should be: {"counter": 0, "user_id": None} // - response should be: {"counter": 0, "user_id": None}
let mut srv = TestServer::new(|| { let srv = test::start(|| {
HttpService::new( App::new()
App::new() .wrap(
.wrap( RedisSession::new("127.0.0.1:6379", &[0; 32])
RedisSession::new("127.0.0.1:6379", &[0; 32]) .cookie_name("test-session"),
.cookie_name("test-session"), )
) .wrap(middleware::Logger::default())
.wrap(middleware::Logger::default()) .service(resource("/").route(get().to(index)))
.service(resource("/").route(get().to(index))) .service(resource("/do_something").route(post().to(do_something)))
.service(resource("/do_something").route(post().to(do_something))) .service(resource("/login").route(post().to(login)))
.service(resource("/login").route(post().to(login))) .service(resource("/logout").route(post().to(logout)))
.service(resource("/logout").route(post().to(logout))),
)
}); });
// Step 1: GET index // Step 1: GET index
// - set-cookie actix-session will be in response (session cookie #1) // - set-cookie actix-session will be in response (session cookie #1)
// - response should be: {"counter": 0, "user_id": None} // - response should be: {"counter": 0, "user_id": None}
let req_1a = srv.get("/").send(); let req_1a = srv.get("/").send();
let mut resp_1 = srv.block_on(req_1a).unwrap(); let mut resp_1 = req_1a.await.unwrap();
let cookie_1 = resp_1 let cookie_1 = resp_1
.cookies() .cookies()
.unwrap() .unwrap()
@ -161,7 +161,7 @@ mod test {
.into_iter() .into_iter()
.find(|c| c.name() == "test-session") .find(|c| c.name() == "test-session")
.unwrap(); .unwrap();
let result_1 = block_on(resp_1.json::<IndexResponse>()).unwrap(); let result_1 = resp_1.json::<IndexResponse>().await.unwrap();
assert_eq!( assert_eq!(
result_1, result_1,
IndexResponse { IndexResponse {
@ -174,7 +174,7 @@ mod test {
// - set-cookie will *not* be in response // - set-cookie will *not* be in response
// - response should be: {"counter": 0, "user_id": None} // - response should be: {"counter": 0, "user_id": None}
let req_2 = srv.get("/").cookie(cookie_1.clone()).send(); let req_2 = srv.get("/").cookie(cookie_1.clone()).send();
let resp_2 = srv.block_on(req_2).unwrap(); let resp_2 = req_2.await.unwrap();
let cookie_2 = resp_2 let cookie_2 = resp_2
.cookies() .cookies()
.unwrap() .unwrap()
@ -187,8 +187,8 @@ mod test {
// - adds new session state in redis: {"counter": 1} // - adds new session state in redis: {"counter": 1}
// - response should be: {"counter": 1, "user_id": None} // - response should be: {"counter": 1, "user_id": None}
let req_3 = srv.post("/do_something").cookie(cookie_1.clone()).send(); let req_3 = srv.post("/do_something").cookie(cookie_1.clone()).send();
let mut resp_3 = srv.block_on(req_3).unwrap(); let mut resp_3 = req_3.await.unwrap();
let result_3 = block_on(resp_3.json::<IndexResponse>()).unwrap(); let result_3 = resp_3.json::<IndexResponse>().await.unwrap();
assert_eq!( assert_eq!(
result_3, result_3,
IndexResponse { IndexResponse {
@ -201,8 +201,8 @@ mod test {
// - updates session state in redis: {"counter": 2} // - updates session state in redis: {"counter": 2}
// - response should be: {"counter": 2, "user_id": None} // - response should be: {"counter": 2, "user_id": None}
let req_4 = srv.post("/do_something").cookie(cookie_1.clone()).send(); let req_4 = srv.post("/do_something").cookie(cookie_1.clone()).send();
let mut resp_4 = srv.block_on(req_4).unwrap(); let mut resp_4 = req_4.await.unwrap();
let result_4 = block_on(resp_4.json::<IndexResponse>()).unwrap(); let result_4 = resp_4.json::<IndexResponse>().await.unwrap();
assert_eq!( assert_eq!(
result_4, result_4,
IndexResponse { IndexResponse {
@ -218,7 +218,7 @@ mod test {
.post("/login") .post("/login")
.cookie(cookie_1.clone()) .cookie(cookie_1.clone())
.send_json(&json!({"user_id": "ferris"})); .send_json(&json!({"user_id": "ferris"}));
let mut resp_5 = srv.block_on(req_5).unwrap(); let mut resp_5 = req_5.await.unwrap();
let cookie_2 = resp_5 let cookie_2 = resp_5
.cookies() .cookies()
.unwrap() .unwrap()
@ -231,7 +231,7 @@ mod test {
cookie_1.value().to_string() != cookie_2.value().to_string() cookie_1.value().to_string() != cookie_2.value().to_string()
); );
let result_5 = block_on(resp_5.json::<IndexResponse>()).unwrap(); let result_5 = resp_5.json::<IndexResponse>().await.unwrap();
assert_eq!( assert_eq!(
result_5, result_5,
IndexResponse { IndexResponse {
@ -243,8 +243,8 @@ mod test {
// Step 6: GET index, including session cookie #2 in request // Step 6: GET index, including session cookie #2 in request
// - response should be: {"counter": 2, "user_id": "ferris"} // - response should be: {"counter": 2, "user_id": "ferris"}
let req_6 = srv.get("/").cookie(cookie_2.clone()).send(); let req_6 = srv.get("/").cookie(cookie_2.clone()).send();
let mut resp_6 = srv.block_on(req_6).unwrap(); let mut resp_6 = req_6.await.unwrap();
let result_6 = block_on(resp_6.json::<IndexResponse>()).unwrap(); let result_6 = resp_6.json::<IndexResponse>().await.unwrap();
assert_eq!( assert_eq!(
result_6, result_6,
IndexResponse { IndexResponse {
@ -257,8 +257,8 @@ mod test {
// - updates session state in redis: {"counter": 3, "user_id": "ferris"} // - updates session state in redis: {"counter": 3, "user_id": "ferris"}
// - response should be: {"counter": 2, "user_id": None} // - response should be: {"counter": 2, "user_id": None}
let req_7 = srv.post("/do_something").cookie(cookie_2.clone()).send(); let req_7 = srv.post("/do_something").cookie(cookie_2.clone()).send();
let mut resp_7 = srv.block_on(req_7).unwrap(); let mut resp_7 = req_7.await.unwrap();
let result_7 = block_on(resp_7.json::<IndexResponse>()).unwrap(); let result_7 = resp_7.json::<IndexResponse>().await.unwrap();
assert_eq!( assert_eq!(
result_7, result_7,
IndexResponse { IndexResponse {
@ -271,7 +271,7 @@ mod test {
// - set-cookie actix-session will be in response (session cookie #3) // - set-cookie actix-session will be in response (session cookie #3)
// - response should be: {"counter": 0, "user_id": None} // - response should be: {"counter": 0, "user_id": None}
let req_8 = srv.get("/").cookie(cookie_1.clone()).send(); let req_8 = srv.get("/").cookie(cookie_1.clone()).send();
let mut resp_8 = srv.block_on(req_8).unwrap(); let mut resp_8 = req_8.await.unwrap();
let cookie_3 = resp_8 let cookie_3 = resp_8
.cookies() .cookies()
.unwrap() .unwrap()
@ -279,7 +279,7 @@ mod test {
.into_iter() .into_iter()
.find(|c| c.name() == "test-session") .find(|c| c.name() == "test-session")
.unwrap(); .unwrap();
let result_8 = block_on(resp_8.json::<IndexResponse>()).unwrap(); let result_8 = resp_8.json::<IndexResponse>().await.unwrap();
assert_eq!( assert_eq!(
result_8, result_8,
IndexResponse { IndexResponse {
@ -293,7 +293,7 @@ mod test {
// - set-cookie actix-session will be in response with session cookie #2 // - set-cookie actix-session will be in response with session cookie #2
// invalidation logic // invalidation logic
let req_9 = srv.post("/logout").cookie(cookie_2.clone()).send(); let req_9 = srv.post("/logout").cookie(cookie_2.clone()).send();
let resp_9 = srv.block_on(req_9).unwrap(); let resp_9 = req_9.await.unwrap();
let cookie_4 = resp_9 let cookie_4 = resp_9
.cookies() .cookies()
.unwrap() .unwrap()
@ -307,8 +307,8 @@ mod test {
// - set-cookie actix-session will be in response (session cookie #3) // - set-cookie actix-session will be in response (session cookie #3)
// - response should be: {"counter": 0, "user_id": None} // - response should be: {"counter": 0, "user_id": None}
let req_10 = srv.get("/").cookie(cookie_2.clone()).send(); let req_10 = srv.get("/").cookie(cookie_2.clone()).send();
let mut resp_10 = srv.block_on(req_10).unwrap(); let mut resp_10 = req_10.await.unwrap();
let result_10 = block_on(resp_10.json::<IndexResponse>()).unwrap(); let result_10 = resp_10.json::<IndexResponse>().await.unwrap();
assert_eq!( assert_eq!(
result_10, result_10,
IndexResponse { IndexResponse {

14
actix_redis/Cargo.toml → redis/Cargo.toml
View File

@ -1,16 +1,16 @@
[package] [package]
name = "actix_redis" name = "actix_redis"
version = "0.1.0" version = "1.0.0"
authors = ["dowwie <dkcdkg@gmail.com>"] authors = ["dowwie <dkcdkg@gmail.com>"]
edition = "2018" edition = "2018"
workspace = ".."
[dependencies] [dependencies]
actix = "0.8.2" actix = "0.9.0-alpha.2"
actix-web = "1.0.3" actix-rt = "1.0.0"
actix-redis = "0.6" actix-web = "2.0.0-alpha.6"
futures = "0.1.23" actix-redis = "0.8.0-alpha.1"
redis-async = "0.4.0" futures = "0.3.1"
redis-async = "0.6.1"
serde = "1.0.71" serde = "1.0.71"
serde_derive = "1.0.71" serde_derive = "1.0.71"
env_logger = "0.6" env_logger = "0.6"

0
actix_redis/README.md → redis/README.md
View File

79
actix_redis/src/main.rs → redis/src/main.rs
View File

@ -4,9 +4,9 @@ extern crate redis_async;
extern crate serde_derive; extern crate serde_derive;
use actix::prelude::*; use actix::prelude::*;
use actix_redis::{Command, Error as ARError, RedisActor}; use actix_redis::{Command, RedisActor};
use actix_web::{middleware, web, App, Error as AWError, HttpResponse, HttpServer}; use actix_web::{middleware, web, App, Error as AWError, HttpResponse, HttpServer};
use futures::future::{join_all, Future}; use futures::future::join_all;
use redis_async::resp::RespValue; use redis_async::resp::RespValue;
#[derive(Deserialize)] #[derive(Deserialize)]
@ -16,10 +16,10 @@ pub struct CacheInfo {
three: String, three: String,
} }
fn cache_stuff( async fn cache_stuff(
info: web::Json<CacheInfo>, info: web::Json<CacheInfo>,
redis: web::Data<Addr<RedisActor>>, redis: web::Data<Addr<RedisActor>>,
) -> impl Future<Item = HttpResponse, Error = AWError> { ) -> Result<HttpResponse, AWError> {
let info = info.into_inner(); let info = info.into_inner();
let one = redis.send(Command(resp_array!["SET", "mydomain:one", info.one])); let one = redis.send(Command(resp_array!["SET", "mydomain:one", info.one]));
@ -33,47 +33,51 @@ fn cache_stuff(
// canceled and an error will be returned immediately. If all futures complete // canceled and an error will be returned immediately. If all futures complete
// successfully, however, then the returned future will succeed with a `Vec` of // successfully, however, then the returned future will succeed with a `Vec` of
// all the successful results. // all the successful results.
let info_set = join_all(vec![one, two, three].into_iter()); let res: Vec<Result<RespValue, AWError>> =
join_all(vec![one, two, three].into_iter())
.await
.into_iter()
.map(|item| {
item.map_err(AWError::from)
.and_then(|res| res.map_err(AWError::from))
})
.collect();
info_set // successful operations return "OK", so confirm that all returned as so
.map_err(AWError::from) if !res.iter().all(|res| match res {
.and_then(|res: Vec<Result<RespValue, ARError>>| Ok(RespValue::SimpleString(x)) if x == "OK" => true,
// successful operations return "OK", so confirm that all returned as so _ => false,
if !res.iter().all(|res| match res { }) {
Ok(RespValue::SimpleString(x)) if x=="OK" => true, Ok(HttpResponse::InternalServerError().finish())
_ => false } else {
}) { Ok(HttpResponse::Ok().body("successfully cached values"))
Ok(HttpResponse::InternalServerError().finish()) }
} else {
Ok(HttpResponse::Ok().body("successfully cached values"))
}
)
} }
fn del_stuff( async fn del_stuff(redis: web::Data<Addr<RedisActor>>) -> Result<HttpResponse, AWError> {
redis: web::Data<Addr<RedisActor>>, let res = redis
) -> impl Future<Item = HttpResponse, Error = AWError> {
redis
.send(Command(resp_array![ .send(Command(resp_array![
"DEL", "DEL",
"mydomain:one", "mydomain:one",
"mydomain:two", "mydomain:two",
"mydomain:three" "mydomain:three"
])) ]))
.map_err(AWError::from) .await?;
.and_then(|res: Result<RespValue, ARError>| match &res {
Ok(RespValue::Integer(x)) if x == &3 => { match res {
Ok(HttpResponse::Ok().body("successfully deleted values")) Ok(RespValue::Integer(x)) if x == 3 => {
} Ok(HttpResponse::Ok().body("successfully deleted values"))
_ => { }
println!("---->{:?}", res); _ => {
Ok(HttpResponse::InternalServerError().finish()) println!("---->{:?}", res);
} Ok(HttpResponse::InternalServerError().finish())
}) }
}
} }
fn main() -> std::io::Result<()> { #[actix_rt::main]
std::env::set_var("RUST_LOG", "actix_web=info,actix_redis=info"); async fn main() -> std::io::Result<()> {
std::env::set_var("RUST_LOG", "actix_web=trace,actix_redis=trace");
env_logger::init(); env_logger::init();
HttpServer::new(|| { HttpServer::new(|| {
@ -84,10 +88,11 @@ fn main() -> std::io::Result<()> {
.wrap(middleware::Logger::default()) .wrap(middleware::Logger::default())
.service( .service(
web::resource("/stuff") web::resource("/stuff")
.route(web::post().to_async(cache_stuff)) .route(web::post().to(cache_stuff))
.route(web::delete().to_async(del_stuff)), .route(web::delete().to(del_stuff)),
) )
}) })
.bind("0.0.0.0:8080")? .bind("0.0.0.0:8080")?
.run() .start()
.await
} }

20
websocket/Cargo.toml
View File

@ -1,9 +1,8 @@
[package] [package]
name = "websocket" name = "websocket"
version = "0.1.0" version = "2.0.0"
authors = ["Nikolay Kim <fafhrd91@gmail.com>"] authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
edition = "2018" edition = "2018"
workspace = ".."
[[bin]] [[bin]]
name = "websocket-server" name = "websocket-server"
@ -14,12 +13,13 @@ name = "websocket-client"
path = "src/client.rs" path = "src/client.rs"
[dependencies] [dependencies]
actix = "0.8.2" actix = "0.9.0-alpha.2"
actix-codec = "0.1.2" actix-codec = "0.2.0"
actix-web = "1.0.0" actix-web = "2.0.0-alpha.6"
actix-web-actors = "1.0.0" actix-web-actors = "2.0.0-alpha.1"
actix-files = "0.1.1" actix-files = "0.2.0-alpha.3"
awc = "0.2.1" actix-rt = "1.0.0"
awc = "1.0.0"
env_logger = "0.6" env_logger = "0.6"
futures = "0.1" futures = "0.3.1"
bytes = "0.4" bytes = "0.5.3"

71
websocket/src/client.rs
View File

@ -10,52 +10,47 @@ use awc::{
ws::{Codec, Frame, Message}, ws::{Codec, Frame, Message},
Client, Client,
}; };
use futures::{ use bytes::Bytes;
lazy, use futures::stream::{SplitSink, StreamExt};
stream::{SplitSink, Stream},
Future,
};
fn main() { #[actix_rt::main]
async fn main() {
::std::env::set_var("RUST_LOG", "actix_web=info"); ::std::env::set_var("RUST_LOG", "actix_web=info");
env_logger::init(); env_logger::init();
let sys = actix::System::new("ws-example");
Arbiter::spawn(lazy(|| { let (response, framed) = Client::new()
Client::new() .ws("http://127.0.0.1:8080/ws/")
.ws("http://127.0.0.1:8080/ws/") .connect()
.connect() .await
.map_err(|e| { .map_err(|e| {
println!("Error: {}", e); println!("Error: {}", e);
}) })
.map(|(response, framed)| { .unwrap();
println!("{:?}", response);
let (sink, stream) = framed.split();
let addr = ChatClient::create(|ctx| {
ChatClient::add_stream(stream, ctx);
ChatClient(SinkWrite::new(sink, ctx))
});
// start console loop println!("{:?}", response);
thread::spawn(move || loop { let (sink, stream) = framed.split();
let mut cmd = String::new(); let addr = ChatClient::create(|ctx| {
if io::stdin().read_line(&mut cmd).is_err() { ChatClient::add_stream(stream, ctx);
println!("error"); ChatClient(SinkWrite::new(sink, ctx))
return; });
}
addr.do_send(ClientCommand(cmd));
});
})
}));
let _ = sys.run(); // start console loop
thread::spawn(move || loop {
let mut cmd = String::new();
if io::stdin().read_line(&mut cmd).is_err() {
println!("error");
return;
}
addr.do_send(ClientCommand(cmd));
});
} }
struct ChatClient<T>(SinkWrite<SplitSink<Framed<T, Codec>>>) struct ChatClient<T>(SinkWrite<Message, SplitSink<Framed<T, Codec>, Message>>)
where where
T: AsyncRead + AsyncWrite; T: AsyncRead + AsyncWrite;
#[derive(Message)] #[derive(Message)]
#[rtype(result = "()")]
struct ClientCommand(String); struct ClientCommand(String);
impl<T: 'static> Actor for ChatClient<T> impl<T: 'static> Actor for ChatClient<T>
@ -83,7 +78,7 @@ where
{ {
fn hb(&self, ctx: &mut Context<Self>) { fn hb(&self, ctx: &mut Context<Self>) {
ctx.run_later(Duration::new(1, 0), |act, ctx| { ctx.run_later(Duration::new(1, 0), |act, ctx| {
act.0.write(Message::Ping(String::new())).unwrap(); act.0.write(Message::Ping(Bytes::from_static(b""))).unwrap();
act.hb(ctx); act.hb(ctx);
// client should also check for a timeout here, similar to the // client should also check for a timeout here, similar to the
@ -105,12 +100,12 @@ where
} }
/// Handle server websocket messages /// Handle server websocket messages
impl<T: 'static> StreamHandler<Frame, WsProtocolError> for ChatClient<T> impl<T: 'static> StreamHandler<Result<Frame, WsProtocolError>> for ChatClient<T>
where where
T: AsyncRead + AsyncWrite, T: AsyncRead + AsyncWrite,
{ {
fn handle(&mut self, msg: Frame, _ctx: &mut Context<Self>) { fn handle(&mut self, msg: Result<Frame, WsProtocolError>, _: &mut Context<Self>) {
if let Frame::Text(txt) = msg { if let Ok(Frame::Text(txt)) = msg {
println!("Server: {:?}", txt) println!("Server: {:?}", txt)
} }
} }

32
websocket/src/main.rs
View File

@ -16,10 +16,10 @@ const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(5);
const CLIENT_TIMEOUT: Duration = Duration::from_secs(10); const CLIENT_TIMEOUT: Duration = Duration::from_secs(10);
/// do websocket handshake and start `MyWebSocket` actor /// do websocket handshake and start `MyWebSocket` actor
fn ws_index(r: HttpRequest, stream: web::Payload) -> Result<HttpResponse, Error> { async fn ws_index(r: HttpRequest, stream: web::Payload) -> Result<HttpResponse, Error> {
println!("{:?}", r); println!("{:?}", r);
let res = ws::start(MyWebSocket::new(), &r, stream); let res = ws::start(MyWebSocket::new(), &r, stream);
println!("{:?}", res.as_ref().unwrap()); println!("{:?}", res);
res res
} }
@ -41,24 +41,28 @@ impl Actor for MyWebSocket {
} }
/// Handler for `ws::Message` /// Handler for `ws::Message`
impl StreamHandler<ws::Message, ws::ProtocolError> for MyWebSocket { impl StreamHandler<Result<ws::Message, ws::ProtocolError>> for MyWebSocket {
fn handle(&mut self, msg: ws::Message, ctx: &mut Self::Context) { fn handle(
&mut self,
msg: Result<ws::Message, ws::ProtocolError>,
ctx: &mut Self::Context,
) {
// process websocket messages // process websocket messages
println!("WS: {:?}", msg); println!("WS: {:?}", msg);
match msg { match msg {
ws::Message::Ping(msg) => { Ok(ws::Message::Ping(msg)) => {
self.hb = Instant::now(); self.hb = Instant::now();
ctx.pong(&msg); ctx.pong(&msg);
} }
ws::Message::Pong(_) => { Ok(ws::Message::Pong(_)) => {
self.hb = Instant::now(); self.hb = Instant::now();
} }
ws::Message::Text(text) => ctx.text(text), Ok(ws::Message::Text(text)) => ctx.text(text),
ws::Message::Binary(bin) => ctx.binary(bin), Ok(ws::Message::Binary(bin)) => ctx.binary(bin),
ws::Message::Close(_) => { Ok(ws::Message::Close(_)) => {
ctx.stop(); ctx.stop();
} }
ws::Message::Nop => (), _ => ctx.stop(),
} }
} }
} }
@ -85,12 +89,13 @@ impl MyWebSocket {
return; return;
} }
ctx.ping(""); ctx.ping(b"");
}); });
} }
} }
fn main() -> std::io::Result<()> { #[actix_rt::main]
async fn main() -> std::io::Result<()> {
std::env::set_var("RUST_LOG", "actix_server=info,actix_web=info"); std::env::set_var("RUST_LOG", "actix_server=info,actix_web=info");
env_logger::init(); env_logger::init();
@ -105,5 +110,6 @@ fn main() -> std::io::Result<()> {
}) })
// start http server on 127.0.0.1:8080 // start http server on 127.0.0.1:8080
.bind("127.0.0.1:8080")? .bind("127.0.0.1:8080")?
.run() .start()
.await
} }