ppc/mize
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

there is now a thread where updates are sent to, instead of only a function that gets called

Browse Source
This commit is contained in:
Sebastian Moser
2023-04-17 18:13:44 +02:00
parent 7823682944
commit 79f58e671e
4 changed files with 99 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
src/error.rs
View File

@@ -216,20 +216,35 @@ impl<T> MizeResultExtension<T> for Result<T, MizeError> {
}
impl From<FromUtf8Error> for MizeError {
#[track_caller]
fn from(utf_err: FromUtf8Error) -> MizeError{
MizeError::new(109).extra_msg("std::string::FromUtf8Error while parsing something from the message")
let caller_location = std::panic::Location::caller();
let mut err = MizeError::new(109).extra_msg(format!("From std::string::FromUtf8Error: {}", utf_err));
err.code_location = Some(caller_location.into());
return err;
}
}
impl From<surrealdb::error::Db> for MizeError {
#[track_caller]
fn from(sur_err: surrealdb::error::Db) -> MizeError {
MizeError::new(129).extra_msg(format!("surrealdb Error: {}", sur_err))
let caller_location = std::panic::Location::caller();
let mut err = MizeError::new(129).extra_msg(format!("From surrealdb Error: {}", sur_err));
err.code_location = Some(caller_location.into());
return err;
}
}
impl From<serde_json::Error> for MizeError {
#[track_caller]
fn from(serde_err: serde_json::Error) -> MizeError {
MizeError::new(11).extra_msg("From an serde_json::Error. TODO: include actual error msg.")
let caller_location = std::panic::Location::caller();
let mut err = MizeError::new(11).extra_msg(format!("From an serde_json::Error: {}", serde_err));
err.code_location = Some(caller_location.into());
return err;
}
}

25
src/server.rs
View File

