actix/examples
1
0
Fork 0
mirror of https://github.com/actix/examples synced 2025-06-28 18:00:37 +02:00
Code Issues Releases Wiki Activity

split websocket-chat example

Browse Source
This commit is contained in:
Nikolay Kim
2018-05-11 14:35:43 -07:00
parent 88194b9cd4
commit 16f7649b29
17 changed files with 1109 additions and 210 deletions
Download Patch File Download Diff File Expand all files Collapse all files

160
websocket-chat/src/client.rs
View File

@ -1,160 +0,0 @@
#[macro_use]
extern crate actix;
extern crate byteorder;
extern crate bytes;
extern crate futures;
extern crate serde;
extern crate serde_json;
extern crate tokio_core;
extern crate tokio_io;
#[macro_use]
extern crate serde_derive;
use actix::prelude::*;
use futures::Future;
use std::str::FromStr;
use std::time::Duration;
use std::{io, net, process, thread};
use tokio_core::net::TcpStream;
use tokio_io::codec::FramedRead;
use tokio_io::io::WriteHalf;
use tokio_io::AsyncRead;
mod codec;
fn main() {
let sys = actix::System::new("chat-client");
// Connect to server
let addr = net::SocketAddr::from_str("127.0.0.1:12345").unwrap();
Arbiter::handle().spawn(
TcpStream::connect(&addr, Arbiter::handle())
.and_then(|stream| {
let addr: Addr<Syn, _> = ChatClient::create(|ctx| {
let (r, w) = stream.split();
ChatClient::add_stream(
FramedRead::new(r, codec::ClientChatCodec),
ctx,
);
ChatClient {
framed: actix::io::FramedWrite::new(
w,
codec::ClientChatCodec,
ctx,
),
}
});
// 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));
});
futures::future::ok(())
})
.map_err(|e| {
println!("Can not connect to server: {}", e);
process::exit(1)
}),
);
println!("Running chat client");
sys.run();
}
struct ChatClient {
framed: actix::io::FramedWrite<WriteHalf<TcpStream>, codec::ClientChatCodec>,
}
#[derive(Message)]
struct ClientCommand(String);
impl Actor for ChatClient {
type Context = Context<Self>;
fn started(&mut self, ctx: &mut Context<Self>) {
// start heartbeats otherwise server will disconnect after 10 seconds
self.hb(ctx)
}
fn stopped(&mut self, _: &mut Context<Self>) {
println!("Disconnected");
// Stop application on disconnect
Arbiter::system().do_send(actix::msgs::SystemExit(0));
}
}
impl ChatClient {
fn hb(&self, ctx: &mut Context<Self>) {
ctx.run_later(Duration::new(1, 0), |act, ctx| {
act.framed.write(codec::ChatRequest::Ping);
act.hb(ctx);
});
}
}
impl actix::io::WriteHandler<io::Error> for ChatClient {}
/// Handle stdin commands
impl Handler<ClientCommand> for ChatClient {
type Result = ();
fn handle(&mut self, msg: ClientCommand, _: &mut Context<Self>) {
let m = msg.0.trim();
if m.is_empty() {
return;
}
// we check for /sss type of messages
if m.starts_with('/') {
let v: Vec<&str> = m.splitn(2, ' ').collect();
match v[0] {
"/list" => {
self.framed.write(codec::ChatRequest::List);
}
"/join" => {
if v.len() == 2 {
self.framed
.write(codec::ChatRequest::Join(v[1].to_owned()));
} else {
println!("!!! room name is required");
}
}
_ => println!("!!! unknown command"),
}
} else {
self.framed
.write(codec::ChatRequest::Message(m.to_owned()));
}
}
}
/// Server communication
impl StreamHandler<codec::ChatResponse, io::Error> for ChatClient {
fn handle(&mut self, msg: codec::ChatResponse, _: &mut Context<Self>) {
match msg {
codec::ChatResponse::Message(ref msg) => {
println!("message: {}", msg);
}
codec::ChatResponse::Joined(ref msg) => {
println!("!!! joined: {}", msg);
}
codec::ChatResponse::Rooms(rooms) => {
println!("\n!!! Available rooms:");
for room in rooms {
println!("{}", room);
}
println!("");
}
_ => (),
}
}
}

17
websocket-chat/src/main.rs
View File

