diff --git a/errorcodes.yml b/errorcodes.yml index 4e99a41..d8507f2 100644 --- a/errorcodes.yml +++ b/errorcodes.yml @@ -1,3 +1,6 @@ +Not sure yet: + code: 11 + category: "don't know yet" Can't delete item 0: code: 100 @@ -31,3 +34,32 @@ could not read index of item: _commit not 8 bytes long: code: 105 category: "don't know yet" + +error while parsing int: + code: 106 + category: "parse_error" + +error with message sent to the server: + code: 107 + category: "faulty_message" + +missing metadata: + code: 108 + category: "faulty_message" + +something that should be utf8 valid was not: + code: 109 + category: "faulty_message" + +ParseIntError: + code: 110 + category: "faulty_message" + +_commit can't be updated: + code: 111 + category: "faulty_update" + +no / on the end of id: + code: 112 + category: "faulty_message" + diff --git a/first-render b/first-render index 51815cb..ace5146 160000 --- a/first-render +++ b/first-render @@ -1 +1 @@ -Subproject commit 51815cb28972b8fb0f905fa6442a13eb8e4ae639 +Subproject commit ace514689e36f2bcf64177df6231b77e5870ce24 diff --git a/js-client b/js-client index a3e42a9..1db0ffb 160000 --- a/js-client +++ b/js-client @@ -1 +1 @@ -Subproject commit a3e42a9e230b7ab1cb96cfaf480d14f240e75219 +Subproject commit 1db0ffbc31e1e449253f1b75b3c065afd43b774e diff --git a/src/error.rs b/src/error.rs index abd0078..13b46ac 100644 --- a/src/error.rs +++ b/src/error.rs @@ -1,4 +1,6 @@ +use std::string::FromUtf8Error; + use json; use crate::server::proto; @@ -19,11 +21,22 @@ impl MizeError { return data; } - pub fn to_message(self: MizeError) -> proto::Response { + pub fn to_message(self: MizeError, origin: proto::Origin) -> proto::Message { let data = self.to_json(); let mut msg = vec![crate::PROTO_VERSION, proto::MSG_ERROR]; msg.extend(json::stringify(data).into_bytes()); - return proto::Response::One(msg); + return proto::Message::from_bytes(msg, origin); + } +} + +impl From for MizeError { + fn from(utf_err: FromUtf8Error) -> MizeError{ + let err = MizeError { + code: 109, + kind: "faulty_message".to_string(), + message: "std::string::FromUtf8Error while parsing something from the message".to_string(), + }; + return err; } } diff --git a/src/integrated_types/async.rs b/src/integrated_types/async.rs new file mode 100644 index 0000000..e69de29 diff --git a/src/integrated_types/keep_same.rs b/src/integrated_types/keep_same.rs new file mode 100644 index 0000000..e69de29 diff --git a/src/integrated_types/sync.rs b/src/integrated_types/sync.rs new file mode 100644 index 0000000..e69de29 diff --git a/src/integrated_types/test_same.rs b/src/integrated_types/test_same.rs new file mode 100644 index 0000000..5806f2e --- /dev/null +++ b/src/integrated_types/test_same.rs @@ -0,0 +1,2 @@ + +pub fn main(){} diff --git a/src/lib.rs b/src/lib.rs index 399d253..f7d7655 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -5,6 +5,9 @@ static PROTO_VERSION: u8 = 1; pub mod server; pub mod error; +pub mod integrated_types { + pub mod test_same; +} diff --git a/src/server.rs b/src/server.rs index 6bfb618..2a83b28 100644 --- a/src/server.rs +++ b/src/server.rs @@ -4,23 +4,25 @@ pub mod server_utils; pub mod itemstore; pub mod proto; -use futures_util::{FutureExt, StreamExt}; +//use futures_util::lock::Mutex; +use futures_util::{FutureExt, StreamExt, SinkExt}; use warp::Filter; use warp::ws::{WebSocket, self}; +use std::collections::HashMap; -use tokio::sync::mpsc::{UnboundedSender, unbounded_channel}; -use tokio_stream::wrappers::UnboundedReceiverStream; +use tokio::sync::mpsc::{Sender, channel, Receiver}; +use tokio_stream::wrappers::ReceiverStream; use std::alloc::handle_alloc_error; use std::fs; use std::sync::{Arc}; use tokio::sync::Mutex; use tokio::sync::mpsc; -use crate::server::proto::{Response}; use crate::server::proto::handle_mize_message; use self::itemstore::Itemstore; //static API_PATH_STRING: &str = "$api"; +static SOCKET_CHANNEL_SIZE: usize = 200; // max file sizez in the mize/data folder (in bytes) static MAX_INDEX_FILE_SIZE: usize = 1_800_000; @@ -51,18 +53,58 @@ static MAX_DATA_FILE_SIZE: usize = 3_000_000_000; //static VERSION_MESSAGE: &str = "\ //Version: 0 //"; -// + +#[derive(Clone, Debug)] pub struct Client { - tx: UnboundedSender> +// rx: UnboundedReceiverStream>, + tx: Sender, + id: u64, } -pub fn run_server(args: Vec) { +#[derive(Clone, Debug)] +pub struct Module { + tx: Sender, + name: String, + client_id: u64, + //later + //registered_types: Vec, +} + +//pub struct Upstream maybe???? ... I'd say later.... + + +// A collection of all the things, that most functions need +#[derive(Clone)] +pub struct Mutexes { + next_free_client_id: Arc>, + clients: Arc>>, + modules: Arc>>, + subs: Arc>>>, + //maybe a list of upstream servers?? + itemstore: Arc>, +} + +impl Mutexes { + pub fn clone(mutexes: &Mutexes) -> Mutexes { + Mutexes { + next_free_client_id: Arc::clone(&mutexes.next_free_client_id), + clients: Arc::clone(&mutexes.clients), + modules: Arc::clone(&mutexes.modules), + subs: Arc::clone(&mutexes.subs), + itemstore: Arc::clone(&mutexes.itemstore), + } + } +} + +#[tokio::main] +pub async fn run_server(args: Vec) { /* * WHAT IT DOES * */ + // //### get the mize_folder let mut mize_folder = String::new(); for i in 1..args.len() { @@ -74,13 +116,27 @@ pub fn run_server(args: Vec) { } } + //create the itemstore + let itemstore = crate::server::itemstore::Itemstore::new(mize_folder.clone() + "/db").await.expect("error creating itemstore"); + + // A collection of all the things, that most functions need + let mutexes: Mutexes = Mutexes { + next_free_client_id: Arc::new(Mutex::new(0)), + clients: Arc::new(Mutex::new(Vec::new())), + subs: Arc::new(Mutex::new(HashMap::new())), + modules: Arc::new(Mutex::new(Vec::new())), + itemstore: Arc::new(Mutex::new(itemstore)), + }; + + //listen on the local unix socket + //local_socket_server(mize_folder); + //run the webserver - warp_server(mize_folder); + warp_server(mize_folder, mutexes).await; } -#[tokio::main] -async fn warp_server(mize_folder: String) { +async fn warp_server(mize_folder: String, mutexes: Mutexes) { /* * * @@ -92,35 +148,30 @@ async fn warp_server(mize_folder: String) { //## $api/rest/ // get itemstore - let itemstore = crate::server::itemstore::Itemstore::new(mize_folder + "/db").await.expect("error creating itemstore"); - let mut clients: Arc>> = Arc::new(Mutex::new(Vec::new())); - let mut itemstore_mutex: Arc> = Arc::new(Mutex::new(itemstore)); + let mutexes_clone = Mutexes::clone(&mutexes); let socket_route = warp::path!("$api" / "socket") .and(warp::ws()) .map(move |ws: warp::ws::Ws| { - let clients_clone = Arc::clone(&clients); - let itemstore_clone = Arc::clone(&itemstore_mutex); - ws.on_upgrade(move |socket| handle_socket_connection( - socket, - clients_clone, - itemstore_clone - ) - ) + let mutexes_clone = Mutexes::clone(&mutexes_clone); + ws.on_upgrade(move |socket| handle_websocket_connection(socket, mutexes_clone)) }); let render_route = warp::path!("$api" / "render").map(move || "render route"); let file_route = warp::path!("$api" / "file").map(move || "file route"); + //temporary let render_route = warp::path!("$api" / "render" / "first").map(move || { let answer = fs::read_to_string("/home/sebastian/work/mize/first-render/src/main.js").unwrap(); warp::http::Response::builder().header("content-type", "text/javascript").body(answer) }); + let main_route = warp::path!("$api" / "client" / "main.js").map(move || { let answer = fs::read_to_string("/home/sebastian/work/mize/js-client/src/main.js").unwrap(); warp::http::Response::builder().header("content-type", "application/javascript").body(answer) }); + let defuatl_route = warp::path::full().map(|path|{ //normally should be included in the binary, but for developing just gonna read the file //so I don't have to recompile all the time @@ -144,54 +195,96 @@ async fn warp_server(mize_folder: String) { warp::serve(routes).run(([127, 0, 0, 1], 3000)).await; } -async fn handle_socket_connection( +async fn handle_websocket_connection( socket: WebSocket, - clients_clone: Arc>>, - itemstore_clone: Arc>, + mutexes: Mutexes, ){ - let (socket_tx, mut socket_rx) = socket.split(); - let (client_tx, client_rx) = mpsc::unbounded_channel(); + let (mut socket_tx, mut socket_rx) = socket.split(); + let (msg_tx, msg_rx): (Sender, Receiver) = mpsc::channel(SOCKET_CHANNEL_SIZE); - let client_rx = UnboundedReceiverStream::new(client_rx); - tokio::spawn(client_rx.forward(socket_tx)); + let mut msg_rx = ReceiverStream::new(msg_rx); - let mut cli = clients_clone.lock().await; - cli.push(Client{tx: client_tx.clone()}); +// tokio::spawn(msg_rx.forward(socket_tx)); + + //my own forward + tokio::spawn(async move { + while let Some(msg) = msg_rx.next().await { + socket_tx.send(warp::ws::Message::binary(msg.raw)).await; + } + }); + + let mut cli = mutexes.clients.lock().await; + let mut client_id = mutexes.next_free_client_id.lock().await; + let client = Client{tx: msg_tx.clone(), id: *client_id}; + cli.push(client.clone()); + *client_id += 1; drop(cli); + drop(client_id); // Reading and broadcasting messages while let Some(result) = socket_rx.next().await { - println!("got message =================================="); - let msg = result.expect("Error when gettin message from WebSocket"); + let msg = result.expect("Error when getting message from WebSocket"); + println!("got message: {:?}", msg); - let itemstore = &*itemstore_clone.lock().await; - let responses = handle_mize_message(proto::Message::new(msg.clone().into_bytes()), itemstore).await; - for response in responses { - match response { - Response::One(response) => { - client_tx.send(Ok(ws::Message::binary(response.clone()))).unwrap() - }, - Response::All(response) => { - send_to_all_clients(ws::Message::binary(response), clients_clone.clone()).await; - }, - Response::AllSubbed(id, response) => { - send_to_all_subed_clients(id, ws::Message::binary(response), clients_clone.clone()).await; - }, - Response::None => {let t = 0;}, - }; + if let Err(err) = handle_mize_message( + proto::Message::from_bytes(msg.clone().into_bytes(),proto::Origin::Client(client.clone())), mutexes.clone(), + ).await { + msg_tx.send(err.to_message(proto::Origin::Client(client.clone()))).await; }; }; } -pub async fn send_to_all_clients(message: ws::Message, clients_clone: Arc>>){ - let clients = clients_clone.lock().await; - for client in &clients[..] { - client.tx.send(Ok(ws::Message::binary(message.clone()))); - } -} +//pub async fn handle_unix_socket_connection(){ +//} -pub async fn send_to_all_subed_clients(id: u64, message: ws::Message, clients: Arc>>){ -} +//async fn handle_connection(con: Connection){ +//} + +//pub async fn send_to_all_clients(message: ws::Message, mutexes_clone: Mutexes){ +// let clients = mutexes_clone.clients.lock().await; +// for client in &clients.clients[..] { +// client.tx.send(Ok(ws::Message::binary(message.clone()))); +// } +// drop(clients); + +// let modules = mutexes_clone.modules.lock().await; +// for module in &modules.modules[..] { +// module.tx.send(Ok(ws::Message::binary(message.clone()))); +// } +//} + +//pub async fn send_to_all_subbed_clients(id: String, message: ws::Message, mutexes_clone: Mutexes){ +// let clients = mutexes_clone.clients.lock().await; +// for client in &clients.clients[..] { +// if client.sub.contains(&id) { +// client.tx.send(Ok(ws::Message::binary(message.clone()))); +// } +// } + +// let modules = mutexes_clone.modules.lock().await; +// for module in &modules.modules[..] { +// if module.sub.contains(&id) { +// module.tx.send(Ok(ws::Message::binary(message.clone()))); +// } +// } +//} + +//that does not work like this + +//impl From for Result { +// fn from(msg: proto::Message) -> Result { +// let msg = warp::ws::Message::binary(msg.raw); +// } +//} + +//impl From> for proto::Message { +// fn from(ws_msg: Result) -> proto::Message { +// let msg = ws_msg.expect("Error in From trait implementation: warp::ws::Messate to proto::Message"); +// } +//} +//191 | cli.push(Client{tx: client_tx, id: client_id}); +// = note: expected struct `tokio::sync::mpsc::Sender` +// found struct `tokio::sync::mpsc::Sender<` diff --git a/src/server/itemstore.rs b/src/server/itemstore.rs index 0c875aa..bebfe77 100644 --- a/src/server/itemstore.rs +++ b/src/server/itemstore.rs @@ -5,7 +5,13 @@ use surrealdb::{Datastore, Key, Val}; use itertools::Itertools; use crate::error; use crate::error::MizeError; +use crate::server::proto; +use super::proto::Delta; + +// The struct to do the storage and updates to it +// is responsible, that no illegal states can occur in the storage, by using transactions +// currently it's just a surrealdb Datastore. should be replaced by a completely custom File-System in the future. pub struct Itemstore { db: surrealdb::Datastore, } @@ -94,57 +100,61 @@ impl Itemstore { return Ok(new_id) } - pub async fn update(&self, id: u64, new_item: Item) -> Result<(), MizeError>{ + pub async fn update(&self, update: proto::Update) -> Result<(), MizeError>{ + let mut tr = self.db.transaction(true, true).await?; - let mut keys: Vec> = Vec::new(); + let old_item_index = tr.get(update.id.clone().into_bytes()).await?; + let old_item = for key in old_item_index { + let mut key_get = format!("{}:", update.id).into_bytes(); + key_get.extend(key.clone()); + tr.get(key_get); + }; - for field in new_item { - let mut key = format!("{}:", id).into_bytes(); - key.extend(field[0].clone()); - let val = field[1].clone(); - keys.push(field[0].clone()); - tr.set(key, val).await?; + let mut keys: Vec> = Vec::new(); + + //check the _type of the item and maybe run type-code + //TODO: typecode + + //read keys + let mut keys = match tr.get(format!("{}", update.id).into_bytes()).await? { + Some(keys) => decode(keys), + None => { + return Err(MizeError{ + code: 104, + kind: "data_storage::index_not_found".to_string(), + message: "Internal Datastorage Error: the Index of the Item was not found".to_string(), + }); + }, + }; + + //write new values + for (mut key, delta) in update.raw { + let mut store_key = (update.id.clone() + ":").into_bytes(); + store_key.append(&mut key); + let mut new_val = Vec::new(); + + if let Some(old_val) = tr.get(store_key.clone()).await? { + new_val = proto::apply_delta(old_val, delta)?; + tr.set(store_key.clone(), new_val.clone()).await?; + } else { + keys.push(key.clone()); + new_val = proto::apply_delta(Vec::new(), delta)?; + } + + //remove field if new_val is empty + if (new_val.len() == 0) { + if let Some(key_index) = keys.iter().position(|x| *x == key.clone()){ + keys.remove(key_index); + } + tr.del(key).await?; + } else { + tr.set(store_key.clone(), new_val).await?; + } } - let keys_res = tr.get(format!("{}", id).into_bytes()).await?; - if let Some(old_keys) = keys_res { - keys.extend(decode(old_keys)); - keys = keys.into_iter().unique().collect::>>(); - tr.set(format!("{}", id).into_bytes(), encode(keys)).await?; - } else { - return Err(MizeError{ - code: 104, - kind: "data_storage::index_not_found".to_string(), - message: "Internal Datastorage Error: the Index of the Item was not found".to_string(), - }); - } - - //increment commit number - let mut commit_key = format!("{}:_commit", id).into_bytes(); - - let commit_res = tr.get(&*commit_key).await?; - if let Some(commit_vec) = commit_res { - let val: [u8;8] = match commit_vec.try_into() { - Ok(val) => val, - Err(_) => { - return Err(MizeError{ - code: 105, - kind: "don't know yet".to_string(), - message: "_commit of this item is no valid u64 (not 8 bytes long)".to_string(), - }); - }, - }; - let mut commit_num = u64::from_be_bytes(val); - commit_num += 1; - tr.set(commit_key, commit_num.to_be_bytes()).await?; - } else { - return Err(MizeError{ - code: 103, - kind: "key_missing::_commit".to_string(), - message: "This item has no _commit key, which every item must have.".to_string(), - }); - } + //write keys again + tr.set(update.id.into_bytes(), encode(keys)).await?; tr.commit().await?; return Ok(()); @@ -152,10 +162,21 @@ impl Itemstore { - pub async fn delete(&self, id: u64) -> Result<(), error::MizeError>{ + pub async fn delete(&self, id_str: String) -> Result<(), error::MizeError>{ + let id_u64: u64 = match id_str.parse() { + Ok(id) => id, + Err(err) => { + return Err(MizeError { + code: 110, + kind: "faulty_message".to_string(), + message: format!("std::num::ParseIntError while parsing id \"{}\" into an Integer", id_str), + + }); + } + }; let mut tr = self.db.transaction(true, true).await?; - let keys_res = tr.get(format!("{}", id).into_bytes()).await?; + let keys_res = tr.get(format!("{}", id_u64).into_bytes()).await?; if let Some(keys) = keys_res { for key in decode(keys) { tr.del(key).await?; @@ -168,7 +189,7 @@ impl Itemstore { }); } - tr.del(format!("{}", id)).await?; + tr.del(id_str.into_bytes()).await?; tr.commit().await?; return Ok(()); diff --git a/src/server/proto.rs b/src/server/proto.rs index b630e44..e1c7a9f 100644 --- a/src/server/proto.rs +++ b/src/server/proto.rs @@ -1,5 +1,4 @@ - // 1: get // 2: give (by default only give the first 1k bytes, and the first 100 fields) // 3: get_val (to get more than 1k bytes) @@ -17,6 +16,16 @@ // 15: get_and_sub // 16: get_fields // 17: error + // 18: give_fields + + + // ID + // just a number: localy storeWEK_K2JNQqkd + // #hello.mize.works#034.312 upstream in the internet ... domain specified + // !youtube!WEK_K2JNQqk upstream in a module ... if module is not on local server + // forwarded to configured upstream server (would happen, if we are running in daemon mode) + +pub static VERSION: u8 = 1; pub static MSG_GET: u8 = 1; pub static MSG_GIVE: u8 = 2; @@ -35,32 +44,221 @@ pub static MSG_UNSUPPORTED_VERSION: u8 = 14; pub static MSG_GET_AND_SUB: u8 = 15; pub static MSG_GET_FIELDS: u8 = 16; pub static MSG_ERROR: u8 = 17; +pub static MSG_GIVE_FIELDS: u8 = 18; - - +use std::time::Duration; //use std::panic::update_hook; use std::vec; +use std::collections::HashMap; use crate::server::itemstore; +use crate::server::Mutexes; use crate::server::itemstore::encode; use crate::error; +use crate::server::Client; +use std::sync::Arc; +use futures_util::stream::Collect; +use tokio::sync::{Mutex, mpsc}; +use crate::server; +use crate::error::MizeError; -pub enum Response { - One(Vec), - All(Vec), - AllSubbed(u64, Vec), - None, -} - +#[derive(Clone)] pub struct Message { - msg: Vec, + pub raw: Vec, version: u8, cmd: u8, + origin: Origin, index: usize, + meta_gotten: bool, + id: Option, +} + +//pub enum Response { +// One(Vec), +// All(Vec), +// AllSubbed(String, Vec), +// None, +//} + +#[derive(Clone, Debug)] +pub enum Origin { + Client(server::Client), //the client id + Module(server::Module), //Module_name + Upstream(String), //a hostname Type, but for now just a string +} + +impl Origin { + pub async fn send(&self, message: Message){ + match self { + Origin::Client(client) => {client.tx.send(message).await;}, + Origin::Module(module) => {module.tx.send(message).await;}, + Origin::Upstream(_) => {}, + } + } + pub fn get_id(&self) -> u64{ + match self { + Origin::Client(client) => client.id, + Origin::Module(module) => module.client_id, + Origin::Upstream(_) => 0, + } + } +} + +#[derive(Clone)] +pub struct Update { + pub id: String, + origin: Origin, + pub raw: Vec<(Vec, Delta)>, +} + +#[derive(Clone, Debug)] +pub struct Delta { + raw: Vec, +} + +#[derive(Clone, Debug)] +pub enum DeltaCmd { + Replace(u32, u32, Vec), + Insert(u32, u32, Vec), + Delete(u32, u32), +} + +impl Delta { + pub fn from_bytes(bytes: Vec) -> Result { + let out_of_bytes = Err(MizeError{ + code: 107, + kind: "faulty_message".to_string(), + message: "There were not eneough Bytes in the message to complete a Delta.".to_string() + }); + let mut bytes = bytes.iter(); + let mut delta = Delta{raw: Vec::new()}; + + while true { + let cmd = if let Some(cmd) = bytes.next() { + cmd + } else { + //end of bytes + return Ok(delta); + }; + + match cmd { + 0 => { + let mut start_bytes: [u8; 4] = [0,0,0,0]; + for i in 0..4 {start_bytes[i] = if let Some(byte) = bytes.next(){*byte} else {return out_of_bytes;}}; + let start = u32::from_be_bytes(start_bytes); + + let mut stop_bytes: [u8; 4] = [0,0,0,0]; + for i in 0..4 {stop_bytes[i] = if let Some(byte) = bytes.next(){*byte} else {return out_of_bytes;}}; + let stop = u32::from_be_bytes(stop_bytes); + + let mut val: Vec = Vec::new(); + for i in 0..stop-start {val.push(if let Some(byte) = bytes.next(){*byte} else {return out_of_bytes;})}; + + delta.raw.push(DeltaCmd::Replace(start, stop, val)); + }, + 1 => { + let mut start_bytes: [u8; 4] = [0,0,0,0]; + for i in 0..4 {start_bytes[i] = if let Some(byte) = bytes.next(){*byte} else {return out_of_bytes;}}; + let start = u32::from_be_bytes(start_bytes); + + let mut stop_bytes: [u8; 4] = [0,0,0,0]; + for i in 0..4 {stop_bytes[i] = if let Some(byte) = bytes.next(){*byte} else {return out_of_bytes;}}; + let stop = u32::from_be_bytes(stop_bytes); + + let mut val: Vec = Vec::new(); + for i in 0..stop-start {val.push(if let Some(byte) = bytes.next(){*byte} else {return out_of_bytes;})}; + + delta.raw.push(DeltaCmd::Insert(start, stop, val)); + }, + 2 => { + let mut start_bytes: [u8; 4] = [0,0,0,0]; + for i in 0..4 {start_bytes[i] = if let Some(byte) = bytes.next(){*byte} else {return out_of_bytes;}}; + let start = u32::from_be_bytes(start_bytes); + + let mut stop_bytes: [u8; 4] = [0,0,0,0]; + for i in 0..4 {stop_bytes[i] = if let Some(byte) = bytes.next(){*byte} else {return out_of_bytes;}}; + let stop = u32::from_be_bytes(stop_bytes); + + delta.raw.push(DeltaCmd::Delete(start, stop)); + }, + _ => { + return Err(MizeError{ + code: 107, + kind: "faulty_message".to_string(), + message: format!("The update message included a DeltaCmd of {}, only 0,1,2 (for: replace, insert, delete) are allowed.",cmd) + }); + }, + } + }; + return Ok(delta); + } +} + +impl Update { + pub async fn change_whole_val(id: String, key: Vec, new_val: Vec, origin: Origin, mutexes: Mutexes) -> Result { + let mut old_val_len: u32 = 0; + let itemstore = mutexes.itemstore.lock().await; + + let id_u64 = match id.parse::(){ + Ok(id) => id, + Err(err) => { + return Err(MizeError{ + code: 11, + kind: "don't know yet".to_string(), + message: "Error while parsing id to u64 (std::num::ParseIntError) in order to get it from the itemstore. (get or get_sub message)".to_string(), + }); + } + }; + + let item = itemstore.get(id_u64).await?; + + for field in item { + if (field[0] == key){ + old_val_len = field[1].len() as u32; + } + } + + let delete_all = DeltaCmd::Delete(0, old_val_len); + let replace_all = DeltaCmd::Replace(0, new_val.len() as u32, new_val); + let deltas = Delta{raw: vec![delete_all, replace_all]}; + let update: Update = Update { id, raw: vec![(key, deltas)], origin}; + return Ok(update); + } + + pub fn to_message(self, origin: Origin) -> Message{ + return Message::from_update(self, origin); + } + + pub fn from_message(mut message: Message) -> Result { + + message.index = 2; + message.id = None; + + let id = message.get_id()?; + + let mut update = Update{id, raw: Vec::new(), origin: message.clone().origin}; + + let num_of_updates = message.get_u32(); + + for i in 0..num_of_updates as usize { + let key_len = message.get_u32(); + let key = message.get_bytes(key_len as usize); + + let update_len = message.get_u32(); + let update_bytes = message.get_bytes(update_len as usize); + + //get DeltaCmds + let delta = Delta::from_bytes(update_bytes)?; + + update.raw.push((key, delta)); + } + return Ok(update); + + } } impl Message { - pub fn new(vec: Vec) -> Message { + pub fn from_bytes(vec: Vec, origin: Origin) -> Message { // this weired version does not panic, while others do for some unknown reason let mut count = 0; @@ -77,17 +275,214 @@ impl Message { } count += 1; + + if count == 2 {break}; }; - Message { msg: vec, version: version, cmd: cmd, index: 2 } + Message { raw: vec, version, cmd, index: 2 , origin, meta_gotten: false, id: None} + } + pub async fn send(self, origin: &Origin) { + origin.send(self).await; } - pub fn get_id(&mut self) -> u64 { - self.get_u64() + pub async fn forward_module(mut self, id: String, mutexes: Mutexes) -> Result<(), MizeError>{ + let mut meta: HashMap = HashMap::new(); + meta.insert("c".to_string(), format!("{}", self.origin.get_id())); + self.set_metadata(meta)?; + let mut id_iter = id.chars(); + //scip the first "!" + id_iter.next(); + let mod_name = id_iter.take_while(|&ch| ch != '!').collect::(); + + let modules = mutexes.modules.lock().await; + for module in &*modules { + if (module.name == mod_name) { + self.send(&Origin::Module(module.clone())); + return Ok(()); + } + } + return Ok(()); + } + + pub fn from_update(update: Update, origin: Origin) -> Message{ + let mut message: Vec = vec![VERSION, MSG_UPDATE]; + + // id + message.extend(update.id.into_bytes()); + message.push("/".to_string().into_bytes()[0]); + + // update_len + let update_len = update.raw.len() as u32; + message.extend(update_len.to_be_bytes()); + + for update in update.raw { + //key_len + let key_len = update.0.len() as u32; + message.extend(key_len.to_be_bytes()); + + // key + message.extend(update.0); + + //delta_len + let mut delta_len: u32 = 0; + + // deltas + let mut delta_part: Vec = Vec::new(); + + // loop + for delta in update.1.raw { + match delta { + DeltaCmd::Replace(start, stop, data) => { + delta_len += 9; //for len of start, stop and cmd + delta_len += data.len() as u32; + delta_part.push(0); + delta_part.extend(start.to_be_bytes()); + delta_part.extend(stop.to_be_bytes()); + delta_part.extend(data) + }, + DeltaCmd::Insert(start, stop, data) => { + delta_len += 9; //for len of start, stop and cmd + delta_len += data.len() as u32; + delta_part.push(1); + delta_part.extend(start.to_be_bytes()); + delta_part.extend(stop.to_be_bytes()); + delta_part.extend(data) + }, + DeltaCmd::Delete(start, stop) => { + delta_len += 9; //for len of start, stop and cmd + // + delta_part.push(2); + delta_part.extend(start.to_be_bytes()); + delta_part.extend(stop.to_be_bytes()); + }, + } + }; + + message.extend(delta_len.to_be_bytes()); + message.extend(delta_part); + + } + return Message::from_bytes(message, origin); + } + + pub fn set_metadata(&mut self, metadata: HashMap) -> Result<(), MizeError> { + let mut meta_vec: Vec = Vec::new(); + meta_vec.push('{' as u8); + + for (mut key, mut val) in metadata.iter(){ + meta_vec.append(&mut key.clone().into_bytes()); + meta_vec.push(';' as u8); + meta_vec.append(&mut val.clone().into_bytes()); + meta_vec.push('/' as u8); + } + meta_vec.remove(meta_vec.len()); + meta_vec.push('}' as u8); + + self.raw.splice(2..2, meta_vec); + return Ok(()); + } + + pub fn get_metadata(&mut self) -> Result, MizeError> { + if !self.has_metadata() { + return Err(MizeError { + code: 108, + kind: "faulty_message".to_string(), + message: "The message should have medatata, but it hasn't. (cmd <= 128)".to_string(), + }) + }; + + let mut meta_vec:Vec = Vec::new(); + let mut msg_iter = self.raw[self.index..].iter(); + + if (msg_iter.nth(0).unwrap() != &('{' as u8)){ + return Err(MizeError { + code: 107, + kind: "faulty_message".to_string(), + message: "Error while parsing the metadata of a message.".to_string(), + }) + }; + + for ch in msg_iter { + meta_vec.push(*ch); + + if (ch == &('}' as u8)){ + let mut map: HashMap = HashMap::new(); + let string = String::from_utf8(meta_vec)?; + let sp: Vec<&str> = string.split("/").collect(); + for st in sp.clone() { + let mut split = sp.split(|c| *c == ";"); + let key = split.nth(0).unwrap()[0]; + let val = split.nth(1).unwrap()[0]; + map.insert(key.to_string(), val.to_string()); + } + self.meta_gotten = true; + return Ok(map); + } + }; + + return Err(MizeError { + code: 107, + kind: "faulty_message".to_string(), + message: "Error while parsing the metadata of a message.".to_string(), + }) + + } + + pub fn has_metadata(&mut self) -> bool { + self.cmd >= 128 + } + + pub fn get_id(&mut self) -> Result { + //get the metadata just to increment the index + if let Some(id) = &self.id { + return Ok(id.clone()); + } + if (self.has_metadata() && !self.meta_gotten){ + self.get_metadata(); + } + let mut ch: u8 = 0; + let mut id = String::new(); + for ch in &self.raw[self.index..] { + + if (*ch == 47 as u8){ + self.index += 1; + self.id = Some(id.clone()); + return Ok(id); + } else { + id.push(*ch as char); + self.index += 1; + } + } + + return Err(MizeError { + code: 112, + kind: "faulty_message".to_string(), + message: "There is no / to indicate where the id ends".to_string(), + }); + +// let len = self.raw.len(); +// while self.index < 10000 { +// if (self.index == len){ +// return Err(MizeError { +// code: 112, +// kind: "faulty_message".to_string(), +// message: "There is no / to indicate where the id ends".to_string(), +// }) +// } +// let ch: u8 = self.raw[self.index]; +// if (ch == 47) {break}; + +// self.index += 1; +// } + //skip "/" +// self.index += 1; + +// self.id_gotten = true; +// return Ok(id); } pub fn get_u32(&mut self) -> u32 { - let tmp = &self.msg[self.index..self.index + 4]; + let tmp = &self.raw[self.index..self.index + 4]; let num: u32 = u32::from_be_bytes([tmp[0], tmp[1], tmp[2], tmp[3]]); self.index += 4; @@ -95,7 +490,7 @@ impl Message { } pub fn get_u64(&mut self) -> u64 { - let tmp = &self.msg[self.index..self.index + 8]; + let tmp = &self.raw[self.index..self.index + 8]; let num: u64 = u64::from_be_bytes([tmp[0], tmp[1], tmp[2], tmp[3], tmp[4], tmp[5], tmp[6], tmp[7]]); self.index += 8; @@ -103,7 +498,7 @@ impl Message { } pub fn get_bytes(&mut self, n: usize) -> Vec{ - let tmp = &self.msg[self.index..self.index + n]; + let tmp = &self.raw[self.index..self.index + n]; self.index += n; return tmp.to_vec(); @@ -111,283 +506,318 @@ impl Message { } -const VERSION: u8 = 1; pub async fn handle_mize_message( mut message: Message, - itemstore: &itemstore::Itemstore, - ) -> Vec { - - let old_message = message.msg.clone(); + mutexes: Mutexes, + ) -> Result<(), MizeError> { - //println!("message: {:?}", message); - //let version = message.clone().into_iter().nth(0).expect("message has no 0th byte"); - //let cmd = message.clone().into_iter().nth(1).expect("message has no 1th byte"); - //println!("after chaos"); + //////////////////////////////////////////////////////////// + //get and get_and_sub cmd from client or module + if ((message.cmd == MSG_GET || message.cmd == MSG_GET_AND_SUB) && + (matches!(&message.origin, Origin::Client(_)) || matches!(&message.origin, Origin::Module(_)))) + { - // for some weired reason this stupid solution does not panic, while the obove one and vec[0] - // do - let mut count = 0; + let id: String = message.get_id()?; + let client_id: u64 = match message.origin { + Origin::Client(ref Client) => Client.id, + Origin::Module(ref Module) => Module.client_id, + Origin::Upstream(_) => 0, + }; - let mut cmd: u8 = 0; - let mut version: u8 = 0; + //sub to item in case of MSG_GET_AND_SUB + if (message.cmd == MSG_GET_AND_SUB) { + let mut subs = mutexes.subs.lock().await; + if let Some(mut vec_ids) = subs.get_mut(&id) { + vec_ids.push(message.origin.clone()); + } else { + subs.insert(id.clone(), vec![message.origin.clone()]); + }; + drop(subs); + }; - for i in old_message.clone() { - if count == 0 { - version = i; - } - if count == 1 { - cmd = i + //forward to module in case the item is from a Module + //append medatata + let first_char = id.chars().nth(0).ok_or(MizeError{ + code: 11, + kind: "don't know yet".to_string(), + message: "Somehow the id is empty. while handling a get or get_sub message.".to_string(), + })?; + + if (first_char == '!'){ + message.clone().forward_module(id.clone(), mutexes.clone()).await; } - count += 1; + //forward to upstream server in case the item is from another Server + if (first_char == '#'){ + return Ok(()); + } + + + // answer: + // u8: version + // u8: cmd (2 for give) + // id: terminated by "/" + // u32: num_of_fields + // as often as num_of_fields: + // u64: key_len + // key_len: key + // u64: val_len + // val_len: val + let mut answer: Vec = vec![VERSION, MSG_GIVE]; + answer.extend(format!("{}/", id).into_bytes()); + + let itemstore = mutexes.itemstore.lock().await; + + let id_u64 = match id.parse::(){ + Ok(id) => id, + Err(err) => { + return Err(MizeError{ + code: 11, + kind: "don't know yet".to_string(), + message: "Error while parsing id to u64 in order to get it from the itemstore. (get or get_sub message)".to_string(), + }); + } + }; + + let mut item = match itemstore.get(id_u64).await { + Ok(item) => item, + Err(err) => {return Err(err)}, + }; + + let num_of_fields = item.len() as u32; + answer.extend(num_of_fields.to_be_bytes()); + + for field in item { + let key_len = field[0].len() as u32; + answer.extend(key_len.to_be_bytes()); + answer.extend(field[0].clone()); + + let val_len = field[1].len() as u32; + answer.extend(val_len.to_be_bytes()); + answer.extend(field[1].clone()); + } + + let msg = Message::from_bytes(answer, message.origin.clone()); + message.origin.send(msg).await; }; - //println!("VERSION: {}", version); - //println!("CMD: {}", cmd); - match cmd { - //get - 1 => { - //let id_bytes = *&message[2..9].to_owned().clone(); - let tmp = &old_message[2..10]; - let id: u64 = u64::from_be_bytes([tmp[0], tmp[1], tmp[2], tmp[3], tmp[4], tmp[5], tmp[6], tmp[7]]); + //////////////////////////////////////////////////////////// + //give from module + if (message.cmd == MSG_GIVE && matches!(&message.origin, Origin::Module(_))){ + //TODO + }; - // answer: - // u8: version - // u8: cmd (2 for give) - // u64: id - // u32: num_of_fields - // as often as num_of_fields: - // u64: key_len - // key_len: key - // u64: val_len - // val_len: val - let mut answer: Vec = vec![VERSION, 2]; - answer.extend(id.to_be_bytes()); - let mut item = match itemstore.get(id).await { - Ok(item) => item, - Err(err) => {return vec![err.to_message()]}, + //////////////////////////////////////////////////////////// + //update_request from module and client + if (message.cmd == MSG_UPDATE_REQUEST && (matches!(&message.origin, Origin::Module(_)) || matches!(&message.origin, Origin::Client(_)))){ + + let id: String = message.get_id()?; + let first_char = id.chars().nth(0).ok_or(MizeError{ + code: 11, + kind: "don't know yet".to_string(), + message: "Somehow the id is empty. while handling a get or get_sub message.".to_string(), + })?; + + if (first_char == '!'){ + message.clone().forward_module(id, mutexes.clone()).await; + } + + let update = Update::from_message(message.clone())?; + handle_update(update, mutexes, message.origin.clone()).await?; + return Ok(()); + }; + + + //////////////////////////////////////////////////////////// + //delete from Client or Module + if (message.cmd == MSG_DELETE && (matches!(&message.origin, Origin::Module(_)) || matches!(&message.origin, Origin::Client(_)))){ + let id: String = message.get_id()?; + if id == "0".to_string() { + let err = error::MizeError { + kind: "don't know yet".to_string(), + code: 100, + message: "You can't delete item number 0. This item contains mandatory Config for the Server.".to_string(), }; - - let num_of_fields = item.len() as u32; - answer.extend(num_of_fields.to_be_bytes()); - - for field in item { - let key_len = field[0].len() as u32; - answer.extend(key_len.to_be_bytes()); - answer.extend(field[0].clone()); - - let val_len = field[1].len() as u32; - answer.extend(val_len.to_be_bytes()); - answer.extend(field[1].clone()); - } - return vec![Response::One(answer)]; - }, - 2 => {return Vec::new()}, - 3 => {return Vec::new()}, - 4 => {return Vec::new()}, - 5 => {return Vec::new()}, - 6 => {return Vec::new()}, - 7 => {return Vec::new()}, - - //update_request - 8 => { - let mut answer: Vec = message.msg.clone(); - answer[1] = 10; - - let id = message.get_id(); - - let mut item = match itemstore.get(id).await { - Ok(item) => item, - Err(err) => {return vec![err.to_message()]}, - }; - - let num_of_updates = message.get_u32(); - - for i in 0..num_of_updates as usize { - let key_len = message.get_u32(); - let key = message.get_bytes(key_len as usize); - //println!("KEY: {}", String::from_utf8(key.clone()).expect("here utf-8")); - - let update_len = message.get_u32(); - let update = message.get_bytes(update_len as usize); - - let mut found = false; - for a in 0..item.len() as usize { - if item[a][0] == key { - item[a][1] = apply_update(&item[a][1], &update); - found = true; - break; - } - } - if found == false { - let index = item.len() +1; - //item[index][1] = apply_update(&item[index][1], &update) - item.push([key.clone(), apply_update(&item[index][1], &update)]) - } - } - - if let Err(err) = itemstore.update(id, item).await { - return vec![err.to_message()]; - }; - - return vec![Response::All(answer)]; - }, - 9 => {return Vec::new()}, - 10 => {return Vec::new()}, - - //delete - 11 => { - let response: Vec = Vec::new(); - let tmp = &old_message[2..10]; - let id: u64 = u64::from_be_bytes([tmp[0], tmp[1], tmp[2], tmp[3], tmp[4], tmp[5], tmp[6], tmp[7]]); - if id == 0 { - let err = error::MizeError{ - kind: "don't know yet".to_string(), - code: 100, - message: "You can't delete item number 0. This item contains mandatory Config for the Server.".to_string(), - }; - return vec![err.to_message()]; - } - - if let Err(err) = itemstore.delete(id).await { - return vec![err.to_message()]; - } - return response; - }, - - //create - 12 => { - let num_of_fields = message.get_u32(); - let mut item: Vec<[Vec; 2]> = Vec::new(); - - let mut index = 6; - - for i in 0..num_of_fields { - let key_len = message.get_u32(); - let key = message.get_bytes(key_len as usize); - - let val_len = message.get_u32(); - let val = message.get_bytes(val_len as usize); - - item.push([key, val]); - } - - if let Err(err) = itemstore.create(item).await { - return vec![err.to_message()]; - }; - - let anser: Vec = vec![1,1]; - return vec![Response::None]; - }, - 13 => {return Vec::new()}, - 14 => {return Vec::new()}, - _ => {return Vec::new()}, + return Err(err); + } } + //////////////////////////////////////////////////////////// + //create from module and Client + if (message.cmd == MSG_CREATE && (matches!(&message.origin, Origin::Module(_)) || matches!(&message.origin, Origin::Client(_)))){ + + let num_of_fields = message.get_u32(); + let mut item: Vec<[Vec; 2]> = Vec::new(); + + let mut index = 6; + + for i in 0..num_of_fields { + let key_len = message.get_u32(); + let key = message.get_bytes(key_len as usize); + + let val_len = message.get_u32(); + let val = message.get_bytes(val_len as usize); + + item.push([key, val]); + } + + let itemstore = mutexes.itemstore.lock().await; + let new_id = itemstore.create(item).await?; + + let mut item0: Vec<[Vec; 2]> = Vec::new(); + match itemstore.get(0).await { + Err(err) => {return Err(err);}, + Ok(item) => {item0 = item;}, + } + + //update item0 + let mut next_free_id: Vec = Vec::new(); + let mut num_of_items: Vec = Vec::new(); + for field in item0 { + if field[0] == "next_free_id".to_string().into_bytes() { + // soooooo ugly + let mut tmp = u64::from_be_bytes(field[1].clone().try_into().unwrap()); + tmp += 1; + next_free_id = tmp.to_be_bytes().to_vec(); + + } + if field[0] == "num_of_items".to_string().into_bytes() { + // soooooo ugly + let mut tmp = u64::from_be_bytes(field[1].clone().try_into().unwrap()); + tmp += 1; + num_of_items = tmp.to_be_bytes().to_vec(); + } + } + + //because Update::change_whole_val trys to lock the itemstore as well + drop(itemstore); + + let update_next_id: Update = Update::change_whole_val("0".to_string(), "next_free_id".to_string().into_bytes(), next_free_id.clone(), message.origin.clone(), mutexes.clone()).await?; + let mut update: Update = Update::change_whole_val("0".to_string(), "num_of_items".to_string().into_bytes(), num_of_items.clone(), message.origin.clone(), mutexes.clone()).await?; + + update.raw.extend(update_next_id.raw); + + handle_update(update, mutexes.clone(), message.origin.clone()).await?; + + //send back MSG_CREATED_ID + let mut created_id_message = vec![VERSION, MSG_CREATED_ID]; + created_id_message.extend(new_id.to_be_bytes()); + message.origin.send(Message::from_bytes(created_id_message, message.origin.clone())).await; + } + + //////////////////////////////////////////////////////////// + //for every other message + return Ok(()); +} + +pub async fn handle_update(mut update: Update, mutexes: Mutexes, origin: Origin) -> Result<(), MizeError>{ + //handle type-code + //and either call itemstore.update(), send the update to the module or upstream, or spawn another + //update by calling handle_update() + + + ///////////////////////////////// TYPE CODE //////////////////////////////////////// + //can change the update and spawn new updates (to a different item) (which would call the handle_update func again) + + //the "type code" for the Null type (code that effects every item) + //eg. increment _commit + let itemstore = mutexes.itemstore.lock().await; + + let id_u64 = match update.id.clone().parse::(){ + Ok(id) => id, + Err(err) => { + return Err(MizeError{ + code: 11, + kind: "don't know yet".to_string(), + message: "Error while parsing id to u64 in order to get it from the itemstore. (get or get_sub message)".to_string(), + }); + } + }; + let item = itemstore.get(id_u64).await?; + let commit_pos = item.iter().position(|field| field[0] == "_commit".to_string().into_bytes()).ok_or(MizeError{ + code: 103, + kind: "key_missing::_commit".to_string(), + message: format!("Item {} has not _commit", id_u64), + })?; + let cur_commit = u64::from_be_bytes(item[commit_pos][1].clone().try_into().map_err(|_|MizeError{ + code: 105, + kind: "don't know yet".to_string(), + message: "_commit is not 8 bytes long".to_string(), + })?); + + //because Update::change_whole_val trys to lock the itemstore as well + drop(itemstore); + + + update.raw.extend(Update::change_whole_val(update.id.clone(), "_commit".to_string().into_bytes(), (cur_commit +1).to_be_bytes().to_vec(), update.origin.clone(), mutexes.clone()).await?.raw); + + + //run some type-code here +// if let Some(index) = item.iter().position(|&field| field[0] == "_type".to_string().into_bytes()) { +// let type_string = String::from_utf8(item[index][0])?; +// let types = type_string.split(" ").collect::Vec(); +// for &typ[..] in types { +// match typ { +// "test-same" => crate::integrated_types::test_same::main(update, item); +// } +// }; +// }; + + //if update is changing _type, then run type::create(), or type::destroy(), for types that got added or deleted + + //apply item in itemstore + let itemstore = mutexes.itemstore.lock().await; + itemstore.update(update.clone()).await?; + + //send update to clients/modules + let message = update.clone().to_message(origin.clone()); + let subs = mutexes.subs.lock().await; + + let empty_vec = &Vec::new(); + let subbed_origins = subs.get(&update.id).unwrap_or(empty_vec); + for origin in subbed_origins { + message.clone().send(origin).await; + } + + Ok(()) } // just like all of this crate, definetly could be done with less clones().... and better error // handling -pub fn apply_update(val: &Vec, updates: &Vec) -> Vec{ - println!("VAL: {:?}", val.clone()); - println!("Update: {:?}", updates.clone()); - let not_enough_bytes = "not enough bytes in update"; - let mut val = val.clone(); - let mut update_iter = updates.clone().into_iter(); - while true { - if let Some(operation) = update_iter.next(){ - let mut new_val: Vec = Vec::new(); - let mut val_iter = val.clone().into_iter(); - match operation { - //r,start:u32,stop:u32,bytes start..stop - 0 => { - //get start and stop - let mut start_bytes: [u8; 4] = [0,0,0,0]; - for i in 0..4 {start_bytes[i] = update_iter.next().expect(not_enough_bytes);}; - let start = u32::from_be_bytes(start_bytes); - let mut stop_bytes: [u8; 4] = [0,0,0,0]; - for i in 0..4 {stop_bytes[i] = update_iter.next().expect(not_enough_bytes);}; - let stop = u32::from_be_bytes(stop_bytes); +pub fn apply_delta(mut val: Vec, delta: Delta) -> Result, MizeError>{ - //add the stuff before - for i in 0..start {new_val.push(val_iter.next().expect(not_enough_bytes));}; + let mut update_iter = delta.raw.iter(); + for delta_cmd in delta.raw.iter() { + match delta_cmd { + //r,start:u32,stop:u32,bytes start..stop + DeltaCmd::Replace(start, stop, bytes) => { + let bytes = (*bytes).clone(); + let start = *start as usize; + let stop = *stop as usize; + let val_len = val.len(); + if (stop > val_len) { + val.extend(vec![0; stop-val_len]) + } + val.splice(start..stop, bytes); + }, + //i,start:u32, stop:u32, bytes stop-start + DeltaCmd::Insert(start, stop, bytes) => { + val.splice((*start as usize)..(*start as usize), (*bytes).clone()); + }, + //d,start:u32,stop:u32 + DeltaCmd::Delete(start, stop) => { + val.splice((*start as usize)..(*stop as usize), Vec::new()); + }, + _ => {panic!("unknown update command")} + } + } - //add the new stuff - for i in start..stop {new_val.push(update_iter.next().expect(not_enough_bytes));}; - - //skip all the bytes that should be replaced - for i in 0..stop-start {val_iter.next().expect(not_enough_bytes);} - - //add stuff after - for byte in val_iter { - new_val.push(byte); - } - }, - //i,start:u32, stop:u32, bytes stop-start - 1 => { - //get start and stop - let mut start_bytes: [u8; 4] = [0,0,0,0]; - for i in 0..4 {start_bytes[i] = update_iter.next().expect(not_enough_bytes);}; - let start = u32::from_be_bytes(start_bytes); - - let mut stop_bytes: [u8; 4] = [0,0,0,0]; - for i in 0..4 {stop_bytes[i] = update_iter.next().expect(not_enough_bytes);}; - let stop = u32::from_be_bytes(stop_bytes); - - //add the stuff before - for i in 0..start {new_val.push(val_iter.next().expect(not_enough_bytes));}; - - //add the new stuff - for i in start..stop {new_val.push(update_iter.next().expect(not_enough_bytes));}; - - //add stuff after - for byte in val_iter { - new_val.push(byte); - } - - //while true { - //if let Some(byte) = update_iter.next() { - //new_val.push(byte); - //} else {break;} - //} - }, - //d,start:u32,stop:u32 - 2 => { - //get start and stop - let mut start_bytes: [u8; 4] = [0,0,0,0]; - for i in 0..4 {start_bytes[i] = update_iter.next().expect(not_enough_bytes);}; - let start = u32::from_be_bytes(start_bytes); - - let mut stop_bytes: [u8; 4] = [0,0,0,0]; - for i in 0..4 {stop_bytes[i] = update_iter.next().expect(not_enough_bytes);}; - let stop = u32::from_be_bytes(stop_bytes); - - //add the stuff before - for i in 0..start {new_val.push(val_iter.next().expect("not_enough_bytes in val_iter"));}; - - //skip all the bytes that should be replaced - for i in 0..stop-start {let h = val_iter.next().expect("not_enough_bytes in val_iter"); println!("VAL ITER: {}", h);} - - //add stuff after - for byte in val_iter { - new_val.push(byte); - } - }, - _ => {panic!("unknown update command")} - } - val = new_val; - } else {break;} - }; - return val; + return Ok(val); } - - - -