We want to see changes

src/main.rs

@@ -1,8 +1,8 @@
-use std::{collections::HashMap, sync::Arc, time::{SystemTime, UNIX_EPOCH}};
+use std::{collections::HashMap, convert::Infallible, sync::Arc, time::{SystemTime, UNIX_EPOCH}};
 
 use bluer::{self, Adapter, AdapterEvent, Address};
 
-use anyhow::{Result, Context};
+use anyhow::Result;
 use futures::{Stream, StreamExt};
 use tokio;
 use warp::Filter;
@@ -61,8 +61,8 @@ impl DataStore {
             let Data { addr, temperature, humidity, co2 } = data;
             let time = point.duration_since(UNIX_EPOCH).expect("Time went backwards").as_millis();
             scrape += &format!("sensirion_temperature_celsius{{id=\"{addr}\"}} {temperature} {time}\n");            
-            scrape += &format!("sensirion_humidity_percent{{id=\"{addr}\"}} {temperature} {time}\n");            
-            scrape += &format!("sensirion_co2_ppm{{id=\"{addr}\"}} {temperature} {time}\n");            
+            scrape += &format!("sensirion_humidity_percent{{id=\"{addr}\"}} {humidity} {time}\n");            
+            scrape += &format!("sensirion_co2_ppm{{id=\"{addr}\"}} {co2} {time}\n");            
         }
 
         scrape
@@ -79,19 +79,21 @@ async fn main() -> Result<()> {
 
     let mut stream = Box::pin(sensor_reports(adapter).await?);
     
-    let (tx, rx)= tokio::sync::mpsc::channel(256);
-
-    let sniffer: tokio::task::JoinHandle<()> = tokio::spawn(async move {
+    let sniffer_store = Arc::clone(&store);
+    tokio::spawn(async move {
         while let Some(data) = stream.next().await {
-            tx.send(data).await;
+            eprintln!("Data point: {:?}", data);
+            sniffer_store.lock().await.insert(data, SystemTime::now());
         }
     });
     
     let filter =
-                 warp::path!("metrics").map(move || store.lock().unwrap().scrape())
-             .or(warp::path!().map(move || { warp::reply::html("<h1>Sensirion BLE exporter</h1>") }));
+                 warp::path!("metrics")
+                    .map(move || Arc::clone(&store))
+                    .and_then(|store: Arc<Mutex<DataStore>>| async move { Ok::<_,Infallible>(store.lock().await.scrape()) })
+             .or(warp::path!().map(|| { warp::reply::html("<h1>Sensirion BLE exporter</h1>") }));
 
-    Ok(warp::serve(filter).run("127.0.0.1:9177").await)
+    Ok(warp::serve(filter).run(([127,0,0,1],9177)).await)
 
 }
 
@@ -106,6 +108,6 @@ async fn extract(adapter: Arc<Adapter>, evt: AdapterEvent) -> Option<Data> {
 }
 
 async fn sensor_reports<'a>(adapter: Arc<Adapter>) -> Result<impl Stream<Item=Data> + 'a> {
-    Ok(adapter.discover_devices().await?
+    Ok(adapter.discover_devices_with_changes().await?
         .filter_map(move |evt| { extract(Arc::clone(&adapter), evt) }))
 }