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

upgrade actix-web to 0.6

Browse Source
This commit is contained in:
Nikolay Kim
2018-05-08 11:08:43 -07:00
parent 27de52b5d5
commit bbeb262a5c
55 changed files with 689 additions and 518 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@ -1,26 +1,27 @@
#[macro_use] extern crate actix;
extern crate bytes;
#[macro_use]
extern crate actix;
extern crate byteorder;
extern crate bytes;
extern crate futures;
extern crate tokio_io;
extern crate tokio_core;
extern crate serde;
extern crate serde_json;
#[macro_use] extern crate serde_derive;
extern crate tokio_core;
extern crate tokio_io;
#[macro_use]
extern crate serde_derive;
use std::{io, net, process, thread};
use actix::prelude::*;
use futures::Future;
use std::str::FromStr;
use std::time::Duration;
use futures::Future;
use tokio_io::AsyncRead;
use tokio_io::io::WriteHalf;
use tokio_io::codec::FramedRead;
use std::{io, net, process, thread};
use tokio_core::net::TcpStream;
use actix::prelude::*;
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");
@ -31,22 +32,28 @@ fn main() {
.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{
ChatClient::add_stream(
FramedRead::new(r, codec::ClientChatCodec),
ctx,
);
ChatClient {
framed: actix::io::FramedWrite::new(
w, codec::ClientChatCodec, ctx)}});
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));
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(())
@ -54,14 +61,13 @@ fn main() {
.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>,
}
@ -103,7 +109,7 @@ impl Handler<ClientCommand> for ChatClient {
fn handle(&mut self, msg: ClientCommand, _: &mut Context<Self>) {
let m = msg.0.trim();
if m.is_empty() {
return
return;
}
// we check for /sss type of messages
@ -112,18 +118,20 @@ impl Handler<ClientCommand> for ChatClient {
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()));
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()));
self.framed
.write(codec::ChatRequest::Message(m.to_owned()));
}
}
}
@ -131,7 +139,6 @@ impl Handler<ClientCommand> for ChatClient {
/// 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) => {

39
websocket-chat/src/codec.rs
View File

@ -1,13 +1,13 @@
#![allow(dead_code)]
use std::io;
use byteorder::{BigEndian, ByteOrder};
use bytes::{BufMut, BytesMut};
use serde_json as json;
use byteorder::{BigEndian , ByteOrder};
use bytes::{BytesMut, BufMut};
use tokio_io::codec::{Encoder, Decoder};
use std::io;
use tokio_io::codec::{Decoder, Encoder};
/// Client request
#[derive(Serialize, Deserialize, Debug, Message)]
#[serde(tag="cmd", content="data")]
#[serde(tag = "cmd", content = "data")]
pub enum ChatRequest {
/// List rooms
List,
@ -16,12 +16,12 @@ pub enum ChatRequest {
/// Send message
Message(String),
/// Ping
Ping
Ping,
}
/// Server response
#[derive(Serialize, Deserialize, Debug, Message)]
#[serde(tag="cmd", content="data")]
#[serde(tag = "cmd", content = "data")]
pub enum ChatResponse {
Ping,
@ -38,15 +38,14 @@ pub enum ChatResponse {
/// Codec for Client -> Server transport
pub struct ChatCodec;
impl Decoder for ChatCodec
{
impl Decoder for ChatCodec {
type Item = ChatRequest;
type Error = io::Error;
fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
let size = {
if src.len() < 2 {
return Ok(None)
return Ok(None);
}
BigEndian::read_u16(src.as_ref()) as usize
};
@ -61,12 +60,13 @@ impl Decoder for ChatCodec
}
}
impl Encoder for ChatCodec
{
impl Encoder for ChatCodec {
type Item = ChatResponse;
type Error = io::Error;
fn encode(&mut self, msg: ChatResponse, dst: &mut BytesMut) -> Result<(), Self::Error> {
fn encode(
&mut self, msg: ChatResponse, dst: &mut BytesMut,
) -> Result<(), Self::Error> {
let msg = json::to_string(&msg).unwrap();
let msg_ref: &[u8] = msg.as_ref();
@ -78,19 +78,17 @@ impl Encoder for ChatCodec
}
}
/// Codec for Server -> Client transport
pub struct ClientChatCodec;
impl Decoder for ClientChatCodec
{
impl Decoder for ClientChatCodec {
type Item = ChatResponse;
type Error = io::Error;
fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
let size = {
if src.len() < 2 {
return Ok(None)
return Ok(None);
}
BigEndian::read_u16(src.as_ref()) as usize
};
@ -105,12 +103,13 @@ impl Decoder for ClientChatCodec
}
}
impl Encoder for ClientChatCodec
{
impl Encoder for ClientChatCodec {
type Item = ChatRequest;
type Error = io::Error;
fn encode(&mut self, msg: ChatRequest, dst: &mut BytesMut) -> Result<(), Self::Error> {
fn encode(
&mut self, msg: ChatRequest, dst: &mut BytesMut,
) -> Result<(), Self::Error> {
let msg = json::to_string(&msg).unwrap();
let msg_ref: &[u8] = msg.as_ref();

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

@ -1,14 +1,15 @@
#![allow(unused_variables)]
extern crate rand;
extern crate bytes;
extern crate byteorder;
extern crate futures;
extern crate tokio_io;
extern crate tokio_core;
extern crate bytes;
extern crate env_logger;
extern crate futures;
extern crate rand;
extern crate serde;
extern crate serde_json;
#[macro_use] extern crate serde_derive;
extern crate tokio_core;
extern crate tokio_io;
#[macro_use]
extern crate serde_derive;
#[macro_use]
extern crate actix;
@ -18,14 +19,15 @@ use std::time::Instant;
use actix::*;
use actix_web::server::HttpServer;
use actix_web::{http, fs, ws, App, HttpRequest, HttpResponse, Error};
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()`
/// This is our websocket route state, this state is shared with all route
/// instances via `HttpContext::state()`
struct WsChatSessionState {
addr: Addr<Syn, server::ChatServer>,
}
@ -38,13 +40,16 @@ fn chat_route(req: HttpRequest<WsChatSessionState>) -> Result<HttpResponse, Erro
id: 0,
hb: Instant::now(),
room: "Main".to_owned(),
name: None})
name: None,
},
)
}
struct WsChatSession {
/// unique session id
id: usize,
/// 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,
/// joined room
room: String,
@ -61,10 +66,14 @@ impl Actor for WsChatSession {
// register self in chat server. `AsyncContext::wait` register
// future within context, but context waits until this future resolves
// before processing any other events.
// HttpContext::state() is instance of WsChatSessionState, state is shared across all
// routes within application
// HttpContext::state() is instance of WsChatSessionState, state is shared
// across all routes within application
let addr: Addr<Syn, _> = ctx.address();
ctx.state().addr.send(server::Connect{addr: addr.recipient()})
ctx.state()
.addr
.send(server::Connect {
addr: addr.recipient(),
})
.into_actor(self)
.then(|res, act, ctx| {
match res {
@ -73,12 +82,15 @@ impl Actor for WsChatSession {
_ => ctx.stop(),
}
fut::ok(())
}).wait(ctx);
})
.wait(ctx);
}
fn stopping(&mut self, ctx: &mut Self::Context) -> Running {
// notify chat server
ctx.state().addr.do_send(server::Disconnect{id: self.id});
ctx.state()
.addr
.do_send(server::Disconnect { id: self.id });
Running::Stop
}
}
@ -94,7 +106,6 @@ impl Handler<session::Message> for WsChatSession {
/// WebSocket message handler
impl StreamHandler<ws::Message, ws::ProtocolError> for WsChatSession {
fn handle(&mut self, msg: ws::Message, ctx: &mut Self::Context) {
println!("WEBSOCKET MESSAGE: {:?}", msg);
match msg {
@ -107,9 +118,12 @@ impl StreamHandler<ws::Message, ws::ProtocolError> for WsChatSession {
let v: Vec<&str> = m.splitn(2, ' ').collect();
match v[0] {
"/list" => {
// Send ListRooms message to chat server and wait for response
// Send ListRooms message to chat server and wait for
// response
println!("List rooms");
ctx.state().addr.send(server::ListRooms)
ctx.state()
.addr
.send(server::ListRooms)
.into_actor(self)
.then(|res, _, ctx| {
match res {
@ -117,33 +131,36 @@ impl StreamHandler<ws::Message, ws::ProtocolError> for WsChatSession {
for room in rooms {
ctx.text(room);
}
},
}
_ => println!("Something is wrong"),
}
fut::ok(())
}).wait(ctx)
})
.wait(ctx)
// .wait(ctx) pauses all events in context,
// so actor wont receive any new messages until it get list
// of rooms back
},
}
"/join" => {
if v.len() == 2 {
self.room = v[1].to_owned();
ctx.state().addr.do_send(
server::Join{id: self.id, name: self.room.clone()});
ctx.state().addr.do_send(server::Join {
id: self.id,
name: self.room.clone(),
});
ctx.text("joined");
} else {
ctx.text("!!! room name is required");
}
},
}
"/name" => {
if v.len() == 2 {
self.name = Some(v[1].to_owned());
} else {
ctx.text("!!! name is required");
}
},
}
_ => ctx.text(format!("!!! unknown command: {:?}", m)),
}
} else {
@ -153,14 +170,14 @@ impl StreamHandler<ws::Message, ws::ProtocolError> for WsChatSession {
m.to_owned()
};
// send message to chat server
ctx.state().addr.do_send(
server::Message{id: self.id,
msg: msg,
room: self.room.clone()})
ctx.state().addr.do_send(server::Message {
id: self.id,
msg: msg,
room: self.room.clone(),
})
}
},
ws::Message::Binary(bin) =>
println!("Unexpected binary"),
}
ws::Message::Binary(bin) => println!("Unexpected binary"),
ws::Message::Close(_) => {
ctx.stop();
}
@ -177,19 +194,19 @@ fn main() {
// 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(())
}));
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
let state = WsChatSessionState { addr: server.clone() };
HttpServer::new(move || {
// Websocket sessions state
let state = WsChatSessionState {
addr: server.clone(),
};
App::with_state(state)
App::with_state(state)
// redirect to websocket.html
.resource("/", |r| r.method(http::Method::GET).f(|_| {
HttpResponse::Found()
@ -200,8 +217,8 @@ fn main() {
.resource("/ws/", |r| r.route().f(chat_route))
// static resources
.handler("/static/", fs::StaticFiles::new("static/"))
})
.bind("127.0.0.1:8080").unwrap()
}).bind("127.0.0.1:8080")
.unwrap()
.start();
println!("Started http server: 127.0.0.1:8080");

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

@ -2,10 +2,10 @@
//! And manages available rooms. Peers send messages to other peers in same
//! room through `ChatServer`.
use actix::prelude::*;
use rand::{self, Rng, ThreadRng};
use std::cell::RefCell;
use std::collections::{HashMap, HashSet};
use rand::{self, Rng, ThreadRng};
use actix::prelude::*;
use session;
@ -51,8 +51,8 @@ pub struct Join {
pub name: String,
}
/// `ChatServer` manages chat rooms and responsible for coordinating chat session.
/// implementation is super primitive
/// `ChatServer` manages chat rooms and responsible for coordinating chat
/// session. implementation is super primitive
pub struct ChatServer {
sessions: HashMap<usize, Recipient<Syn, session::Message>>,
rooms: HashMap<String, HashSet<usize>>,
@ -112,7 +112,10 @@ impl Handler<Connect> for ChatServer {
self.sessions.insert(id, msg.addr);
// auto join session to Main room
self.rooms.get_mut(&"Main".to_owned()).unwrap().insert(id);
self.rooms
.get_mut(&"Main".to_owned())
.unwrap()
.insert(id);
// send id back
id
@ -174,7 +177,7 @@ impl Handler<Join> for ChatServer {
type Result = ();
fn handle(&mut self, msg: Join, _: &mut Context<Self>) {
let Join {id, name} = msg;
let Join { id, name } = msg;
let mut rooms = Vec::new();
// remove session from all rooms

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

@ -1,19 +1,18 @@
//! `ClientSession` is an actor, it manages peer tcp connection and
//! proxies commands from peer to `ChatServer`.
use std::{io, net};
use std::str::FromStr;
use std::time::{Instant, Duration};
use futures::Stream;
use tokio_io::AsyncRead;
use tokio_io::io::WriteHalf;
use std::str::FromStr;
use std::time::{Duration, Instant};
use std::{io, net};
use tokio_core::net::{TcpListener, TcpStream};
use tokio_io::codec::FramedRead;
use tokio_core::net::{TcpStream, TcpListener};
use tokio_io::io::WriteHalf;
use tokio_io::AsyncRead;
use actix::prelude::*;
use codec::{ChatCodec, ChatRequest, ChatResponse};
use server::{self, ChatServer};
use codec::{ChatRequest, ChatResponse, ChatCodec};
/// Chat server sends this messages to session
#[derive(Message)]
@ -25,7 +24,8 @@ pub struct ChatSession {
id: usize,
/// this is address of chat server
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,
/// joined room
room: String,
@ -46,7 +46,10 @@ impl Actor for ChatSession {
// future within context, but context waits until this future resolves
// before processing any other events.
let addr: Addr<Syn, _> = ctx.address();
self.addr.send(server::Connect{addr: addr.recipient()})
self.addr
.send(server::Connect {
addr: addr.recipient(),
})
.into_actor(self)
.then(|res, act, ctx| {
match res {
@ -55,12 +58,13 @@ impl Actor for ChatSession {
_ => ctx.stop(),
}
actix::fut::ok(())
}).wait(ctx);
})
.wait(ctx);
}
fn stopping(&mut self, ctx: &mut Self::Context) -> Running {
// notify chat server
self.addr.do_send(server::Disconnect{id: self.id});
self.addr.do_send(server::Disconnect { id: self.id });
Running::Stop
}
}
@ -69,49 +73,54 @@ impl actix::io::WriteHandler<io::Error> for ChatSession {}
/// To use `Framed` we have to define Io type and Codec
impl StreamHandler<ChatRequest, io::Error> for ChatSession {
/// This is main event loop for client requests
fn handle(&mut self, msg: ChatRequest, ctx: &mut Context<Self>) {
match msg {
ChatRequest::List => {
// Send ListRooms message to chat server and wait for response
println!("List rooms");
self.addr.send(server::ListRooms)
self.addr
.send(server::ListRooms)
.into_actor(self)
.then(|res, act, ctx| {
match res {
Ok(rooms) => {
act.framed.write(ChatResponse::Rooms(rooms));
},
}
_ => println!("Something is wrong"),
}
actix::fut::ok(())
}).wait(ctx)
})
.wait(ctx)
// .wait(ctx) pauses all events in context,
// so actor wont receive any new messages until it get list of rooms back
},
}
ChatRequest::Join(name) => {
println!("Join to room: {}", name);
self.room = name.clone();
self.addr.do_send(server::Join{id: self.id, name: name.clone()});
self.addr.do_send(server::Join {
id: self.id,
name: name.clone(),
});
self.framed.write(ChatResponse::Joined(name));
},
}
ChatRequest::Message(message) => {
// send message to chat server
println!("Peer message: {}", message);
self.addr.do_send(
server::Message{id: self.id,
msg: message, room:
self.room.clone()})
self.addr.do_send(server::Message {
id: self.id,
msg: message,
room: self.room.clone(),
})
}
// we update heartbeat time on ping from peer
ChatRequest::Ping =>
self.hb = Instant::now(),
ChatRequest::Ping => self.hb = Instant::now(),
}
}
}
/// Handler for Message, chat server sends this message, we just send string to peer
/// Handler for Message, chat server sends this message, we just send string to
/// peer
impl Handler<Message> for ChatSession {
type Result = ();
@ -123,13 +132,19 @@ impl Handler<Message> for ChatSession {
/// Helper methods
impl ChatSession {
pub fn new(addr: Addr<Syn,ChatServer>,
framed: actix::io::FramedWrite<WriteHalf<TcpStream>, ChatCodec>) -> ChatSession {
ChatSession {id: 0, addr: addr, hb: Instant::now(),
room: "Main".to_owned(), framed: framed}
pub fn new(
addr: Addr<Syn, ChatServer>,
framed: actix::io::FramedWrite<WriteHalf<TcpStream>, ChatCodec>,
) -> ChatSession {
ChatSession {
id: 0,
addr: addr,
hb: Instant::now(),
room: "Main".to_owned(),
framed: framed,
}
}
/// helper method that sends ping to client every second.
///
/// also this method check heartbeats from client
@ -141,7 +156,7 @@ impl ChatSession {
println!("Client heartbeat failed, disconnecting!");
// notify chat server
act.addr.do_send(server::Disconnect{id: act.id});
act.addr.do_send(server::Disconnect { id: act.id });
// stop actor
ctx.stop();
@ -154,7 +169,6 @@ impl ChatSession {
}
}
/// Define tcp server that will accept incoming tcp connection and create
/// chat actors.
pub struct TcpServer {
@ -169,14 +183,18 @@ impl TcpServer {
// 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>`
// 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}
ctx.add_message_stream(
listener
.incoming()
.map_err(|_| ())
.map(|(t, a)| TcpConnect(t, a)),
);
TcpServer { chat: chat }
});
}
}