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

update dependencies

Browse source
This commit is contained in:
Robin Appelman 2021-03-23 20:22:41 +01:00
commit 0d5a00f269
4 changed files with 361 additions and 683 deletions
Download patch file Download diff file Expand all files Collapse all files

1021
Cargo.lock generated
View file

File diff suppressed because it is too large Load diff

21
Cargo.toml
View file

@ -7,15 +7,16 @@ license = "MIT OR Apache-2.0"
repository = "https://github.com/icewind1991/tasproxy" repository = "https://github.com/icewind1991/tasproxy"
[dependencies] [dependencies]
rumqttc = "0.2.0" rumqttc = "0.5"
tokio = "0.2" tokio = { version = "1", features = ["macros", "rt-multi-thread"] }
dashmap = "3.11" futures-util = "0.3"
json = "0.12.4" dashmap = "3"
warp = "0.2.5" json = "0.12"
warp = "0.3"
dotenv = "0.15.0" dotenv = "0.15.0"
ctrlc = { version = "3.1.7", features = ["termination"] } ctrlc = { version = "3", features = ["termination"] }
color-eyre = "0.5.7" color-eyre = "0.5"
async-stream = "0.3.0" async-stream = "0.3"
pin-utils = "0.1.0" pin-utils = "0.1"
hostname = "^0.3" hostname = "^0.3"
warp-reverse-proxy = { version = "0.2.0", git = "https://github.com/icewind1991/warp-reverse-proxy", branch = "rustls", default_features = false, features = ["rustls-tls"] } warp-reverse-proxy = { version = "0.3", default_features = false, features = ["rustls-tls"] }

8
src/main.rs
View file

@ -4,12 +4,12 @@ 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 futures_util::stream::StreamExt;
use pin_utils::pin_mut; use pin_utils::pin_mut;
use rumqttc::{AsyncClient, 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::time::sleep;
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};
@ -40,7 +40,7 @@ async fn main() -> Result<()> {
eprintln!("lost mqtt collection: {:#}", e); eprintln!("lost mqtt collection: {:#}", e);
} }
eprintln!("reconnecting after 1s"); eprintln!("reconnecting after 1s");
tokio::time::delay_for(Duration::from_secs(1)).await; tokio::time::sleep(Duration::from_secs(1)).await;
} }
}); });
@ -125,7 +125,7 @@ async fn query_device(client: AsyncClient, device: Device) {
// one device boot, the discovery event can happen before the device is ready to respond to our messages // 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 // thus we wait 5 seconds before asking
delay_for(Duration::from_secs(5)).await; sleep(Duration::from_secs(5)).await;
if let Err(e) = client if let Err(e) = client
.publish( .publish(

12
src/mqtt.rs
View file

@ -1,7 +1,7 @@
use async_stream::try_stream; use async_stream::try_stream;
use color_eyre::Result; use color_eyre::Result;
use futures_util::stream::{Stream, StreamExt};
use rumqttc::{AsyncClient, Event, EventLoop, MqttOptions, Packet, Publish, QoS}; use rumqttc::{AsyncClient, Event, EventLoop, MqttOptions, Packet, Publish, QoS};
use tokio::stream::{Stream, StreamExt};
pub async fn mqtt_stream( pub async fn mqtt_stream(
mqtt_options: MqttOptions, mqtt_options: MqttOptions,
@ -10,10 +10,12 @@ pub async fn mqtt_stream(
client.subscribe("tele/+/LWT", QoS::AtMostOnce).await?; client.subscribe("tele/+/LWT", QoS::AtMostOnce).await?;
client.subscribe("stat/+/RESULT", QoS::AtMostOnce).await?; client.subscribe("stat/+/RESULT", QoS::AtMostOnce).await?;
let stream = event_loop_to_stream(event_loop).filter_map(|event| match event { let stream = event_loop_to_stream(event_loop).filter_map(|event| async move {
Ok(Event::Incoming(Packet::Publish(message))) => Some(Ok(message)), match event {
Ok(_) => None, Ok(Event::Incoming(Packet::Publish(message))) => Some(Ok(message)),
Err(e) => Some(Err(e)), Ok(_) => None,
Err(e) => Some(Err(e)),
}
}); });
Ok((client, stream)) Ok((client, stream))