refactor & remove delay
- move duplication in recipe streaming - remove unnessecary delay when sending back to client - increase chunk size from 100 to 200 Signed-off-by: Pakin <pakin.t@forth.co.th>
This commit is contained in:
Pakin
parent
377b0df681
commit
8cdba50c40
3 changed files with 322 additions and 245 deletions
389
src/main.rs
389
src/main.rs
|
|
@ -17,6 +17,7 @@ use futures::{
|
|||
stream::{SplitSink, SplitStream},
|
||||
};
|
||||
use libtbr::models::recipe::{MaterialSetting, Recipe, Recipe01};
|
||||
use log::info;
|
||||
use rayon::iter::{IntoParallelRefIterator, ParallelIterator};
|
||||
use redis::{TypedCommands, cmd};
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
|
@ -41,6 +42,8 @@ use crate::{
|
|||
mod stream;
|
||||
mod tx;
|
||||
|
||||
const CHUNK_SIZE: usize = 200;
|
||||
|
||||
// features
|
||||
// - get result from recipe_repo
|
||||
// - store in redis
|
||||
|
|
@ -105,6 +108,22 @@ impl DevConfig {
|
|||
// Ok(())
|
||||
// }
|
||||
|
||||
async fn invoke_checkout_request(
|
||||
config: DevConfig,
|
||||
path: String,
|
||||
) -> Result<String, Box<dyn std::error::Error>> {
|
||||
let client = reqwest::Client::new();
|
||||
|
||||
let req_path = config.get_file_from_recipe_repo(path);
|
||||
// println!("dbg: {req_path}");
|
||||
let res = client.get(req_path).send().await?;
|
||||
|
||||
match res.text().await {
|
||||
Ok(raw) => Ok(raw),
|
||||
Err(e) => Err(format!("{e}").into()),
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn create_recipe_repo_router() -> Router<Arc<AppState>> {
|
||||
Router::new()
|
||||
// .route("/", get(get_root_files))
|
||||
|
|
@ -160,59 +179,6 @@ async fn handle_socket(
|
|||
|
||||
tokio::spawn(write(sender, rx));
|
||||
tokio::spawn(read(state, receiver, tx.clone(), user_sys_rx));
|
||||
// NOTE:
|
||||
|
||||
// while let Some(msg) = socket.recv().await {
|
||||
// if let Ok(msg) = msg {
|
||||
// match msg {
|
||||
// Message::Text(utf8_bytes) => {
|
||||
// println!("text recv: {utf8_bytes}");
|
||||
|
||||
// let req = utf8_bytes.clone().as_str().to_string();
|
||||
// let req_struct: serde_json::Value = match serde_json::from_str(&req) {
|
||||
// Ok(r) => r,
|
||||
// Err(_) => serde_json::Value::Null,
|
||||
// };
|
||||
|
||||
// // response
|
||||
// let n_state = state.clone();
|
||||
// let response =
|
||||
// handle_recipe_websocket_message(req_struct, n_state, &mut socket).await?;
|
||||
// let response_next = match serde_json::to_string(&response) {
|
||||
// Ok(p) => p,
|
||||
// Err(_) => "{'result': 'error'}".to_string(),
|
||||
// };
|
||||
|
||||
// let result = socket.send(Message::Text(response_next.into())).await;
|
||||
// if let Err(error) = result {
|
||||
// println!("Error sending: {error}");
|
||||
// send_close_message(socket, 1011, &format!("Error occured: {error}")).await;
|
||||
// break;
|
||||
// }
|
||||
// }
|
||||
// Message::Binary(bytes) => {
|
||||
// println!("recv bytes len: {}", bytes.len());
|
||||
// let result = socket
|
||||
// .send(Message::Text(
|
||||
// format!("recv bytes len: {}", bytes.len()).into(),
|
||||
// ))
|
||||
// .await;
|
||||
// if let Err(error) = result {
|
||||
// println!("Error sending: {error}");
|
||||
// send_close_message(socket, 1011, &format!("Error occured: {error}")).await;
|
||||
// break;
|
||||
// }
|
||||
// }
|
||||
// _ => {}
|
||||
// }
|
||||
// } else {
|
||||
// let error = msg.err().unwrap();
|
||||
// println!("Error while receiving message: {error:?}");
|
||||
// send_close_message(socket, 1011, &format!("Error occured: {error}")).await;
|
||||
// break;
|
||||
// }
|
||||
// }
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
|
@ -257,18 +223,6 @@ async fn fetch_content_from_redis_byte(redis: redis::Client, key: &str) -> Resul
|
|||
return Ok(vec![]);
|
||||
}
|
||||
}
|
||||
|
||||
// match res {
|
||||
// Ok(s) => {
|
||||
// if let Some(res) = s {
|
||||
// Ok(res.as_bytes().to_vec())
|
||||
// } else {
|
||||
// Err(format!("result error from key: {key}"))
|
||||
// }
|
||||
// }
|
||||
// Err(e) => Err(format!("redis get failed: {e}")),
|
||||
// }
|
||||
//
|
||||
}
|
||||
|
||||
#[derive(Serialize, Deserialize)]
|
||||
|
|
@ -309,27 +263,6 @@ fn convert_ack_command(cmd_req: &serde_json::Value) -> Option<CommandRequestPayl
|
|||
}
|
||||
}
|
||||
|
||||
// /// Result to send to client
|
||||
// #[derive(Serialize, Deserialize)]
|
||||
// struct RecipeOverview {
|
||||
// productCode: String,
|
||||
// name: Option<String>,
|
||||
// description: Option<String>,
|
||||
// tags: String,
|
||||
// status: MenuStatus,
|
||||
// }
|
||||
|
||||
// #[derive(Serialize, Deserialize)]
|
||||
// enum MenuStatus {
|
||||
// #[serde(rename = "ready")]
|
||||
// Ready,
|
||||
// #[serde(rename = "obsolete")]
|
||||
// Obsolete,
|
||||
// #[serde(rename = "drafted")]
|
||||
// Drafted,
|
||||
// Status(String), // additional status
|
||||
// }
|
||||
|
||||
async fn read(
|
||||
// redis: redis::Client,
|
||||
mut state: Arc<AppState>,
|
||||
|
|
@ -339,6 +272,7 @@ async fn read(
|
|||
// cmd_atom: crossbeam_queue::ArrayQueue<CommandRequestPayload>,
|
||||
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
|
||||
let redis = state.redis_cli.clone();
|
||||
let config = state.dev_config.clone();
|
||||
let tx_to_client = tx.clone();
|
||||
|
||||
tokio::spawn(async move {
|
||||
|
|
@ -365,10 +299,9 @@ async fn read(
|
|||
// get actual version
|
||||
//
|
||||
let latest_key =
|
||||
format!("master:{country}/version", country = recipe_param.country);
|
||||
format!("{country}/version", country = recipe_param.country);
|
||||
|
||||
println!("latest key: {latest_key}");
|
||||
let latest_version =
|
||||
let mut latest_version =
|
||||
match fetch_content_from_redis_byte(redis.clone(), &latest_key).await {
|
||||
Ok(x) => {
|
||||
// decode brotli
|
||||
|
|
@ -390,6 +323,18 @@ async fn read(
|
|||
}
|
||||
};
|
||||
|
||||
if latest_version.is_empty() {
|
||||
// cannot get actual version, try get from git
|
||||
latest_version =
|
||||
match invoke_checkout_request(config.clone(), latest_key).await {
|
||||
Ok(version) => version,
|
||||
Err(e) => {
|
||||
println!("Error on checkout: {e}");
|
||||
"".to_string()
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
let req_file = if is_req_patch(&recipe_param) {
|
||||
format!(
|
||||
"stx_{country}_{version}.json",
|
||||
|
|
@ -405,7 +350,6 @@ async fn read(
|
|||
};
|
||||
|
||||
let mut retry_cnt = 0;
|
||||
let sid = Uuid::new_v4();
|
||||
println!("init req: {req_file}");
|
||||
|
||||
match get_local_file(req_file) {
|
||||
|
|
@ -416,67 +360,15 @@ async fn read(
|
|||
// split send
|
||||
let recipe: Recipe = serde_json::from_str(&file_content)?;
|
||||
|
||||
// concat all recipes including subs
|
||||
let r01s: Vec<Recipe01> = recipe
|
||||
.Recipe01
|
||||
.par_iter()
|
||||
.flat_map(|x| {
|
||||
let mut v = Vec::new();
|
||||
v.push(x.clone());
|
||||
|
||||
if let Some(sub) = x.clone().SubMenu {
|
||||
v.extend(sub);
|
||||
}
|
||||
|
||||
v
|
||||
})
|
||||
.collect();
|
||||
|
||||
let matset: Vec<MaterialSetting> = recipe.MaterialSetting.clone();
|
||||
|
||||
let ss = StreamDataStart::new(
|
||||
r01s.len(),
|
||||
100,
|
||||
Some("local".to_string()),
|
||||
);
|
||||
let sid = ss.get_id();
|
||||
|
||||
if let Some(err) = tx.send(ss.as_msg()).await.err() {
|
||||
println!("ERR: send tx error, {err:?}");
|
||||
}
|
||||
|
||||
for (index, chunk) in r01s.chunks(100).enumerate() {
|
||||
let sda =
|
||||
StreamDataChunk::new(&sid, index * 100, chunk.to_vec());
|
||||
|
||||
// no validate
|
||||
if let Some(err) = tx.send(sda.as_msg()).await.err() {
|
||||
println!("ERR: send tx error, {err:?}");
|
||||
}
|
||||
}
|
||||
|
||||
let mat_exid = sid.clone();
|
||||
let extp = "matset";
|
||||
for (index, chunk) in matset.chunks(100).enumerate() {
|
||||
let curr_ch_id = format!("{mat_exid}_{index}");
|
||||
|
||||
let extra_matset =
|
||||
StreamDataExtra::new(&curr_ch_id, &extp, chunk.to_vec());
|
||||
|
||||
if let Some(err) = tx.send(extra_matset.as_msg()).await.err() {
|
||||
println!("ERR: send tx extra error: {err:?}");
|
||||
}
|
||||
}
|
||||
|
||||
let end_msg = StreamDataEnd::new(&sid);
|
||||
|
||||
if let Some(err) = tx.send(end_msg.as_msg()).await.err() {
|
||||
println!("ERR: send tx error, {err:?}");
|
||||
}
|
||||
throttle_send_recipe(
|
||||
&recipe,
|
||||
&tx,
|
||||
recipe_param.country,
|
||||
latest_version,
|
||||
)
|
||||
.await;
|
||||
}
|
||||
Err(_) => {
|
||||
// savable sid means it could find data but not guarantee if sendable
|
||||
let mut success_sid = String::new();
|
||||
// concurrent fetch
|
||||
for i in 1..5 {
|
||||
retry_cnt = i;
|
||||
|
|
@ -505,109 +397,23 @@ async fn read(
|
|||
{
|
||||
let recipe: Recipe =
|
||||
serde_json::from_str(&sbuf)?;
|
||||
|
||||
let r01s: Vec<Recipe01> = recipe
|
||||
.Recipe01
|
||||
.par_iter()
|
||||
.flat_map(|x| {
|
||||
let mut v = Vec::new();
|
||||
v.push(x.clone());
|
||||
|
||||
if let Some(sub) = x.clone().SubMenu {
|
||||
v.extend(sub);
|
||||
}
|
||||
|
||||
v
|
||||
})
|
||||
.collect();
|
||||
|
||||
let matset: Vec<MaterialSetting> =
|
||||
recipe.MaterialSetting.clone();
|
||||
|
||||
// test stream start model
|
||||
let ss = StreamDataStart::new(
|
||||
r01s.len(),
|
||||
100,
|
||||
Some("redis".to_string()),
|
||||
);
|
||||
|
||||
let sid = ss.get_id();
|
||||
success_sid = sid.clone();
|
||||
|
||||
if let Some(err) =
|
||||
tx.send(ss.as_msg()).await.err()
|
||||
{
|
||||
println!("ERR: send tx error, {err:?}");
|
||||
}
|
||||
|
||||
// split send
|
||||
|
||||
for (index, chunk) in
|
||||
r01s.chunks(100).enumerate()
|
||||
{
|
||||
let sda = StreamDataChunk::new(
|
||||
&sid,
|
||||
index * 100,
|
||||
chunk.to_vec(),
|
||||
);
|
||||
|
||||
// no validate
|
||||
if let Some(err) =
|
||||
tx.send(sda.as_msg()).await.err()
|
||||
{
|
||||
println!("ERR: send tx error, {err:?}");
|
||||
}
|
||||
}
|
||||
|
||||
let mat_exid = sid.clone();
|
||||
let extp = "matset";
|
||||
for (index, chunk) in
|
||||
matset.chunks(100).enumerate()
|
||||
{
|
||||
let curr_ch_id =
|
||||
format!("{mat_exid}_{index}");
|
||||
|
||||
let extra_matset = StreamDataExtra::new(
|
||||
&curr_ch_id,
|
||||
&extp,
|
||||
chunk.to_vec(),
|
||||
);
|
||||
|
||||
if let Some(err) = tx
|
||||
.send(extra_matset.as_msg())
|
||||
.await
|
||||
.err()
|
||||
{
|
||||
println!(
|
||||
"ERR: send tx extra error: {err:?}"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
let rp_clone = recipe.clone();
|
||||
tokio::task::spawn(async move {
|
||||
rp_clone.export_to_json_file(Some(
|
||||
format!(
|
||||
"result.{country}.{version}.json",
|
||||
country = recipe_param.country,
|
||||
version = latest_version
|
||||
),
|
||||
));
|
||||
});
|
||||
throttle_send_recipe(
|
||||
&recipe,
|
||||
&tx,
|
||||
recipe_param.country,
|
||||
latest_version,
|
||||
)
|
||||
.await;
|
||||
|
||||
break;
|
||||
}
|
||||
}
|
||||
Err(_) => {}
|
||||
}
|
||||
} else {
|
||||
// retry get from git
|
||||
}
|
||||
}
|
||||
|
||||
let end_msg = StreamDataEnd::new(&success_sid);
|
||||
|
||||
if let Some(err) = tx.send(end_msg.as_msg()).await.err() {
|
||||
println!("ERR: send tx error, {err:?}");
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -653,8 +459,11 @@ async fn write(
|
|||
while let Some(res) = rx.recv().await {
|
||||
// no check
|
||||
// println!("sending {res:?}");
|
||||
info!("sending to client ...");
|
||||
let _ = sender.send(res.to_string().into()).await;
|
||||
let _ = tokio::time::sleep(Duration::from_millis(100)).await;
|
||||
|
||||
// リミットブレく - limito breaku!! (uncomment to slow down messages)
|
||||
// let _ = tokio::time::sleep(Duration::from_millis(100)).await;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
|
@ -691,6 +500,94 @@ fn get_key_cache(country: String, version: String, is_patch: bool, retry_cnt: i3
|
|||
}
|
||||
}
|
||||
|
||||
async fn throttle_send_recipe(
|
||||
recipe: &Recipe,
|
||||
tx: &Sender<serde_json::Value>,
|
||||
country: String,
|
||||
version: String,
|
||||
) {
|
||||
let r01s: Vec<Recipe01> = recipe
|
||||
.Recipe01
|
||||
.par_iter()
|
||||
.flat_map(|x| {
|
||||
let mut v = Vec::new();
|
||||
v.push(x.clone());
|
||||
|
||||
if let Some(sub) = x.clone().SubMenu {
|
||||
v.extend(sub);
|
||||
}
|
||||
|
||||
v
|
||||
})
|
||||
.collect();
|
||||
|
||||
let matset: Vec<MaterialSetting> = recipe.MaterialSetting.clone();
|
||||
|
||||
// test stream start model
|
||||
let ss = StreamDataStart::new(r01s.len(), CHUNK_SIZE, Some("redis".to_string()));
|
||||
|
||||
let sid = ss.get_id();
|
||||
info!("starting {sid}");
|
||||
|
||||
if let Some(err) = tx.send(ss.as_msg()).await.err() {
|
||||
println!("ERR: send tx error, {err:?}");
|
||||
}
|
||||
|
||||
// split send
|
||||
|
||||
for (index, chunk) in r01s.chunks(CHUNK_SIZE).enumerate() {
|
||||
let sda = StreamDataChunk::new(&sid, index * CHUNK_SIZE, chunk.to_vec());
|
||||
|
||||
// no validate
|
||||
if let Some(err) = tx.send(sda.as_msg()).await.err() {
|
||||
println!("ERR: send tx error, {err:?}");
|
||||
}
|
||||
}
|
||||
|
||||
let mat_exid = sid.clone();
|
||||
let extp = "matset";
|
||||
for (index, chunk) in matset.chunks(CHUNK_SIZE).enumerate() {
|
||||
let curr_ch_id = format!("{mat_exid}_{index}");
|
||||
|
||||
let extra_matset = StreamDataExtra::new(&curr_ch_id, &extp, chunk.to_vec());
|
||||
|
||||
if let Some(err) = tx.send(extra_matset.as_msg()).await.err() {
|
||||
println!("ERR: send tx extra error: {err:?}");
|
||||
}
|
||||
}
|
||||
|
||||
let extl = "topplist";
|
||||
for (index, chunk) in recipe.Topping.ToppingList.chunks(CHUNK_SIZE).enumerate() {
|
||||
let curr_ch_id = format!("{mat_exid}_tl{index}");
|
||||
let extra_topplist = StreamDataExtra::new(&curr_ch_id, &extl, chunk.to_vec());
|
||||
if let Some(err) = tx.send(extra_topplist.as_msg()).await.err() {
|
||||
println!("ERR: send tx extra2 error: {err:?}");
|
||||
}
|
||||
}
|
||||
|
||||
let extg = "toppgrp";
|
||||
for (index, chunk) in recipe.Topping.ToppingGroup.chunks(CHUNK_SIZE).enumerate() {
|
||||
let curr_ch_id = format!("{mat_exid}_tg{index}");
|
||||
let extra_toppgrp = StreamDataExtra::new(&curr_ch_id, &extg, chunk.to_vec());
|
||||
if let Some(err) = tx.send(extra_toppgrp.as_msg()).await.err() {
|
||||
println!("ERR: send tx extra2 error: {err:?}");
|
||||
}
|
||||
}
|
||||
|
||||
let rp_clone = recipe.clone();
|
||||
tokio::task::spawn(async move {
|
||||
rp_clone.export_to_json_file(Some(format!("result.{country}.{version}.json")));
|
||||
});
|
||||
info!("sending {sid}");
|
||||
|
||||
// return sid;
|
||||
let end_msg = StreamDataEnd::new(&sid);
|
||||
|
||||
if let Some(err) = tx.send(end_msg.as_msg()).await.err() {
|
||||
println!("ERR: send tx error, {err:?}");
|
||||
}
|
||||
}
|
||||
|
||||
pub struct AppState {
|
||||
dev_config: DevConfig,
|
||||
redis_cli: redis::Client,
|
||||
|
|
@ -809,6 +706,8 @@ impl AppState {
|
|||
async fn main() -> Result<(), Box<dyn std::error::Error>> {
|
||||
println!("Hello, world!");
|
||||
dotenv::dotenv().ok();
|
||||
|
||||
env_logger::init();
|
||||
// send req to repo service
|
||||
|
||||
let server_port = env::var("PORT").unwrap_or("36579".to_string());
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue