Started to implement proper event processing.

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<I>(conn: &mut Connection, paths: I) -> CustomResult<()>
-where
-    I: Iterator,
-    I::Item: Borrow<PathBuf>,
-{
-    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),
     }
 }

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),
+}

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<T> {
@@ -516,17 +516,18 @@ impl CommandLineUserLogic for ProCommandLineLogic {
         conn: &mut Connection,
         files: ValuesRef<PathBuf>,
     ) -> 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)
     }
 }
 

src/util.rs

@@ -11,6 +11,7 @@ pub enum Error {
     DoubleSpendingError(String),
     IOError(io::Error),
     UserInputError(String),
+    FuckingError(String),
     Testing(String),
 }
 

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::<i32>();
+    let (s, r) = async_channel::unbounded::<Event>();
 
     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.

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<async_channel::Sender<i32>>,
+    event_channel: web::Data<async_channel::Sender<Event>>,
     web::Json(form): web::Json<CreateEventPayload>,
 ) -> actix_web::Result<impl Responder> {
-    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)
 }

src/webui/worker.rs

@@ -1,13 +1,21 @@
 use std::io;
 
-pub async fn worker_loop(r: async_channel::Receiver<i32>) -> 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<Event>) -> 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()));
             }
         }
     }