diff --git a/src/event_processor.rs b/src/event_processor.rs index 600f10f..9542ab0 100644 --- a/src/event_processor.rs +++ b/src/event_processor.rs @@ -1,52 +1,21 @@ -use std::borrow::Borrow; - -use std::fs::File; - -use std::path::PathBuf; +use rusqlite::{Connection, Transaction}; +use crate::events::{Bogus, CreateNewUser, Event}; use crate::util::{CustomResult, Error}; -use rusqlite::Connection; -use serde::{Deserialize, Serialize}; - -#[derive(Deserialize, Serialize, Debug, PartialEq)] -struct CacheChequeEvent { - account: i64, -} - -#[derive(Deserialize, Serialize, Debug, PartialEq)] -enum Event { - CacheCheque(CacheChequeEvent), -} - -fn process_event_file(_conn: &mut Connection, _file: File) -> CustomResult<()> { - return Ok(()); -} - -pub fn process_event_files(conn: &mut Connection, paths: I) -> CustomResult<()> -where - I: Iterator, - I::Item: Borrow, -{ - for path in paths { - let file = File::open(path.borrow()).map_err(Error::IOError)?; - process_event_file(conn, file)?; - } +fn process_bogus_event(tx: &Transaction, event: Bogus) -> CustomResult<()> { Ok(()) } -mod test { - use tempfile::tempfile; - use super::*; - - - +fn process_create_new_user_event(tx: &Transaction, event: CreateNewUser) -> CustomResult<()> { + Ok(()) +} - #[test] - fn test_basic() -> CustomResult<()> { - let file = tempfile().unwrap(); - let mut conn = Connection::open_in_memory().unwrap(); - let ret = process_event_file(&mut conn, file); - ret + +pub fn process_event(conn: &mut Connection, event: Event) -> CustomResult<()> { + let tx = conn.transaction().map_err(Error::SqlError)?; + match event { + Event::Bogus(e) => process_bogus_event(&tx, e), + Event::CreateNewUser(e) => process_create_new_user_event(&tx, e), } } diff --git a/src/events.rs b/src/events.rs new file mode 100644 index 0000000..06f4776 --- /dev/null +++ b/src/events.rs @@ -0,0 +1,20 @@ +use serde::Deserialize; + + +#[derive(Deserialize, Debug)] +pub struct Bogus { + pub name: String, + pub age: i64, +} + +#[derive(Deserialize, Debug)] +pub struct CreateNewUser { + pub name: String, + pub score: i64, +} + +#[derive(Deserialize, Debug)] +pub enum Event { + Bogus(Bogus), + CreateNewUser(CreateNewUser), +} \ No newline at end of file diff --git a/src/main.rs b/src/main.rs index 462b719..857fc6e 100644 --- a/src/main.rs +++ b/src/main.rs @@ -10,7 +10,6 @@ use simple_logger::SimpleLogger; use db::securities::list_securities; use crate::db::users::{create_user, list_users}; -use crate::event_processor::process_event_files; use crate::migration::apply_migrations; use crate::util::{CustomResult, Error}; use crate::webui::web_ui_main; @@ -21,6 +20,7 @@ mod migration; mod services; mod util; mod webui; +mod events; #[derive(Debug, Serialize)] struct Output { @@ -516,17 +516,18 @@ impl CommandLineUserLogic for ProCommandLineLogic { conn: &mut Connection, files: ValuesRef, ) -> CustomResult<()> { - process_event_files(conn, files) + // process_event_files(conn, files) + Ok(()) } fn run_webui( &mut self, - _conn: &mut Connection, + conn: &mut Connection, port: u16, enable_template_autoreload: bool, ) -> CustomResult<()> { info!("Serving web-ui on port {:?}", port); - web_ui_main(port, enable_template_autoreload) + web_ui_main(conn, port, enable_template_autoreload) } } diff --git a/src/util.rs b/src/util.rs index ddb24dc..678e958 100644 --- a/src/util.rs +++ b/src/util.rs @@ -11,6 +11,7 @@ pub enum Error { DoubleSpendingError(String), IOError(io::Error), UserInputError(String), + FuckingError(String), Testing(String), } diff --git a/src/webui/mod.rs b/src/webui/mod.rs index b1ff9fd..ae3420e 100644 --- a/src/webui/mod.rs +++ b/src/webui/mod.rs @@ -4,19 +4,21 @@ use std::{env, path::PathBuf}; use actix_files; use actix_web::{ + App, dev::ServiceResponse, error, - http::{header::ContentType, StatusCode}, - middleware::{ErrorHandlerResponse, ErrorHandlers, Logger}, - web, App, FromRequest, HttpRequest, HttpResponse, HttpServer, Responder, Result, + FromRequest, + http::{header::ContentType, StatusCode}, HttpRequest, HttpResponse, HttpServer, middleware::{ErrorHandlerResponse, ErrorHandlers, Logger}, Responder, Result, web, }; use actix_web_lab::respond::Html; use async_channel; -use futures::{pin_mut, select, FutureExt}; +use futures::{FutureExt, pin_mut, select}; use minijinja; use minijinja_autoreload::AutoReloader; +use rusqlite::Connection; -use crate::util::CustomResult; +use crate::events::Event; +use crate::util::{CustomResult, Error}; use crate::webui::worker::worker_loop; mod views; @@ -57,7 +59,9 @@ impl FromRequest for MiniJinjaRenderer { } #[tokio::main] -pub async fn web_ui_main(port: u16, enable_template_autoreload: bool) -> CustomResult<()> { +pub async fn web_ui_main(conn: &mut Connection, + port: u16, + enable_template_autoreload: bool) -> CustomResult<()> { // The closure is invoked every time the environment is outdated to recreate it. let tmpl_reloader = AutoReloader::new(move |notifier| { let mut env: minijinja::Environment<'static> = minijinja::Environment::new(); @@ -73,7 +77,7 @@ pub async fn web_ui_main(port: u16, enable_template_autoreload: bool) -> CustomR Ok(env) }); - let (s, r) = async_channel::unbounded::(); + let (s, r) = async_channel::unbounded::(); let tmpl_reloader = web::Data::new(tmpl_reloader); @@ -98,14 +102,13 @@ pub async fn web_ui_main(port: u16, enable_template_autoreload: bool) -> CustomR .unwrap(); let t1 = server.run().fuse(); - let t2 = worker_loop(r).fuse(); + let t2 = worker_loop(conn, r).fuse(); pin_mut!(t1, t2); select! { - x = t1 => x, + x = t1 => x.map_err(Error::IOError), y = t2 => y, - }; - Ok(()) + } } /// Error handler for a 404 Page not found error. diff --git a/src/webui/views.rs b/src/webui/views.rs index 0bfa217..36d683a 100644 --- a/src/webui/views.rs +++ b/src/webui/views.rs @@ -1,12 +1,12 @@ +use std::collections::HashMap; -use actix_web::{web, HttpRequest, Responder}; +use actix_web::{error, HttpRequest, Responder, web}; use clap::builder::TypedValueParser; use log::info; use minijinja::context; - use serde::Deserialize; -use std::collections::HashMap; +use crate::events::Event; use crate::webui::MiniJinjaRenderer; #[actix_web::get("/")] @@ -52,23 +52,6 @@ pub async fn actions_partials( tmpl_env.render(template_location.as_str(), ctx) } -#[derive(Deserialize, Debug)] -struct Event1 { - name: String, - age: i64, -} - -#[derive(Deserialize, Debug)] -struct Event2 { - name: String, - score: i64, -} - -#[derive(Deserialize, Debug)] -enum Event { - Event1(Event1), - Event2(Event2), -} #[derive(Deserialize, Debug)] struct CreateEventPayload { @@ -80,14 +63,16 @@ struct CreateEventPayload { pub async fn push_event( req: HttpRequest, tmpl_env: MiniJinjaRenderer, - event_channel: web::Data>, + event_channel: web::Data>, web::Json(form): web::Json, ) -> actix_web::Result { - let shit = req.match_info().get("event_type").unwrap(); - println!("payload: {:?}", form); - let _ = event_channel.send(56).await; + let path_event_type = req.match_info().get("event_type").unwrap(); + if path_event_type.to_string() != form.event_type { + return Err(error::ErrorBadRequest("Inconsistent event type.")) + } + let _ = event_channel.send(form.event).await; let ctx = context! { - event_type => shit, + event_type => "bob le chef", }; tmpl_env.render("partials/v1/event_processed.html", ctx) } diff --git a/src/webui/worker.rs b/src/webui/worker.rs index 9de653d..9540263 100644 --- a/src/webui/worker.rs +++ b/src/webui/worker.rs @@ -1,13 +1,21 @@ use std::io; -pub async fn worker_loop(r: async_channel::Receiver) -> io::Result<()> { +use rusqlite::Connection; + +use crate::event_processor::process_event; +use crate::events::Event; +use crate::util::CustomResult; +use crate::util::Error::FuckingError; + +pub async fn worker_loop(conn: &mut Connection, + r: async_channel::Receiver) -> CustomResult<()> { loop { - let x = r.recv().await; - match x { - Ok(v) => println!("Processed value: {:?}", v), + let res = r.recv().await; + match res { + Ok(event) => process_event(conn, event)?, // todo, remove ? Err(e) => { println!("Error: {:?}", e); - return Ok(()); + return Err(FuckingError(e.to_string())); } } }