Added channel and worker.

This commit is contained in:
didier amyot committed 2023-02-11 08:54:54 -05:00
1 parent af687df9de
commit 5ff946af8f
5 files changed
+85 -38

No files matched your search

Generated
+36
View File
@@ -353,6 +353,17 @@ version = "0.10.3"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "619743e34b5ba4e9703bba34deac3427c72507c7159f5fd030aea8cac0cfe341" checksum = "619743e34b5ba4e9703bba34deac3427c72507c7159f5fd030aea8cac0cfe341"
[[package]]
name = "async-channel"
version = "1.8.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cf46fee83e5ccffc220104713af3292ff9bc7c64c7de289f66dae8e38d826833"
dependencies = [
"concurrent-queue",
"event-listener",
"futures-core",
]
[[package]] [[package]]
name = "async-trait" name = "async-trait"
version = "0.1.63" version = "0.1.63"
@@ -529,6 +540,15 @@ dependencies = [
"winapi", "winapi",
] ]
[[package]]
name = "concurrent-queue"
version = "2.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c278839b831783b70278b14df4d45e1beb1aad306c07bb796637de9a0e323e8e"
dependencies = [
"crossbeam-utils",
]
[[package]] [[package]]
name = "convert_case" name = "convert_case"
version = "0.4.0" version = "0.4.0"
@@ -564,6 +584,15 @@ dependencies = [
"cfg-if", "cfg-if",
] ]
[[package]]
name = "crossbeam-utils"
version = "0.8.14"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4fb766fa798726286dbbb842f174001dab8abc7b627a1dd86e0b7222a95d929f"
dependencies = [
"cfg-if",
]
[[package]] [[package]]
name = "crypto-common" name = "crypto-common"
version = "0.1.6" version = "0.1.6"
@@ -643,6 +672,7 @@ dependencies = [
"actix-web", "actix-web",
"actix-web-lab", "actix-web-lab",
"apache-avro", "apache-avro",
"async-channel",
"clap", "clap",
"clap-verbosity-flag", "clap-verbosity-flag",
"fraction", "fraction",
@@ -664,6 +694,12 @@ dependencies = [
"tokio", "tokio",
] ]
[[package]]
name = "event-listener"
version = "2.5.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0206175f82b8d6bf6652ff7d71a1e27fd2e4efde587fd368662814d6ec1d9ce0"
[[package]] [[package]]
name = "fallible-iterator" name = "fallible-iterator"
version = "0.2.0" version = "0.2.0"
+1
View File
@@ -29,6 +29,7 @@ actix-web-lab = "0.18.9"
minijinja-autoreload = "0.30.2" minijinja-autoreload = "0.30.2"
futures = "0.3.26" futures = "0.3.26"
actix-files = "0.6.2" actix-files = "0.6.2"
async-channel = "1.8.0"
[dependencies.apache-avro] [dependencies.apache-avro]
version = "0.14.0" version = "0.14.0"
@@ -1,10 +1,15 @@
<div class="uk-margin"> <div class="uk-margin">
<label class="uk-form-label">Event [JSON5]</label> <label class="uk-form-label"> Name
<textarea class="uk-textarea" <input class="uk-input" type="text" placeholder="Caroline Ellison" aria-label="Default"
name="event" name="event.NewUser.name">
style="font: 0.875rem/1.5 Consolas,monaco,monospace" </label>
rows="5" </div>
placeholder="JSON"></textarea>
<div class="uk-margin">
<label class="uk-form-label"> Public Key
<input class="uk-input" type="text" placeholder="xxxxx" aria-label="Default"
name="event.NewUser.PublicKey">
</label>
</div> </div>
{%include "partials/actions/submit_button.html"%} {%include "partials/actions/submit_button.html"%}
+35 -32
View File
@@ -1,29 +1,28 @@
#![feature(future_join)] #![feature(future_join)]
use std::{collections::HashMap, env, fs, io, path::PathBuf};
use std::io::{Error, ErrorKind};
use std::path::Path;
use actix_files;
use actix_utils::future::{ready, Ready};
use actix_web::{
App,
dev::ServiceResponse,
error,
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::{FutureExt, pin_mut, select};
use minijinja;
use minijinja_autoreload::AutoReloader;
use tokio::time::{Duration, sleep};
mod views; mod views;
mod config; mod config;
use actix_utils::future::{ready, Ready};
use minijinja;
use minijinja_autoreload::AutoReloader;
use std::io::{Error, ErrorKind};
use std::path::Path;
use std::{collections::HashMap, env, fs, io, path::PathBuf};
use tokio::time::{sleep, Duration};
use futures::{pin_mut, select, FutureExt};
use actix_files;
use actix_web::{
dev::ServiceResponse,
error,
http::{header::ContentType, StatusCode},
middleware::{ErrorHandlerResponse, ErrorHandlers, Logger},
web, App, FromRequest, HttpRequest, HttpResponse, HttpServer, Responder, Result,
};
use actix_web_lab::respond::Html;
struct MiniJinjaRenderer { struct MiniJinjaRenderer {
tmpl_env: web::Data<minijinja_autoreload::AutoReloader>, tmpl_env: web::Data<minijinja_autoreload::AutoReloader>,
} }
@@ -62,12 +61,16 @@ impl FromRequest for MiniJinjaRenderer {
} }
} }
async fn print_shit() -> std::io::Result<()>{ async fn worker_loop(r: async_channel::Receiver<i32>) -> io::Result<()> {
let mut i = 0;
loop { loop {
sleep(Duration::from_millis(10_000)).await; let x = r.recv().await;
println!("Loop on a different thread {:?}", i); match x {
i +=1; Ok(v) => println!("Processed value: {:?}", v),
Err(e) => {
println!("Error: {:?}", e);
return Ok(())
}
}
} }
} }
@@ -88,16 +91,16 @@ pub async fn web_ui_main(port: u16, enable_template_autoreload: bool) -> std::io
Ok(env) Ok(env)
}); });
let (s, r) = async_channel::unbounded::<i32>();
let tmpl_reloader = web::Data::new(tmpl_reloader); let tmpl_reloader = web::Data::new(tmpl_reloader);
let server = HttpServer::new(move || { let server = HttpServer::new(move || {
App::new() App::new()
.app_data(tmpl_reloader.clone()) .app_data(tmpl_reloader.clone())
.service( .app_data(web::Data::new(s.clone()))
actix_files::Files::new("/static", "./resources/static/") .service(actix_files::Files::new("/static", "./resources/static/")
.show_files_listing() .use_last_modified(true))
.use_last_modified(true),
)
.service(views::index) .service(views::index)
.service(views::actions) .service(views::actions)
.service(views::overview) .service(views::overview)
@@ -111,7 +114,7 @@ pub async fn web_ui_main(port: u16, enable_template_autoreload: bool) -> std::io
.bind(("0.0.0.0", port))?; .bind(("0.0.0.0", port))?;
let t1 = server.run().fuse(); let t1 = server.run().fuse();
let t2 = print_shit().fuse(); let t2 = worker_loop(r).fuse();
pin_mut!(t1, t2); pin_mut!(t1, t2);
select!{ select!{
+2
View File
@@ -84,10 +84,12 @@ struct CreateEventPayload {
pub async fn push_event( pub async fn push_event(
req: HttpRequest, req: HttpRequest,
tmpl_env: MiniJinjaRenderer, tmpl_env: MiniJinjaRenderer,
event_channel: web::Data<async_channel::Sender<i32>>,
web::Json(form): web::Json<CreateEventPayload>, web::Json(form): web::Json<CreateEventPayload>,
) -> actix_web::Result<impl Responder> { ) -> actix_web::Result<impl Responder> {
let shit = req.match_info().get("event_type").unwrap(); let shit = req.match_info().get("event_type").unwrap();
println!("payload: {:?}", form); println!("payload: {:?}", form);
let _ = event_channel.send(56).await;
let ctx = context! { let ctx = context! {
event_type => shit, event_type => shit,
}; };