Browse Source

Adding initial UI and Websocket server.

feature/settings-cleanup
Dessalines 4 years ago
parent
commit
a008534e7e
  1. 3
      API.md
  2. 4
      README.md
  3. 1792
      server/Cargo.lock
  4. 7
      server/Cargo.toml
  5. 7
      server/src/actions/comment.rs
  6. 2
      server/src/apub.rs
  7. 270
      server/src/bin/main.rs
  8. 22
      server/src/lib.rs
  9. 1
      server/src/websocket_server/mod.rs
  10. 269
      server/src/websocket_server/server.rs

3
API.md

@ -147,13 +147,11 @@
}
```
## Actions
- These are all posts to a user's outbox.
- The server then creates a post to the necessary inbox of the recipient, or the followers.
- Whenever a user accesses the site, they do a get from their inbox.
### Comments
#### [Create](https://www.w3.org/TR/activitystreams-vocabulary/#dfn-create)
```
{
@ -163,7 +161,6 @@
"object": comment_id, or post_id
}
```
#### [Delete](https://www.w3.org/TR/activitystreams-vocabulary/#dfn-delete)
```
{

4
README.md

@ -33,8 +33,12 @@ We have a twitter alternative (mastodon), a facebook alternative (friendica), so
- [helpful diesel examples](http://siciarz.net/24-days-rust-diesel/)
- [Mastodan public key server example](https://blog.joinmastodon.org/2018/06/how-to-implement-a-basic-activitypub-server/)
- [Recursive query for adjacency list for nested comments](https://stackoverflow.com/questions/192220/what-is-the-most-efficient-elegant-way-to-parse-a-flat-table-into-a-tree/192462#192462)
- https://github.com/sparksuite/simplemde-markdown-editor
- [Sticky Sidebar](https://stackoverflow.com/questions/38382043/how-to-use-css-position-sticky-to-keep-a-sidebar-visible-with-bootstrap-4/49111934)
## TODOs
- Endpoints
- DB
- Followers / following

1792
server/Cargo.lock
File diff suppressed because it is too large
View File

7
server/Cargo.toml

@ -11,3 +11,10 @@ activitypub = "0.1.4"
chrono = { version = "0.4", features = ["serde"] }
failure = "0.1.5"
serde_json = "*"
serde = { version = "1.0", features = ["derive"] }
actix = "*"
actix-web = "*"
env_logger = "*"
rand = "0.6.5"
strum = "0.14.0"
strum_macros = "0.14.0"

7
server/src/actions/comment.rs

@ -4,6 +4,13 @@ use diesel::*;
use diesel::result::Error;
use {Crud, Likeable};
// WITH RECURSIVE MyTree AS (
// SELECT * FROM comment WHERE parent_id IS NULL
// UNION ALL
// SELECT m.* FROM comment AS m JOIN MyTree AS t ON m.parent_id = t.id
// )
// SELECT * FROM MyTree;
#[derive(Queryable, Identifiable, PartialEq, Debug)]
#[table_name="comment"]
pub struct Comment {

2
server/src/apub.rs

@ -34,7 +34,7 @@ mod tests {
use super::User_;
use naive_now;
#[test]
#[test]
fn test_person() {
let expected_user = User_ {
id: 52,

270
server/src/bin/main.rs

@ -0,0 +1,270 @@
extern crate server;
use std::time::{Instant, Duration};
use server::actix::*;
use server::actix_web::server::HttpServer;
use server::actix_web::{fs, http, ws, App, Error, HttpRequest, HttpResponse};
/// 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);
use server::websocket_server::server::*;
use std::str::FromStr;
// use server::websocket_server::server::UserOperation::from_str;
/// This is our websocket route state, this state is shared with all route
/// instances via `HttpContext::state()`
struct WsChatSessionState {
addr: Addr<ChatServer>,
}
/// Entry point for our route
fn chat_route(req: &HttpRequest<WsChatSessionState>) -> Result<HttpResponse, Error> {
ws::start(
req,
WSSession {
id: 0,
hb: Instant::now()
},
)
}
struct WSSession {
/// unique session id
id: usize,
/// Client must send ping at least once per 10 seconds (CLIENT_TIMEOUT),
/// otherwise we drop connection.
hb: Instant
}
impl Actor for WSSession {
type Context = ws::WebsocketContext<Self, WsChatSessionState>;
/// 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();
ctx.state()
.addr
.send(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::ok(())
})
.wait(ctx);
}
fn stopping(&mut self, ctx: &mut Self::Context) -> Running {
// notify chat server
ctx.state().addr.do_send(Disconnect { id: self.id });
Running::Stop
}
}
/// Handle messages from chat server, we simply send it to peer websocket
impl Handler<WSMessage> for WSSession {
type Result = ();
fn handle(&mut self, msg: WSMessage, ctx: &mut Self::Context) {
ctx.text(msg.0);
}
}
use server::serde_json::Value;
/// WebSocket message handler
impl StreamHandler<ws::Message, ws::ProtocolError> for WSSession {
fn handle(&mut self, msg: ws::Message, ctx: &mut Self::Context) {
// 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();
let json: Value = serde_json::from_str(m).unwrap();
// Get the OP command, and its data
let op: &str = &json["op"].as_str().unwrap();
let data: &Value = &json["data"];
let user_operation: UserOperation = UserOperation::from_str(op).unwrap();
match user_operation {
UserOperation::Login => {
let login: Login = serde_json::from_str(&data.to_string()).unwrap();
ctx.state()
.addr
.do_send(login);
},
UserOperation::Register => {
let register: Register = serde_json::from_str(&data.to_string()).unwrap();
ctx.state()
.addr
.send(register)
.into_actor(self)
.then(|res, _, ctx| {
match res {
Ok(wut) => ctx.text(wut),
_ => println!("Something is wrong"),
}
fut::ok(())
})
.wait(ctx)
}
_ => ctx.text(format!("!!! unknown command: {:?}", m)),
}
// 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");
// ctx.state()
// .addr
// .send(ListRooms)
// .into_actor(self)
// .then(|res, _, ctx| {
// match res {
// Ok(rooms) => {
// for room in rooms {
// ctx.text(room);
// }
// }
// _ => println!("Something is wrong"),
// }
// fut::ok(())
// })
// .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(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
// ctx.state().addr.do_send(ClientMessage {
// id: self.id,
// msg: msg,
// room: self.room.clone(),
// })
// }
}
ws::Message::Binary(_bin) => println!("Unexpected binary"),
ws::Message::Close(_) => {
ctx.stop();
},
}
}
}
impl WSSession {
/// helper method that sends ping to client every second.
///
/// also this method checks heartbeats from client
fn hb(&self, ctx: &mut ws::WebsocketContext<Self, WsChatSessionState>) {
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
ctx.state()
.addr
.do_send(Disconnect { id: act.id });
// stop actor
ctx.stop();
// don't try to send a ping
return;
}
ctx.ping("");
});
}
}
fn main() {
let _ = env_logger::init();
let sys = actix::System::new("rust-reddit-fediverse-server");
// Start chat server actor in separate thread
let server = Arbiter::start(|_| ChatServer::default());
// Create Http server with websocket support
HttpServer::new(move || {
// Websocket sessions state
let state = WsChatSessionState {
addr: server.clone(),
};
App::with_state(state)
// redirect to websocket.html
// .resource("/", |r| r.method(http::Method::GET).f(|_| {
// HttpResponse::Found()
// .header("LOCATION", "/static/websocket.html")
// .finish()
// }))
// // websocket
.resource("/service/ws", |r| r.route().f(chat_route))
// static resources
// .handler("/static/", fs::StaticFiles::new("static/").unwrap())
}).bind("127.0.0.1:8080")
.unwrap()
.start();
println!("Started http server: 127.0.0.1:8080");
let _ = sys.run();
}

22
server/src/lib.rs

@ -1,7 +1,19 @@
#[macro_use]
extern crate diesel;
extern crate dotenv;
extern crate chrono;
pub extern crate diesel;
pub extern crate dotenv;
pub extern crate chrono;
pub extern crate serde;
pub extern crate serde_json;
pub extern crate actix;
pub extern crate actix_web;
pub extern crate rand;
pub extern crate strum;
#[macro_use] pub extern crate strum_macros;
pub mod schema;
pub mod apub;
pub mod actions;
pub mod websocket_server;
use diesel::*;
use diesel::pg::PgConnection;
@ -9,11 +21,7 @@ use diesel::result::Error;
use dotenv::dotenv;
use std::env;
pub mod schema;
pub mod apub;
pub mod actions;
// pub trait Likeable;
pub trait Crud<T> {
fn create(conn: &PgConnection, form: T) -> Result<Self, Error> where Self: Sized;
fn read(conn: &PgConnection, id: i32) -> Self;

1
server/src/websocket_server/mod.rs

@ -0,0 +1 @@
pub mod server;

269
server/src/websocket_server/server.rs

@ -0,0 +1,269 @@
//! `ChatServer` is an actor. It maintains list of connection client session.
//! And manages available rooms. Peers send messages to other peers in same
//! room through `ChatServer`.
use actix::prelude::*;
use rand::{rngs::ThreadRng, Rng};
use std::collections::{HashMap, HashSet};
use serde::{Deserialize, Serialize};
use {Crud,establish_connection};
#[derive(EnumString,ToString,Debug)]
pub enum UserOperation {
Login, Register, Logout, Join, Edit, Reply, Vote, Delete, NextPage, Sticky
}
pub enum MessageType {
Comments, Users, Ping, Pong
}
/// Chat server sends this messages to session
#[derive(Message)]
pub struct WSMessage(pub String);
/// Message for chat server communications
/// New chat session is created
#[derive(Message)]
#[rtype(usize)]
pub struct Connect {
pub addr: Recipient<WSMessage>,
}
/// Session is disconnected
#[derive(Message)]
pub struct Disconnect {
pub id: usize,
}
/// Send message to specific room
#[derive(Message)]
pub struct ClientMessage {
/// Id of the client session
pub id: usize,
/// Peer message
pub msg: String,
/// Room name
pub room: String,
}
/// List of available rooms
pub struct ListRooms;
impl actix::Message for ListRooms {
type Result = Vec<String>;
}
/// Join room, if room does not exists create new one.
#[derive(Message)]
pub struct Join {
/// Client id
pub id: usize,
/// Room name
pub name: String,
}
#[derive(Message)]
#[derive(Serialize, Deserialize)]
pub struct Login {
pub username: String,
pub password: String
}
// #[derive(Message)]
#[derive(Serialize, Deserialize)]
pub struct Register {
username: String,
email: Option<String>,
password: String,
password_verify: String
}
impl actix::Message for Register {
type Result = String;
}
/// `ChatServer` manages chat rooms and responsible for coordinating chat
/// session. implementation is super primitive
pub struct ChatServer {
sessions: HashMap<usize, Recipient<WSMessage>>, // A map from generated random ID to session addr
rooms: HashMap<String, HashSet<usize>>, // A map from room name to set of connectionIDs
rng: ThreadRng,
}
impl Default for ChatServer {
fn default() -> ChatServer {
// default room
let mut rooms = HashMap::new();
rooms.insert("Main".to_owned(), HashSet::new());
ChatServer {
sessions: HashMap::new(),
rooms: rooms,
rng: rand::thread_rng(),
}
}
}
impl ChatServer {
/// Send message to all users in the room
fn send_room_message(&self, room: &str, message: &str, skip_id: usize) {
if let Some(sessions) = self.rooms.get(room) {
for id in sessions {
if *id != skip_id {
if let Some(addr) = self.sessions.get(id) {
let _ = addr.do_send(WSMessage(message.to_owned()));
}
}
}
}
}
}
/// Make actor from `ChatServer`
impl Actor for ChatServer {
/// We are going to use simple Context, we just need ability to communicate
/// with other actors.
type Context = Context<Self>;
}
/// Handler for Connect message.
///
/// Register new session and assign unique id to this session
impl Handler<Connect> for ChatServer {
type Result = usize;
fn handle(&mut self, msg: Connect, _: &mut Context<Self>) -> Self::Result {
println!("Someone joined");
// notify all users in same room
self.send_room_message(&"Main".to_owned(), "Someone joined", 0);
// register session with random id
let id = self.rng.gen::<usize>();
self.sessions.insert(id, msg.addr);
// auto join session to Main room
self.rooms.get_mut(&"Main".to_owned()).unwrap().insert(id);
// send id back
id
}
}
/// Handler for Disconnect message.
impl Handler<Disconnect> for ChatServer {
type Result = ();
fn handle(&mut self, msg: Disconnect, _: &mut Context<Self>) {
println!("Someone disconnected");
let mut rooms: Vec<String> = Vec::new();
// remove address
if self.sessions.remove(&msg.id).is_some() {
// remove session from all rooms
for (name, sessions) in &mut self.rooms {
if sessions.remove(&msg.id) {
rooms.push(name.to_owned());
}
}
}
// send message to other users
for room in rooms {
self.send_room_message(&room, "Someone disconnected", 0);
}
}
}
/// Handler for Message message.
impl Handler<ClientMessage> for ChatServer {
type Result = ();
fn handle(&mut self, msg: ClientMessage, _: &mut Context<Self>) {
self.send_room_message(&msg.room, msg.msg.as_str(), msg.id);
}
}
/// Handler for `ListRooms` message.
impl Handler<ListRooms> for ChatServer {
type Result = MessageResult<ListRooms>;
fn handle(&mut self, _: ListRooms, _: &mut Context<Self>) -> Self::Result {
let mut rooms = Vec::new();
for key in self.rooms.keys() {
rooms.push(key.to_owned())
}
MessageResult(rooms)
}
}
/// Join room, send disconnect message to old room
/// send join message to new room
impl Handler<Join> for ChatServer {
type Result = ();
fn handle(&mut self, msg: Join, _: &mut Context<Self>) {
let Join { id, name } = msg;
let mut rooms = Vec::new();
// remove session from all rooms
for (n, sessions) in &mut self.rooms {
if sessions.remove(&id) {
rooms.push(n.to_owned());
}
}
// send message to other users
for room in rooms {
self.send_room_message(&room, "Someone disconnected", 0);
}
if self.rooms.get_mut(&name).is_none() {
self.rooms.insert(name.clone(), HashSet::new());
}
self.send_room_message(&name, "Someone connected", id);
self.rooms.get_mut(&name).unwrap().insert(id);
}
}
impl Handler<Login> for ChatServer {
type Result = ();
fn handle(&mut self, msg: Login, _: &mut Context<Self>) {
println!("{}", msg.password);
}
}
impl Handler<Register> for ChatServer {
type Result = MessageResult<Register>;
fn handle(&mut self, msg: Register, _: &mut Context<Self>) -> Self::Result {
use actions::user::*;
let conn = establish_connection();
// TODO figure out how to return values, and throw errors
// Register the new user
let user_form = UserForm {
name: &msg.username,
email: msg.email.as_ref().map(|x| &**x),
password_encrypted: &msg.password,
preferred_username: None,
updated: None
};
let inserted_user = User_::create(&conn, user_form).unwrap();
// Return the jwt
MessageResult("hi".to_string())
}
}
Loading…
Cancel
Save