actix/actix-extras
1
0
Fork 0
mirror of https://github.com/actix/actix-extras.git synced 2024-11-28 01:32:57 +01:00
Code Issues Releases Wiki Activity

update websocket-chat example

Browse Source
This commit is contained in:
Nikolay Kim 2018-02-12 17:42:10 -08:00
parent 335ca8ff33
commit 7ccacb92ce
4 changed files with 40 additions and 35 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
examples/websocket-chat/src/client.rs
View File

@ -29,7 +29,7 @@ fn main() {
Arbiter::handle().spawn( Arbiter::handle().spawn(
TcpStream::connect(&addr, Arbiter::handle()) TcpStream::connect(&addr, Arbiter::handle())
.and_then(|stream| { .and_then(|stream| {
let addr: Addr<Syn<_>> = ChatClient::create(|ctx| { let addr: Addr<Syn, _> = ChatClient::create(|ctx| {
let (r, w) = stream.split(); let (r, w) = stream.split();
ChatClient::add_stream(FramedRead::new(r, codec::ClientChatCodec), ctx); ChatClient::add_stream(FramedRead::new(r, codec::ClientChatCodec), ctx);
ChatClient{ ChatClient{

36
examples/websocket-chat/src/main.rs
View File

@ -26,7 +26,7 @@ mod session;
/// This is our websocket route state, this state is shared with all route instances /// This is our websocket route state, this state is shared with all route instances
/// via `HttpContext::state()` /// via `HttpContext::state()`
struct WsChatSessionState { struct WsChatSessionState {
addr: Addr<Syn<server::ChatServer>>, addr: Addr<Syn, server::ChatServer>,
} }
/// Entry point for our route /// Entry point for our route
@ -62,10 +62,10 @@ impl Actor for WsChatSession {
// before processing any other events. // before processing any other events.
// HttpContext::state() is instance of WsChatSessionState, state is shared across all // HttpContext::state() is instance of WsChatSessionState, state is shared across all
// routes within application // routes within application
let addr: Addr<Syn<_>> = ctx.address(); let addr: Addr<Syn, _> = ctx.address();
ctx.state().addr.call( ctx.state().addr.call(server::Connect{addr: addr.subscriber()})
self, server::Connect{addr: addr.subscriber()}).then( .into_actor(self)
|res, act, ctx| { .then(|res, act, ctx| {
match res { match res {
Ok(res) => act.id = res, Ok(res) => act.id = res,
// something is wrong with chat server // something is wrong with chat server
@ -109,17 +109,19 @@ impl Handler<ws::Message> for WsChatSession {
"/list" => { "/list" => {
// Send ListRooms message to chat server and wait for response // Send ListRooms message to chat server and wait for response
println!("List rooms"); println!("List rooms");
ctx.state().addr.call(self, server::ListRooms).then(|res, _, ctx| { ctx.state().addr.call(server::ListRooms)
match res { .into_actor(self)
Ok(rooms) => { .then(|res, _, ctx| {
for room in rooms { match res {
ctx.text(room); Ok(rooms) => {
} for room in rooms {
}, ctx.text(room);
_ => println!("Something is wrong"), }
} },
fut::ok(()) _ => println!("Something is wrong"),
}).wait(ctx) }
fut::ok(())
}).wait(ctx)
// .wait(ctx) pauses all events in context, // .wait(ctx) pauses all events in context,
// so actor wont receive any new messages until it get list // so actor wont receive any new messages until it get list
// of rooms back // of rooms back
@ -172,7 +174,7 @@ fn main() {
let sys = actix::System::new("websocket-example"); let sys = actix::System::new("websocket-example");
// Start chat server actor in separate thread // Start chat server actor in separate thread
let server: Addr<Syn<_>> = Arbiter::start(|_| server::ChatServer::default()); let server: Addr<Syn, _> = Arbiter::start(|_| server::ChatServer::default());
// Start tcp server in separate thread // Start tcp server in separate thread
let srv = server.clone(); let srv = server.clone();

4
examples/websocket-chat/src/server.rs
View File

@ -15,7 +15,7 @@ use session;
#[derive(Message)] #[derive(Message)]
#[rtype(usize)] #[rtype(usize)]
pub struct Connect { pub struct Connect {
pub addr: SyncSubscriber<session::Message>, pub addr: Subscriber<Syn, session::Message>,
} }
/// Session is disconnected /// Session is disconnected
@ -54,7 +54,7 @@ pub struct Join {
/// `ChatServer` manages chat rooms and responsible for coordinating chat session. /// `ChatServer` manages chat rooms and responsible for coordinating chat session.
/// implementation is super primitive /// implementation is super primitive
pub struct ChatServer { pub struct ChatServer {
sessions: HashMap<usize, SyncSubscriber<session::Message>>, sessions: HashMap<usize, Subscriber<Syn, session::Message>>,
rooms: HashMap<String, HashSet<usize>>, rooms: HashMap<String, HashSet<usize>>,
rng: RefCell<ThreadRng>, rng: RefCell<ThreadRng>,
} }

33
examples/websocket-chat/src/session.rs
View File

@ -24,7 +24,7 @@ pub struct ChatSession {
/// unique session id /// unique session id
id: usize, id: usize,
/// this is address of chat server /// this is address of chat server
addr: Addr<Syn<ChatServer>>, addr: Addr<Syn, ChatServer>,
/// Client must send ping at least once per 10 seconds, otherwise we drop connection. /// Client must send ping at least once per 10 seconds, otherwise we drop connection.
hb: Instant, hb: Instant,
/// joined room /// joined room
@ -45,8 +45,9 @@ impl Actor for ChatSession {
// register self in chat server. `AsyncContext::wait` register // register self in chat server. `AsyncContext::wait` register
// future within context, but context waits until this future resolves // future within context, but context waits until this future resolves
// before processing any other events. // before processing any other events.
let addr: Addr<Syn<_>> = ctx.address(); let addr: Addr<Syn, _> = ctx.address();
self.addr.call(self, server::Connect{addr: addr.subscriber()}) self.addr.call(server::Connect{addr: addr.subscriber()})
.into_actor(self)
.then(|res, act, ctx| { .then(|res, act, ctx| {
match res { match res {
Ok(res) => act.id = res, Ok(res) => act.id = res,
@ -75,15 +76,17 @@ impl StreamHandler<ChatRequest, io::Error> for ChatSession {
ChatRequest::List => { ChatRequest::List => {
// Send ListRooms message to chat server and wait for response // Send ListRooms message to chat server and wait for response
println!("List rooms"); println!("List rooms");
self.addr.call(self, server::ListRooms).then(|res, act, ctx| { self.addr.call(server::ListRooms)
match res { .into_actor(self)
Ok(rooms) => { .then(|res, act, ctx| {
act.framed.write(ChatResponse::Rooms(rooms)); match res {
}, Ok(rooms) => {
_ => println!("Something is wrong"), act.framed.write(ChatResponse::Rooms(rooms));
} },
actix::fut::ok(()) _ => println!("Something is wrong"),
}).wait(ctx) }
actix::fut::ok(())
}).wait(ctx)
// .wait(ctx) pauses all events in context, // .wait(ctx) pauses all events in context,
// so actor wont receive any new messages until it get list of rooms back // so actor wont receive any new messages until it get list of rooms back
}, },
@ -121,7 +124,7 @@ impl Handler<Message> for ChatSession {
/// Helper methods /// Helper methods
impl ChatSession { impl ChatSession {
pub fn new(addr: Addr<Syn<ChatServer>>, pub fn new(addr: Addr<Syn,ChatServer>,
framed: actix::io::FramedWrite<WriteHalf<TcpStream>, ChatCodec>) -> ChatSession { framed: actix::io::FramedWrite<WriteHalf<TcpStream>, ChatCodec>) -> ChatSession {
ChatSession {id: 0, addr: addr, hb: Instant::now(), ChatSession {id: 0, addr: addr, hb: Instant::now(),
room: "Main".to_owned(), framed: framed} room: "Main".to_owned(), framed: framed}
@ -155,11 +158,11 @@ impl ChatSession {
/// Define tcp server that will accept incoming tcp connection and create /// Define tcp server that will accept incoming tcp connection and create
/// chat actors. /// chat actors.
pub struct TcpServer { pub struct TcpServer {
chat: Addr<Syn<ChatServer>>, chat: Addr<Syn, ChatServer>,
} }
impl TcpServer { impl TcpServer {
pub fn new(s: &str, chat: Addr<Syn<ChatServer>>) { pub fn new(s: &str, chat: Addr<Syn, ChatServer>) {
// Create server listener // Create server listener
let addr = net::SocketAddr::from_str("127.0.0.1:12345").unwrap(); let addr = net::SocketAddr::from_str("127.0.0.1:12345").unwrap();
let listener = TcpListener::bind(&addr, Arbiter::handle()).unwrap(); let listener = TcpListener::bind(&addr, Arbiter::handle()).unwrap();