aljaz/sphinx-key
1
1
Fork 0
mirror of https://github.com/stakwork/sphinx-key.git synced 2026-01-06 00:54:23 +01:00
Code Issues Packages Projects Releases Wiki Activity

use blocking_recv to cross rumqttd sync channel and tokio channle

Browse Source
This commit is contained in:
Evan Feenstra
2023-02-20 15:16:10 -08:00
parent f96280a222
commit 90342391da
4 changed files with 42 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
broker/src/main.rs
View File

@@ -95,9 +95,9 @@ async fn rocket() -> _ {
async fn run_main(parent_fd: i32) -> rocket::Rocket<rocket::Build> {
let settings = read_broker_config(BROKER_CONFIG_PATH);
let (tx, rx) = mpsc::channel(1000);
let (status_tx, mut status_rx) = mpsc::channel(1000);
let (error_tx, error_rx) = broadcast::channel(1000);
let (tx, rx) = mpsc::channel(10000);
let (status_tx, mut status_rx) = mpsc::channel(10000);
let (error_tx, error_rx) = broadcast::channel(10000);
error_log::log_errors(error_rx);
log::info!("=> start broker on network: {}", settings.network);

55
broker/src/mqtt.rs
View File

@@ -1,15 +1,12 @@
use crate::util::Settings;
use crate::{ChannelReply, ChannelRequest};
use rocket::tokio::time::timeout;
use rocket::tokio::{self, sync::broadcast, sync::mpsc};
use rocket::tokio::{sync::broadcast, sync::mpsc};
use rumqttd::{Alert, AlertEvent, Broker, Config, Notification};
use sphinx_signer::sphinx_glyph::topics;
use std::time::Duration;
// pub const INTERNAL_CONTROL: &str = "INTERNAL_CONTROL";
// must get a reply within this time, or disconnects
const REPLY_TIMEOUT_MS: u64 = 10000;
// const REPLY_TIMEOUT_MS: u64 = 10000;
pub fn start_broker(
mut receiver: mpsc::Receiver<ChannelRequest>,
@@ -35,8 +32,7 @@ pub fn start_broker(
// connected/disconnected status alerts
let status_sender_ = status_sender.clone();
let _alerts_handle = tokio::spawn(async move {
loop {
let _alerts_handle = std::thread::spawn(move || loop {
let alert = alerts.poll();
println!("Alert: {:?}", alert);
match alert.1 {
@@ -47,38 +43,31 @@ pub fn start_broker(
AlertEvent::Disconnect => Some(false),
_ => None,
} {
let _ = status_sender_.send(status).await;
let _ = status_sender_.blocking_send(status);
}
}
}
_ => (),
}
}
});
// msg forwarding
let (msg_tx, mut msg_rx): (mpsc::Sender<Vec<u8>>, mpsc::Receiver<Vec<u8>>) =
mpsc::channel(1000);
// link_tx.subscribe(INTERNAL_CONTROL)?;
link_tx.subscribe(topics::VLS_RETURN)?;
link_tx.subscribe(topics::CONTROL_RETURN)?;
link_tx.subscribe(topics::ERROR)?;
mpsc::channel(10000);
link_tx.subscribe(topics::VLS_RETURN).unwrap();
link_tx.subscribe(topics::CONTROL_RETURN).unwrap();
link_tx.subscribe(topics::ERROR).unwrap();
let _sub_task = tokio::spawn(async move {
println!("ummm....");
let _sub_task = std::thread::spawn(move || {
while let Ok(message) = link_rx.recv() {
if let Some(n) = message {
match n {
Notification::Forward(f) => {
println!("GOT A FORWARDED MSG! FORWARD! {:?}", f.publish.topic);
if f.publish.topic == topics::ERROR {
let _ = error_sender.send(f.publish.topic.to_vec());
} else {
println!("send now on msg_tx {:?}", f.publish.payload.to_vec());
if let Err(e) = msg_tx.send(f.publish.payload.to_vec()).await {
if let Err(e) = msg_tx.blocking_send(f.publish.payload.to_vec()) {
log::error!("failed to pub to msg_tx! {:?}", e);
}
println!("sent on msg_tx");
}
}
_ => (),
@@ -87,36 +76,26 @@ pub fn start_broker(
}
});
let _relay_task = tokio::spawn(async move {
while let Some(msg) = receiver.recv().await {
let _relay_task = std::thread::spawn(move || {
while let Some(msg) = receiver.blocking_recv() {
if let Err(e) = link_tx.publish(msg.topic, msg.message) {
log::error!("failed to pub to link_tx! {:?}", e);
}
println!("PUBBED TO LINKTX....");
// let rep = msg_rx.recv().await;
// println!("REPPPP {:?}", rep);
match timeout(Duration::from_millis(REPLY_TIMEOUT_MS), msg_rx.recv()).await {
Ok(rep) => {
println!("GOT A REPLY {:?}", rep);
let rep = msg_rx.blocking_recv();
if let Some(reply) = rep {
if let Err(_) = msg.reply_tx.send(ChannelReply { reply }) {
log::warn!("could not send on reply_tx");
}
}
}
Err(e) => {
log::warn!("reply_tx timed out {:?}", e);
let _ = status_sender.send(false).await;
}
}
}
});
// _sub_task.await.unwrap();
// _relay_task.await.unwrap();
// alerts_handle.await?;
std::thread::sleep(Duration::from_secs(1));
// alerts_handle.await?;
// sub_task.await?;
// relay_task.await?;
Ok(())
}

2
broker/src/routes.rs
View File

@@ -27,6 +27,7 @@ pub async fn control(sender: &State<Sender<ChannelRequest>>, msg: &str) -> Resul
let _ = sender.send(request).await.map_err(|_| Error::Fail)?;
// wait for reply
let reply = reply_rx.await.map_err(|_| Error::Fail)?;
Ok(hex::encode(reply.reply).to_string())
}
@@ -83,6 +84,7 @@ impl<'r, 'o: 'r> Responder<'r, 'o> for Error {
fn respond_to(self, req: &'r rocket::Request<'_>) -> response::Result<'o> {
// log `self` to your favored error tracker, e.g.
// sentry::capture_error(&self);
println!("ERROR {:?}", self);
match self {
// in our simplistic example, we're happy to respond with the default 500 responder in all cases
_ => Status::InternalServerError.respond_to(req),

6
broker/src/run_test.rs
View File

@@ -25,6 +25,12 @@ pub async fn run_test() -> rocket::Rocket<rocket::Build> {
start_broker(rx, status_tx, error_tx.clone(), CLIENT_ID, settings)
.expect("FAILED TO START BROKER");
log::info!("BROKER started!");
log::info!("=> wait for connected status");
// wait for connection = true
let status = status_rx.recv().await.expect("couldnt receive");
log::info!("=> connection status: {}", status);
// let mut connected = false;
// let tx_ = tx.clone();
tokio::spawn(async move {