robin/tasproxy
1
0
Fork 0
mirror of https://codeberg.org/icewind/tasproxy.git synced 2026-06-03 18:24:08 +02:00
Code Issues Projects Releases Wiki Activity

delay device queries

Browse source
This commit is contained in:
Robin Appelman 2020-12-28 20:23:28 +01:00
commit 498627c29f
1 changed files with 46 additions and 40 deletions
Download patch file Download diff file Expand all files Collapse all files

72
src/main.rs
View file

@ -1,14 +1,15 @@
use crate::config::Config; use crate::config::Config;
use crate::devices::DeviceState; use crate::devices::{Device, DeviceState};
use crate::mqtt::mqtt_stream; use crate::mqtt::mqtt_stream;
use crate::topic::Topic; use crate::topic::Topic;
use color_eyre::{eyre::WrapErr, Result}; use color_eyre::{eyre::WrapErr, Result};
use dashmap::DashMap; use dashmap::DashMap;
use pin_utils::pin_mut; use pin_utils::pin_mut;
use rumqttc::QoS; use rumqttc::{AsyncClient, QoS};
use std::sync::Arc; use std::sync::Arc;
use std::time::Duration; use std::time::Duration;
use tokio::stream::StreamExt; use tokio::stream::StreamExt;
use tokio::time::delay_for;
use warp::hyper::http::uri::Authority; use warp::hyper::http::uri::Authority;
use warp::Filter; use warp::Filter;
use warp_reverse_proxy::{extract_request_data_filter, proxy_to_and_forward_response}; use warp_reverse_proxy::{extract_request_data_filter, proxy_to_and_forward_response};
@ -67,7 +68,7 @@ async fn main() -> Result<()> {
Err(warp::reject::not_found()) Err(warp::reject::not_found())
} }
} else { } else {
eprintln!("Error {} has no discovered", requested_device); eprintln!("Error {} has not been discovered", requested_device);
Err(warp::reject::not_found()) Err(warp::reject::not_found())
} }
}, },
@ -96,44 +97,17 @@ async fn mqtt_client(config: &Config, device_states: DeviceStates) -> Result<()>
let topic = Topic::from(message.topic.as_str()); let topic = Topic::from(message.topic.as_str());
match topic { match topic {
Topic::LWT(device) => { Topic::LWT(device) => match payload {
match payload {
"Online" => { "Online" => {
println!("Discovered {}", device.hostname); println!("Discovered {}", device.hostname);
// on discovery, ask the device for it's ip and name query_device(client.clone(), device).await;
let send_client = client.clone();
tokio::task::spawn(async move {
if let Err(e) = send_client
.publish(
device.get_topic("cmnd", "IPADDRESS"),
QoS::AtLeastOnce,
false,
"",
)
.await
{
eprintln!("Failed to ask for device IP: {:#}", e);
}
if let Err(e) = send_client
.publish(
device.get_topic("cmnd", "DeviceName"),
QoS::AtLeastOnce,
false,
"",
)
.await
{
eprintln!("Failed to ask for device name: {:#}", e);
}
});
} }
"Offline" => { "Offline" => {
println!("Removing {}", device.hostname); println!("Removing {}", device.hostname);
device_states.remove(&device.hostname); device_states.remove(&device.hostname);
} }
_ => {} _ => {}
} },
}
Topic::Result(device) => { Topic::Result(device) => {
if let Ok(json) = json::parse(payload) { if let Ok(json) = json::parse(payload) {
let mut device_state = device_states.entry(device.hostname).or_default(); let mut device_state = device_states.entry(device.hostname).or_default();
@ -145,3 +119,35 @@ async fn mqtt_client(config: &Config, device_states: DeviceStates) -> Result<()>
} }
Ok(()) Ok(())
} }
async fn query_device(client: AsyncClient, device: Device) {
tokio::task::spawn(async move {
// one device boot, the discovery event can happen before the device is ready to respond to our messages
// thus we wait 5 seconds before asking
delay_for(Duration::from_secs(5)).await;
if let Err(e) = client
.publish(
device.get_topic("cmnd", "IPADDRESS"),
QoS::AtLeastOnce,
false,
"",
)
.await
{
eprintln!("Failed to ask for device IP: {:#}", e);
}
if let Err(e) = client
.publish(
device.get_topic("cmnd", "DeviceName"),
QoS::AtLeastOnce,
false,
"",
)
.await
{
eprintln!("Failed to ask for device name: {:#}", e);
}
});
}