@@ -15,6 +15,8 @@ use crate::error::MizeError;
use crate::error::ERRORS;
use crate::server::proto::MizeMessage;
use crate::server::proto::MizeId;
use crate::server::proto::Update;
use crate::server::proto::update_thread;
use crate::error::MizeResultExtension;
use serde_json::Value as JsonValue;
@@ -47,6 +49,7 @@ use tower_http::services::{ServeDir, ServeFile};
//static API_PATH_STRING: &str = "$api";
static SOCKET_CHANNEL_SIZE: usize = 200;
static UPDATE_CHANNEL_SIZE: usize = 200;
// max file sizez in the mize/data folder (in bytes)
static MAX_INDEX_FILE_SIZE: usize = 1_800_000;
@@ -178,6 +181,7 @@ pub struct Mutexes {
itemstore: Arc<Mutex<Itemstore>>,
mize_folder: String,
server_uuid: Uuid,
update_tx: mpsc::Sender<(Update, Peer)>,
}
impl Mutexes {
@@ -191,6 +195,7 @@ impl Mutexes {
mize_folder: mutexes.mize_folder.clone(),
itemstore: Arc::clone(&mutexes.itemstore),
server_uuid: mutexes.server_uuid,
update_tx: mutexes.update_tx.clone(),
}
}
}
@@ -217,6 +222,8 @@ pub async fn run_server(args: Vec<String>) {
let (server_uuid, itemstore, renders) = init_server(mize_folder.clone()).await.extra_msg("Could not init server").handle().is_critical();
// the channel to send updates to, that are then handeled in another thread
let (update_tx, update_rx) = mpsc::channel::<(Update, Peer)>(UPDATE_CHANNEL_SIZE);
// Collection of all the things, that most functions need
let mutexes: Mutexes = Mutexes {
@@ -228,8 +235,15 @@ pub async fn run_server(args: Vec<String>) {
itemstore: Arc::new(Mutex::new(itemstore)),
mize_folder: mize_folder.clone(),
server_uuid,
update_tx,
};
// the update handling task
let update_mutexes = mutexes.clone();
tokio::spawn(async move {
proto::update_thread(update_rx, update_mutexes);
});
//listen on the local unix socket
//local_socket_server(mize_folder);
@@ -250,8 +264,6 @@ async fn axum_server(mize_folder: String, mutexes: Mutexes) {
//## $api/socket/id
//## $api/rest/
let mutexes_clone = Mutexes::clone(&mutexes);
let serve_client = get_service(ServeDir::new("js-client/src")).handle_error(handle_error);
@@ -341,7 +353,7 @@ async fn handle_websocket_connection(
let mut msg_rx = ReceiverStream::new(msg_rx);
//my own forward
// my own forward
tokio::spawn(async move {
while let Some(msg) = msg_rx.next().await {
match msg {
@@ -364,7 +376,8 @@ async fn handle_websocket_connection(
}
});
//check if mize-module header exists
// check if mize-module header exists
let origin = if let Some(mod_header_val) = headers.get("mize-module"){
let mut mods = mutexes.modules.lock().await;
@@ -410,7 +423,6 @@ async fn handle_websocket_connection(
msg_tx.send(proto::MizeMessage::Json(err.to_json_message())).await;
continue;
};
match msg {
Message::Binary(b) => {
println!("Recieved a Binary Message. Those are not implemented yet.")
@@ -418,12 +430,13 @@ async fn handle_websocket_connection(
Message::Text(text) => {
println!("GOT: {}", text);
let json_msg: proto::JsonMessage = serde_json::from_str(&text).expect("error parsing json");
if let Err(err) = proto::handle_json_msg(json_msg, origin.clone(), mutexes.clone()).await {
let err_msg: MizeMessage = err.handle().into();
origin.send(err_msg).await;
};
}
},
Message::Close(_) => {
match origin {

16
src/server/itemstore.rs
View File

@@ -7,6 +7,7 @@ use crate::error;
use crate::error::MizeError;
use crate::server::proto;
use crate::server::Mutexes;
use crate::server::Peer;
use serde_json::Value as JsonValue;
use serde::{Serialize, Deserialize};
@@ -29,7 +30,7 @@ pub struct Item {
impl Item {
pub fn get_id(&self) -> Result<String, MizeError> {
if let JsonValue::Object(ob) = &self.json{
if let Some(id_val) = ob.get("mize-id") {
if let Some(id_val) = ob.get("__item__") {
if let JsonValue::String(id) = id_val {
return Ok(id.to_string())
} else {
@@ -119,7 +120,7 @@ impl Itemstore {
return Ok(Itemstore {db:ds});
}
pub async fn create(&self, mut item: Item, mutexes: Mutexes) -> Result<u64, MizeError> {
pub async fn create(&self, mut item: Item, mutexes: Mutexes, origin: Peer) -> Result<u64, MizeError> {
let mut tr = self.db.transaction(true, false).await?;
//get next_free_id
@@ -135,7 +136,9 @@ impl Itemstore {
let new_num = num_of_items + 1;
//add mize-id field to item
let delta: Delta = serde_json::from_str(&format!("[[[\"mize-id\"], \"{}\"]]", new_id))?;
//let delta: Delta = serde_json::from_str(&format!("[[[\"__item__\"], \"{}\"]]", new_id))?;
let mut delta = Delta::new();
delta.append(vec!["__item__"], serde_json::json!(new_id));
item.apply_delta(delta)?;
tr.set(new_id.to_be_bytes(), item.json.to_string()).await?;
@@ -143,8 +146,11 @@ impl Itemstore {
tr.commit().await?;
//handle update to item0
let delta_item0: Delta = serde_json::from_str(&format!("[[[\"num_of_items\"], {}], [[\"next_free_id\"], {}]]", new_num, new_next_id))?;
proto::handle_update(vec![(proto::MizeId::new(format!("0")), delta_item0)], mutexes.clone(), None).await?;
let mut delta_item0: Delta = Delta::new();
delta_item0.append(vec!["num_of_items"], serde_json::json!(new_num));
delta_item0.append(vec!["next_free_id"], serde_json::json!(new_next_id));
proto::handle_update(vec![(proto::MizeId::new(format!("0")), delta_item0)], origin, mutexes.clone()).await?;
return Ok(new_id);
}

55
src/server/proto.rs
View File

@@ -160,12 +160,13 @@ pub struct UpdateMessage {
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct Delta {
#[serde(flatten)]
pub delta: Vec<(Path, JsonValue)>,
}
type Path = Vec<String>;
pub type Path = Vec<String>;
type Update = Vec<(MizeId, Delta)>;
pub type Update = Vec<(MizeId, Delta)>;
#[derive(Debug, Default, Clone, Serialize, Deserialize, Eq, Hash, PartialEq)]
@@ -228,6 +229,17 @@ impl MizeId {
}
impl Delta {
pub fn new() -> Delta{
Delta { delta: Vec::new() }
}
pub fn append(&mut self, path: Vec<&str>, json: JsonValue){
let new_path: Vec<String> = path.into_iter().map(|e| e.to_owned()).collect();
self.delta.push((new_path, json));
}
}
//###//===================================================
//From trait implementations, to be able to treat everything as a MizeMessage that can be
@@ -324,29 +336,29 @@ pub async fn handle_json_msg(msg: JsonMessage, origin: Peer, mutexes: Mutexes) -
JsonMessage::UpdateRequest(mut msg) => {
let update: Update = vec![(msg.id, msg.delta)];
let new_update = handle_update(update , mutexes.clone(), Some(origin)).await?;
handle_update(update, origin, mutexes.clone()).await?;
let subs = mutexes.subs.lock().await;
// let subs = mutexes.subs.lock().await;
for (id, delta) in new_update {
// for (id, delta) in new_update {
//send that delta to every origin that is subbed the item
let peers = match subs.get(&id) {
Some(vec) => vec,
None => continue,
};
let message = UpdateMessage{id, delta};
// let peers = match subs.get(&id) {
// Some(vec) => vec,
// None => continue,
// };
// let message = UpdateMessage{id, delta};
for peer in peers {
// for peer in peers {
//TODO: send should only take a reference
peer.send(message.clone()).await
}
}
// peer.send(message.clone()).await
// }
// }
},
JsonMessage::Create(msg) => {
let itemstore = mutexes.itemstore.lock().await;
let id = itemstore.create(msg.item, mutexes.clone()).await?;
let id = itemstore.create(msg.item, mutexes.clone(), origin.clone()).await?;
let response = CreatedIdMessage {id: MizeId::new(format!("{}", id))};
origin.send(response).await;
@@ -373,20 +385,25 @@ pub async fn handle_json_msg(msg: JsonMessage, origin: Peer, mutexes: Mutexes) -
return Ok(())
}
pub async fn handle_update(update: Update, origin: Peer, mutexes: Mutexes) -> Result<(), MizeError> {
mutexes.update_tx.send((update, origin));
Ok(())
}
pub async fn handle_update(update: Update, mutexes: Mutexes, origin: Option<Peer>) -> Result<Update, MizeError> {
pub async fn update_thread(mut update_rx: mpsc::Receiver<(Update, Peer)>, mutexes: Mutexes) -> Result<(), MizeError> {
while let Some((update, origin)) = update_rx.recv().await {
//TODO: run update code from modules and types
let itemstore = mutexes.itemstore.lock().await;
for (id, delta) in update.clone() {
for (id, delta) in update {
if let MizeIdKind::Local(id) = id.kind()? {
itemstore.update(id, delta).await?;
} else {
return Err(MizeError::new(11).extra_msg("updates to non local items are not handeld yet"));
}
}
Ok(update)
}
Ok(())
}