harmony/harsh
1
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

implemented channel queries

Browse source
This commit is contained in:
mb 2022-08-22 11:25:49 +03:00
parent 62788c1b26
commit 28a3812812
10 changed files with 208 additions and 67 deletions
Download patch file Download diff file Expand all files Collapse all files

59
harsh-server/src/gateway.rs
View file

@ -1,9 +1,7 @@
use harsh_common::{Ping, Pong, ServerRequest};
use harsh_common::{client, server, ClientRequest, ServerRequest};
use telecomande::{Processor, Remote};
use harsh_common::ClientRequest;
use crate::{sessions, Addr, SessionProc, StorageProc};
use crate::{Addr, Id, SessionCmd, SessionProc, StorageCmd, StorageProc};
#[derive(Debug)]
pub enum GatewayCmd {
@ -12,26 +10,50 @@ pub enum GatewayCmd {
}
pub struct GatewayProc {
client_handler: Remote<SessionProc>,
sessions: Remote<SessionProc>,
storage: Remote<StorageProc>,
}
impl GatewayProc {
pub fn new(client_handler: Remote<SessionProc>, storage: Remote<StorageProc>) -> Self {
Self {
client_handler,
storage,
}
pub fn new(sessions: Remote<SessionProc>, storage: Remote<StorageProc>) -> Self {
Self { sessions, storage }
}
async fn handle_request(&mut self, address: Addr, request: ClientRequest) {
match request {
ClientRequest::Ping(Ping { content }) => {
ClientRequest::Ping(client::Ping { content }) => {
println!("received ping! '{content:?}'");
let response = ServerRequest::Pong(Pong { content });
let content = response.serialize();
self.client_handler
.send(sessions::SessionCmd::Send(address, content))
let request = ServerRequest::Pong(server::Pong { content });
self.sessions
.send(SessionCmd::new_send(address, request))
.unwrap();
}
ClientRequest::ChannelList(client::ChannelList {}) => {
let (cmd, rec) = StorageCmd::new_channel_list();
self.storage.send(cmd).unwrap();
let channels = rec.await.unwrap().iter().map(|id| id.to_u64()).collect();
let request = ServerRequest::new_channel_list(channels);
self.sessions
.send(SessionCmd::new_send(address, request))
.unwrap();
}
ClientRequest::ChannelCreate(client::ChannelCreate { name }) => {
let (cmd, rec) = StorageCmd::new_channel_create(name);
let _id = rec.await.unwrap();
self.storage.send(cmd).unwrap();
}
ClientRequest::ChannelDelete(client::ChannelDelete { channel_id }) => {
self.storage
.send(StorageCmd::ChannelDelete(Id::from_u64(channel_id)))
.unwrap();
}
ClientRequest::ChannelGetName(client::ChannelGetName { channel_id }) => {
let (cmd, rec) = StorageCmd::new_channel_get_name(Id::from_u64(channel_id));
self.storage.send(cmd).unwrap();
let name = rec.await.unwrap();
let request = ServerRequest::new_channel_get_name(channel_id, name);
self.sessions
.send(SessionCmd::new_send(address, request))
.unwrap();
}
}
@ -46,14 +68,15 @@ impl Processor for GatewayProc {
match command {
GatewayCmd::Request(address, request) => {
if let Some(request) = ClientRequest::try_parse(&request) {
println!("[session/info] received command '{request:?}'");
self.handle_request(address, request).await;
} else {
println!("failed to parse command");
println!("[session/warn] failed to parse command");
}
}
GatewayCmd::ClosedConnection(address) => self
.client_handler
.send(sessions::SessionCmd::RemoveSession(address))
.sessions
.send(SessionCmd::RemoveSession(address))
.unwrap(),
}
Ok(())

20
harsh-server/src/main.rs
View file

@ -4,20 +4,22 @@ use tokio::net::TcpListener;
#[tokio::main]
async fn main() {
println!("starting server ...");
let client_handler = SimpleExecutor::new(SessionProc::default()).spawn();
let storage = SimpleExecutor::new(StorageProc::new("./db")).spawn();
println!("[main/info] starting server ...");
let sessions = SimpleExecutor::new(SessionProc::default()).spawn();
println!("[main/info] spawned sessions");
let storage = SimpleExecutor::new(StorageProc::new("/tmp/db.test")).spawn();
println!("[main/info] spawned storage");
let gateway =
SimpleExecutor::new(GatewayProc::new(client_handler.remote(), storage.remote())).spawn();
println!("spawned gateway");
SimpleExecutor::new(GatewayProc::new(sessions.remote(), storage.remote())).spawn();
println!("[main/info] spawned gateway");
let listener = TcpListener::bind("localhost:8080").await.unwrap();
println!("listening on 'localhost:8080' ...");
println!("[main/info] listening on 'localhost:8080' ...");
let client_handler = client_handler.remote();
let client_handler = sessions.remote();
loop {
let (stream, address) = listener.accept().await.unwrap();
println!("new connection from '{address:?}'");
println!("[main/info] new connection from '{address:?}'");
client_handler
.send(sessions::SessionCmd::AddSession(
@ -38,5 +40,5 @@ pub use gateway::{GatewayCmd, GatewayProc};
mod sessions;
pub use sessions::{SessionCmd, SessionProc};
pub use storage::{StorageCmd, StorageProc};
mod storage;
pub use storage::{StorageCmd, StorageProc};

10
harsh-server/src/sessions.rs
View file

@ -1,5 +1,6 @@
use std::{collections::HashMap, net::SocketAddr};
use harsh_common::ServerRequest;
use telecomande::{Processor, Remote};
use tokio::{
io::{AsyncBufReadExt, AsyncWriteExt, BufReader},
@ -18,6 +19,13 @@ pub enum SessionCmd {
Send(Addr, String),
}
impl SessionCmd {
pub fn new_send(address: Addr, request: ServerRequest) -> Self {
let content = request.serialize();
Self::Send(address, content)
}
}
#[derive(Debug, Default)]
pub struct SessionProc {
clients: HashMap<Addr, (OwnedWriteHalf, JoinHandle<()>)>,
@ -69,7 +77,7 @@ async fn session(address: Addr, reader: OwnedReadHalf, remote: Remote<gateway::G
loop {
let mut line = String::new();
if let Err(error) = reader.read_line(&mut line).await {
eprintln!("{error}");
eprintln!("[session/error] {error}");
break;
}
remote

18
harsh-server/src/storage.rs
View file

@ -8,23 +8,23 @@ use crate::Id;
pub enum StorageCmd {
ChannelCreate(String, Sender<Id>),
ChannelDelete(Id),
ChannelGetAll(Sender<Vec<Id>>),
ChannelList(Sender<Vec<Id>>),
ChannelGetName(Id, Sender<Option<String>>),
}
impl StorageCmd {
fn new_channel_create(name: impl ToString) -> (Self, Receiver<Id>) {
pub fn new_channel_create(name: impl ToString) -> (Self, Receiver<Id>) {
let (s, r) = oneshot::channel();
(Self::ChannelCreate(name.to_string(), s), r)
}
fn new_channel_delete(id: Id) -> Self {
pub fn new_channel_delete(id: Id) -> Self {
Self::ChannelDelete(id)
}
fn new_channel_get_all() -> (Self, Receiver<Vec<Id>>) {
pub fn new_channel_list() -> (Self, Receiver<Vec<Id>>) {
let (s, r) = oneshot::channel();
(Self::ChannelGetAll(s), r)
(Self::ChannelList(s), r)
}
fn new_channel_get_name(id: Id) -> (Self, Receiver<Option<String>>) {
pub fn new_channel_get_name(id: Id) -> (Self, Receiver<Option<String>>) {
let (s, r) = oneshot::channel();
(Self::ChannelGetName(id, s), r)
}
@ -93,7 +93,7 @@ impl Processor for StorageProc {
self.set(format!("/channels/{id}"), item);
sender.send(id).unwrap();
}
StorageCmd::ChannelGetAll(sender) => {
StorageCmd::ChannelList(sender) => {
let results = self.list("/channels/");
sender.send(results).unwrap();
}
@ -170,7 +170,7 @@ async fn test_channels() {
let id = rec.await.unwrap();
// query all
let (cmd, rec) = StorageCmd::new_channel_get_all();
let (cmd, rec) = StorageCmd::new_channel_list();
remote.send(cmd).unwrap();
let result = rec.await.unwrap();
assert_eq!(result.len(), 1);
@ -189,7 +189,7 @@ async fn test_channels() {
let id2 = rec.await.unwrap();
// query all
let (cmd, rec) = StorageCmd::new_channel_get_all();
let (cmd, rec) = StorageCmd::new_channel_list();
remote.send(cmd).unwrap();
let result = rec.await.unwrap();
assert_eq!(result.len(), 2);

8
harsh-server/src/utils.rs
View file

@ -27,6 +27,14 @@ impl Id {
let inner: u64 = input.parse().ok()?;
Some(Self(inner))
}
pub fn from_u64(input: u64) -> Self {
Self(input)
}
pub fn to_u64(&self) -> u64 {
self.0
}
}
#[test]