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

update all websocket examples to v4

Browse Source
This commit is contained in:
Rob Ede
2022-02-18 01:44:53 +00:00
parent 1b23e3ff3d
commit 4d8573c3fe
40 changed files with 1340 additions and 1682 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
websockets/chat/Cargo.toml
View File

@ -8,15 +8,13 @@ name = "websocket-chat-server"
path = "src/main.rs"
[dependencies]
actix = "0.10"
actix-web = "3"
actix-web-actors = "3"
actix-files = "0.3"
actix = "0.12"
actix-files = "0.6.0-beta.16"
actix-web = "4.0.0-rc.3"
actix-web-actors = "4.0.0-beta.12"
rand = "0.7"
bytes = "0.5"
byteorder = "1.3"
futures = "0.3"
env_logger = "0.8"
env_logger = "0.9"
log = "0.4"
rand = "0.8"
serde = "1"
serde_json = "1"

254
websockets/chat/src/main.rs
View File

@ -1,20 +1,25 @@
use std::sync::{
atomic::{AtomicUsize, Ordering},
Arc,
use std::{
sync::{
atomic::{AtomicUsize, Ordering},
Arc,
},
time::Instant,
};
use std::time::{Duration, Instant};
use actix::*;
use actix_files as fs;
use actix_web::{web, App, Error, HttpRequest, HttpResponse, HttpServer, Responder};
use actix_files::{Files, NamedFile};
use actix_web::{
middleware::Logger, web, App, Error, HttpRequest, HttpResponse, HttpServer,
Responder,
};
use actix_web_actors::ws;
mod server;
mod session;
/// How often heartbeat pings are sent
const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(5);
/// How long before lack of client response causes a timeout
const CLIENT_TIMEOUT: Duration = Duration::from_secs(10);
async fn index() -> impl Responder {
NamedFile::open_async("./static/index.html").await.unwrap()
}
/// Entry point for our websocket route
async fn chat_route(
@ -23,7 +28,7 @@ async fn chat_route(
srv: web::Data<Addr<server::ChatServer>>,
) -> Result<HttpResponse, Error> {
ws::start(
WsChatSession {
session::WsChatSession {
id: 0,
hb: Instant::now(),
room: "Main".to_owned(),
@ -35,229 +40,36 @@ async fn chat_route(
)
}
/// Displays and affects state
async fn get_count(count: web::Data<Arc<AtomicUsize>>) -> impl Responder {
let current_count = count.fetch_add(1, Ordering::SeqCst);
/// Displays state
async fn get_count(count: web::Data<AtomicUsize>) -> impl Responder {
let current_count = count.load(Ordering::SeqCst);
format!("Visitors: {}", current_count)
}
struct WsChatSession {
/// unique session id
id: usize,
/// Client must send ping at least once per 10 seconds (CLIENT_TIMEOUT),
/// otherwise we drop connection.
hb: Instant,
/// joined room
room: String,
/// peer name
name: Option<String>,
/// Chat server
addr: Addr<server::ChatServer>,
}
impl Actor for WsChatSession {
type Context = ws::WebsocketContext<Self>;
/// Method is called on actor start.
/// We register ws session with ChatServer
fn started(&mut self, ctx: &mut Self::Context) {
// we'll start heartbeat process on session start.
self.hb(ctx);
// 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
let addr = ctx.address();
self.addr
.send(server::Connect {
addr: addr.recipient(),
})
.into_actor(self)
.then(|res, act, ctx| {
match res {
Ok(res) => act.id = res,
// something is wrong with chat server
_ => ctx.stop(),
}
fut::ready(())
})
.wait(ctx);
}
fn stopping(&mut self, _: &mut Self::Context) -> Running {
// notify chat server
self.addr.do_send(server::Disconnect { id: self.id });
Running::Stop
}
}
/// Handle messages from chat server, we simply send it to peer websocket
impl Handler<server::Message> for WsChatSession {
type Result = ();
fn handle(&mut self, msg: server::Message, ctx: &mut Self::Context) {
ctx.text(msg.0);
}
}
/// WebSocket message handler
impl StreamHandler<Result<ws::Message, ws::ProtocolError>> for WsChatSession {
fn handle(
&mut self,
msg: Result<ws::Message, ws::ProtocolError>,
ctx: &mut Self::Context,
) {
let msg = match msg {
Err(_) => {
ctx.stop();
return;
}
Ok(msg) => msg,
};
println!("WEBSOCKET MESSAGE: {:?}", msg);
match msg {
ws::Message::Ping(msg) => {
self.hb = Instant::now();
ctx.pong(&msg);
}
ws::Message::Pong(_) => {
self.hb = Instant::now();
}
ws::Message::Text(text) => {
let m = text.trim();
// we check for /sss type of messages
if m.starts_with('/') {
let v: Vec<&str> = m.splitn(2, ' ').collect();
match v[0] {
"/list" => {
// Send ListRooms message to chat server and wait for
// response
println!("List rooms");
self.addr
.send(server::ListRooms)
.into_actor(self)
.then(|res, _, ctx| {
match res {
Ok(rooms) => {
for room in rooms {
ctx.text(room);
}
}
_ => println!("Something is wrong"),
}
fut::ready(())
})
.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();
self.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 {
let msg = if let Some(ref name) = self.name {
format!("{}: {}", name, m)
} else {
m.to_owned()
};
// send message to chat server
self.addr.do_send(server::ClientMessage {
id: self.id,
msg,
room: self.room.clone(),
})
}
}
ws::Message::Binary(_) => println!("Unexpected binary"),
ws::Message::Close(reason) => {
ctx.close(reason);
ctx.stop();
}
ws::Message::Continuation(_) => {
ctx.stop();
}
ws::Message::Nop => (),
}
}
}
impl WsChatSession {
/// helper method that sends ping to client every second.
///
/// also this method checks heartbeats from client
fn hb(&self, ctx: &mut ws::WebsocketContext<Self>) {
ctx.run_interval(HEARTBEAT_INTERVAL, |act, ctx| {
// check client heartbeats
if Instant::now().duration_since(act.hb) > CLIENT_TIMEOUT {
// heartbeat timed out
println!("Websocket Client heartbeat failed, disconnecting!");
// notify chat server
act.addr.do_send(server::Disconnect { id: act.id });
// stop actor
ctx.stop();
// don't try to send a ping
return;
}
ctx.ping(b"");
});
}
}
#[actix_web::main]
async fn main() -> std::io::Result<()> {
env_logger::init();
env_logger::init_from_env(env_logger::Env::new().default_filter_or("info"));
// App state
// We are keeping a count of the number of visitors
// set up applications state
// keep a count of the number of visitors
let app_state = Arc::new(AtomicUsize::new(0));
// Start chat server actor
// start chat server actor
let server = server::ChatServer::new(app_state.clone()).start();
// Create Http server with websocket support
log::info!("starting HTTP server at http://localhost:8080");
HttpServer::new(move || {
App::new()
.data(app_state.clone())
.data(server.clone())
// redirect to websocket.html
.service(web::resource("/").route(web::get().to(|| {
HttpResponse::Found()
.header("LOCATION", "/static/websocket.html")
.finish()
})))
.route("/count/", web::get().to(get_count))
// websocket
.service(web::resource("/ws/").to(chat_route))
// static resources
.service(fs::Files::new("/static/", "static/"))
.app_data(web::Data::from(app_state.clone()))
.app_data(web::Data::new(server.clone()))
.service(web::resource("/").to(index))
.route("/count", web::get().to(get_count))
.route("/ws", web::get().to(chat_route))
.service(Files::new("/static", "./static"))
.wrap(Logger::default())
})
.workers(2)
.bind(("127.0.0.1", 8080))?
.run()
.await

26
websockets/chat/src/server.rs
View File

@ -2,15 +2,16 @@
//! And manages available rooms. Peers send messages to other peers in same
//! room through `ChatServer`.
use actix::prelude::*;
use rand::{self, rngs::ThreadRng, Rng};
use std::sync::{
atomic::{AtomicUsize, Ordering},
Arc,
use std::{
collections::{HashMap, HashSet},
sync::{
atomic::{AtomicUsize, Ordering},
Arc,
},
};
use std::collections::{HashMap, HashSet};
use actix::prelude::*;
use rand::{self, rngs::ThreadRng, Rng};
/// Chat server sends this messages to session
#[derive(Message)]
@ -56,14 +57,17 @@ impl actix::Message for ListRooms {
#[derive(Message)]
#[rtype(result = "()")]
pub struct Join {
/// Client id
/// Client ID
pub id: usize,
/// Room name
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 very naïve.
#[derive(Debug)]
pub struct ChatServer {
sessions: HashMap<usize, Recipient<Message>>,
rooms: HashMap<String, HashSet<usize>>,
@ -118,7 +122,7 @@ impl Handler<Connect> for ChatServer {
println!("Someone joined");
// notify all users in same room
self.send_message(&"Main".to_owned(), "Someone joined", 0);
self.send_message("Main", "Someone joined", 0);
// register session with random id
let id = self.rng.gen::<usize>();

206
websockets/chat/src/session.rs Normal file
View File

@ -0,0 +1,206 @@
use std::time::{Duration, Instant};
use actix::prelude::*;
use actix_web_actors::ws;
use crate::server;
/// How often heartbeat pings are sent
const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(5);
/// How long before lack of client response causes a timeout
const CLIENT_TIMEOUT: Duration = Duration::from_secs(10);
#[derive(Debug)]
pub struct WsChatSession {
/// unique session id
pub id: usize,
/// Client must send ping at least once per 10 seconds (CLIENT_TIMEOUT),
/// otherwise we drop connection.
pub hb: Instant,
/// joined room
pub room: String,
/// peer name
pub name: Option<String>,
/// Chat server
pub addr: Addr<server::ChatServer>,
}
impl WsChatSession {
/// helper method that sends ping to client every second.
///
/// also this method checks heartbeats from client
fn hb(&self, ctx: &mut ws::WebsocketContext<Self>) {
ctx.run_interval(HEARTBEAT_INTERVAL, |act, ctx| {
// check client heartbeats
if Instant::now().duration_since(act.hb) > CLIENT_TIMEOUT {
// heartbeat timed out
println!("Websocket Client heartbeat failed, disconnecting!");
// notify chat server
act.addr.do_send(server::Disconnect { id: act.id });
// stop actor
ctx.stop();
// don't try to send a ping
return;
}
ctx.ping(b"");
});
}
}
impl Actor for WsChatSession {
type Context = ws::WebsocketContext<Self>;
/// Method is called on actor start.
/// We register ws session with ChatServer
fn started(&mut self, ctx: &mut Self::Context) {
// we'll start heartbeat process on session start.
self.hb(ctx);
// 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
let addr = ctx.address();
self.addr
.send(server::Connect {
addr: addr.recipient(),
})
.into_actor(self)
.then(|res, act, ctx| {
match res {
Ok(res) => act.id = res,
// something is wrong with chat server
_ => ctx.stop(),
}
fut::ready(())
})
.wait(ctx);
}
fn stopping(&mut self, _: &mut Self::Context) -> Running {
// notify chat server
self.addr.do_send(server::Disconnect { id: self.id });
Running::Stop
}
}
/// Handle messages from chat server, we simply send it to peer websocket
impl Handler<server::Message> for WsChatSession {
type Result = ();
fn handle(&mut self, msg: server::Message, ctx: &mut Self::Context) {
ctx.text(msg.0);
}
}
/// WebSocket message handler
impl StreamHandler<Result<ws::Message, ws::ProtocolError>> for WsChatSession {
fn handle(
&mut self,
msg: Result<ws::Message, ws::ProtocolError>,
ctx: &mut Self::Context,
) {
let msg = match msg {
Err(_) => {
ctx.stop();
return;
}
Ok(msg) => msg,
};
log::debug!("WEBSOCKET MESSAGE: {:?}", msg);
match msg {
ws::Message::Ping(msg) => {
self.hb = Instant::now();
ctx.pong(&msg);
}
ws::Message::Pong(_) => {
self.hb = Instant::now();
}
ws::Message::Text(text) => {
let m = text.trim();
// we check for /sss type of messages
if m.starts_with('/') {
let v: Vec<&str> = m.splitn(2, ' ').collect();
match v[0] {
"/list" => {
// Send ListRooms message to chat server and wait for
// response
println!("List rooms");
self.addr
.send(server::ListRooms)
.into_actor(self)
.then(|res, _, ctx| {
match res {
Ok(rooms) => {
for room in rooms {
ctx.text(room);
}
}
_ => println!("Something is wrong"),
}
fut::ready(())
})
.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();
self.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 {
let msg = if let Some(ref name) = self.name {
format!("{}: {}", name, m)
} else {
m.to_owned()
};
// send message to chat server
self.addr.do_send(server::ClientMessage {
id: self.id,
msg,
room: self.room.clone(),
})
}
}
ws::Message::Binary(_) => println!("Unexpected binary"),
ws::Message::Close(reason) => {
ctx.close(reason);
ctx.stop();
}
ws::Message::Continuation(_) => {
ctx.stop();
}
ws::Message::Nop => (),
}
}
}

198
websockets/chat/static/index.html Normal file
View File

@ -0,0 +1,198 @@
<!DOCTYPE html>
<html>
<head>
<meta charset="utf-8" />
<title>Chat!</title>
<style>
:root {
font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto,
Oxygen, Ubuntu, Cantarell, 'Open Sans', 'Helvetica Neue', sans-serif;
font-size: 18px;
}
input[type='text'] {
font-size: inherit;
}
#log {
width: 30em;
height: 20em;
overflow: auto;
margin: 0.5em 0;
border: 1px solid black;
}
#status {
padding: 0 0.2em;
}
#text {
width: 17em;
padding: 0.5em;
}
.msg {
margin: 0;
padding: 0.25em 0.5em;
}
.msg--status {
/* a light yellow */
background-color: #ffffc9;
}
.msg--message {
/* a light blue */
background-color: #d2f4ff;
}
.msg--error {
background-color: pink;
}
</style>
</head>
<body>
<h1>Chat!</h1>
<div>
<button id="connect">Connect</button>
<span>Status:</span>
<span id="status">disconnected</span>
</div>
<div id="log"></div>
<form id="chatform">
<input type="text" id="text" />
<input type="submit" id="send" />
</form>
<hr />
<section>
<h2>Commands</h2>
<table style="border-spacing: 0.5em">
<tr>
<td>
<code>/list</code>
</td>
<td>list all available rooms</td>
</tr>
<tr>
<td>
<code>/join name</code>
</td>
<td>join room, if room does not exist, create new one</td>
</tr>
<tr>
<td>
<code>/name name</code>
</td>
<td>set session name</td>
</tr>
<tr>
<td>
<code>some message</code>
</td>
<td>just string, send message to all peers in same room</td>
</tr>
</table>
</section>
<script>
const $status = document.querySelector('#status')
const $connectButton = document.querySelector('#connect')
const $log = document.querySelector('#log')
const $form = document.querySelector('#chatform')
const $input = document.querySelector('#text')
/** @type {WebSocket | null} */
var socket = null
function log(msg, type = 'status') {
$log.innerHTML += `<p class="msg msg--${type}">${msg}</p>`
$log.scrollTop += 1000
}
function connect() {
disconnect()
const { location } = window
const proto = location.protocol.startsWith('https') ? 'wss' : 'ws'
const wsUri = `${proto}://${location.host}/ws`
log('Connecting...')
socket = new WebSocket(wsUri)
socket.onopen = () => {
log('Connected')
updateConnectionStatus()
}
socket.onmessage = (ev) => {
log('Received: ' + ev.data, 'message')
}
socket.onclose = () => {
log('Disconnected')
socket = null
updateConnectionStatus()
}
}
function disconnect() {
if (socket) {
log('Disconnecting...')
socket.close()
socket = null
updateConnectionStatus()
}
}
function updateConnectionStatus() {
if (socket) {
$status.style.backgroundColor = 'transparent'
$status.style.color = 'green'
$status.textContent = `connected`
$connectButton.innerHTML = 'Disconnect'
$input.focus()
} else {
$status.style.backgroundColor = 'red'
$status.style.color = 'white'
$status.textContent = 'disconnected'
$connectButton.textContent = 'Connect'
}
}
$connectButton.addEventListener('click', () => {
if (socket) {
disconnect()
} else {
connect()
}
updateConnectionStatus()
})
$form.addEventListener('submit', (ev) => {
ev.preventDefault()
const text = $input.value
log('Sending: ' + text)
socket.send(text)
$input.value = ''
$input.focus()
})
updateConnectionStatus()
</script>
</body>
</html>

112
websockets/chat/static/websocket.html
View File

@ -1,112 +0,0 @@
<!DOCTYPE html>
<html>
<head>
<title>Chat!</title>
<meta charset="utf-8" />
<script>
'use strict'
window.onload = () => {
let conn = null
const log = (msg) => {
div_log.innerHTML += msg + '<br>'
div_log.scroll(0, div_log.scrollTop + 1000)
}
const connect = () => {
disconnect()
const wsUri =
(window.location.protocol === 'https:' ? 'wss://' : 'ws://') +
window.location.host +
'/ws/'
conn = new WebSocket(wsUri)
log('Connecting...')
conn.onopen = function () {
log('Connected.')
update_ui()
}
conn.onmessage = function (e) {
log('Received: ' + e.data)
}
conn.onclose = function () {
log('Disconnected.')
conn = null
update_ui()
}
}
const disconnect = () => {
if (conn) {
log('Disconnecting...')
conn.close()
conn = null
update_ui()
}
}
const update_ui = () => {
if (!conn) {
span_status.textContent = 'disconnected'
btn_connect.textContent = 'Connect'
} else {
span_status.textContent = `connected (${conn.protocol})`
btn_connect.textContent = 'Disconnect'
}
}
btn_connect.onclick = () => {
if (!conn) {
connect()
} else {
disconnect()
}
update_ui()
}
btn_send.onclick = () => {
if (!conn) return
const text = input_text.value
log('Sending: ' + text)
conn.send(text)
input_text.value = ''
input_text.focus()
}
input_text.onkeyup = (e) => {
if (e.key === 'Enter') {
btn_send.click()
}
}
}
</script>
</head>
<body>
<h3>Chat!</h3>
<div>
<button id="btn_connect">Connect</button>
Status: <span id="span_status">disconnected</span>
</div>
<div
id="div_log"
style="width: 20em; height: 15em; overflow: auto; border: 1px solid black"
></div>
<input id="input_text" type="text" />
<input id="btn_send" type="button" value="Send" />
</body>
</html>