1 files changed, 82 insertions, 0 deletions
diff --git a/src/bin/pissd/main.rs b/src/bin/pissd/main.rs
new file mode 100644
index 0000000..9f579b7
--- /dev/null
+++ b/src/bin/pissd/main.rs
@@ -0,0 +1,82 @@
+use std::{
+    error::Error,
+    sync::{atomic::Ordering, Arc},
+};
+
+use atomic_float::AtomicF32;
+use lazy_static::lazy_static;
+use lightstreamer_client::{
+    ls_client::{LightstreamerClient, LogType, Transport},
+    subscription::{Snapshot, Subscription, SubscriptionMode},
+    subscription_listener::SubscriptionListener,
+};
+use tokio::sync::Notify;
+use tracing::info;
+use zbus::interface;
+
+lazy_static! {
+    static ref PISS_PERCENTAGE: AtomicF32 = AtomicF32::new(0.0);
+}
+
+#[derive(Debug)]
+struct PissSubscriberListener;
+
+impl SubscriptionListener for PissSubscriberListener {
+    fn on_item_update(&self, update: &lightstreamer_client::item_update::ItemUpdate) {
+        if let Some(value) = update.get_value("Value") {
+            info!("Piss tank at {value}%");
+            PISS_PERCENTAGE.store(value.parse::<f32>().unwrap(), Ordering::Relaxed);
+        }
+    }
+}
+
+struct PissdInterface;
+
+#[interface(name = "dev.exvacuum.pissd")]
+impl PissdInterface {
+    fn get_piss_percentage(&self) -> f32 {
+        PISS_PERCENTAGE.load(Ordering::Relaxed)
+    }
+}
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn Error>> {
+    tracing_subscriber::fmt::init();
+    let mut ls = match LightstreamerClient::new(
+        Some("http://push.lightstreamer.com/lightstreamer"),
+        Some("ISSLIVE"),
+        None,
+        None,
+    ) {
+        Ok(ls) => ls,
+        Err(e) => panic!("Could not establish ISS telemetry connection: {e:?}"),
+    };
+    ls.set_logging_type(LogType::TracingLogs);
+    ls.connection_options.set_slowing_enabled(false);
+    ls.connection_options
+        .set_forced_transport(Some(Transport::WsStreaming));
+
+    let mut subscription = Subscription::new(
+        SubscriptionMode::Merge,
+        Some(vec!["NODE3000005".into()]),
+        Some(vec!["TimeStamp".into(), "Value".into()]),
+    )
+    .unwrap();
+    subscription
+        .set_requested_snapshot(Some(Snapshot::Yes))
+        .unwrap();
+    subscription.add_listener(Box::new(PissSubscriberListener));
+
+    let shutdown = Arc::new(Notify::new());
+    ls.subscribe(subscription);
+
+    let pissd_interface = PissdInterface;
+    let connection = zbus::connection::Builder::session()?
+        .name("dev.exvacuum.pissd")?
+        .serve_at("/dev/exvacuum/pissd", pissd_interface)?
+        .build()
+        .await?;
+
+    ls.connect(shutdown.clone()).await?;
+    Ok(())
+}