@ -8,8 +8,6 @@ extern crate serde;
extern crate serde_json;
extern crate tokio_core;
extern crate tokio_io;
#[macro_use]
extern crate serde_derive;
#[macro_use]
extern crate actix;
@ -22,9 +20,7 @@ use actix_web::server::HttpServer;
use actix_web::ws::WsWriter;
use actix_web::{fs, http, ws, App, Error, HttpRequest, HttpResponse};
mod codec;
mod server;
mod session;
/// This is our websocket route state, this state is shared with all route
/// instances via `HttpContext::state()`
@ -96,10 +92,10 @@ impl Actor for WsChatSession {
}
/// Handle messages from chat server, we simply send it to peer websocket
impl Handler<session::Message> for WsChatSession {
impl Handler<server::Message> for WsChatSession {
type Result = ();
fn handle(&mut self, msg: session::Message, ctx: &mut Self::Context) {
fn handle(&mut self, msg: server::Message, ctx: &mut Self::Context) {
ctx.text(msg.0);
}
}
@ -170,7 +166,7 @@ impl StreamHandler<ws::Message, ws::ProtocolError> for WsChatSession {
m.to_owned()
};
// send message to chat server
ctx.state().addr.do_send(server::Message {
ctx.state().addr.do_send(server::ClientMessage {
id: self.id,
msg: msg,
room: self.room.clone(),
@ -192,13 +188,6 @@ fn main() {
// Start chat server actor in separate thread
let server: Addr<Syn, _> = Arbiter::start(|_| server::ChatServer::default());
// Start tcp server in separate thread
let srv = server.clone();
Arbiter::new("tcp-server").do_send::<msgs::Execute>(msgs::Execute::new(move || {
session::TcpServer::new("127.0.0.1:12345", srv);
Ok(())
}));
// Create Http server with websocket support
HttpServer::new(move || {
// Websocket sessions state

16
websocket-chat/src/server.rs
View File

@ -7,7 +7,9 @@ use rand::{self, Rng, ThreadRng};
use std::cell::RefCell;
use std::collections::{HashMap, HashSet};
use session;
/// Chat server sends this messages to session
#[derive(Message)]
pub struct Message(pub String);
/// Message for chat server communications
@ -15,7 +17,7 @@ use session;
#[derive(Message)]
#[rtype(usize)]
pub struct Connect {
pub addr: Recipient<Syn, session::Message>,
pub addr: Recipient<Syn, Message>,
}
/// Session is disconnected
@ -26,7 +28,7 @@ pub struct Disconnect {
/// Send message to specific room
#[derive(Message)]
pub struct Message {
pub struct ClientMessage {
/// Id of the client session
pub id: usize,
/// Peer message
@ -54,7 +56,7 @@ pub struct Join {
/// `ChatServer` manages chat rooms and responsible for coordinating chat
/// session. implementation is super primitive
pub struct ChatServer {
sessions: HashMap<usize, Recipient<Syn, session::Message>>,
sessions: HashMap<usize, Recipient<Syn, Message>>,
rooms: HashMap<String, HashSet<usize>>,
rng: RefCell<ThreadRng>,
}
@ -80,7 +82,7 @@ impl ChatServer {
for id in sessions {
if *id != skip_id {
if let Some(addr) = self.sessions.get(id) {
let _ = addr.do_send(session::Message(message.to_owned()));
let _ = addr.do_send(Message(message.to_owned()));
}
}
}
@ -148,10 +150,10 @@ impl Handler<Disconnect> for ChatServer {
}
/// Handler for Message message.
impl Handler<Message> for ChatServer {
impl Handler<ClientMessage> for ChatServer {
type Result = ();
fn handle(&mut self, msg: Message, _: &mut Context<Self>) {
fn handle(&mut self, msg: ClientMessage, _: &mut Context<Self>) {
self.send_message(&msg.room, msg.msg.as_str(), msg.id);
}
}

55
websocket-chat/src/session.rs
View File

@ -168,58 +168,3 @@ impl ChatSession {
});
}
}
/// Define tcp server that will accept incoming tcp connection and create
/// chat actors.
pub struct TcpServer {
chat: Addr<Syn, ChatServer>,
}
impl TcpServer {
pub fn new(s: &str, chat: Addr<Syn, ChatServer>) {
// Create server listener
let addr = net::SocketAddr::from_str("127.0.0.1:12345").unwrap();
let listener = TcpListener::bind(&addr, Arbiter::handle()).unwrap();
// Our chat server `Server` is an actor, first we need to start it
// and then add stream on incoming tcp connections to it.
// TcpListener::incoming() returns stream of the (TcpStream, net::SocketAddr)
// items So to be able to handle this events `Server` actor has to
// implement stream handler `StreamHandler<(TcpStream,
// net::SocketAddr), io::Error>`
let _: () = TcpServer::create(|ctx| {
ctx.add_message_stream(
listener
.incoming()
.map_err(|_| ())
.map(|(t, a)| TcpConnect(t, a)),
);
TcpServer { chat: chat }
});
}
}
/// Make actor from `Server`
impl Actor for TcpServer {
/// Every actor has to provide execution `Context` in which it can run.
type Context = Context<Self>;
}
#[derive(Message)]
struct TcpConnect(TcpStream, net::SocketAddr);
/// Handle stream of TcpStream's
impl Handler<TcpConnect> for TcpServer {
type Result = ();
fn handle(&mut self, msg: TcpConnect, _: &mut Context<Self>) {
// For each incoming connection we create `ChatSession` actor
// with out chat server address.
let server = self.chat.clone();
let _: () = ChatSession::create(|ctx| {
let (r, w) = msg.0.split();
ChatSession::add_stream(FramedRead::new(r, ChatCodec), ctx);
ChatSession::new(server, actix::io::FramedWrite::new(w, ChatCodec, ctx))
});
}
}