3 files changed, 291 insertions, 0 deletions

diff --git a/Cargo.lock b/Cargo.lock
index 1fadd29..61ede9e 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -27,6 +27,41 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "790eea4361631c5e7d22598ecd5723ff611904e3344ce8720784c93e3d83d40b"
 
 [[package]]
+name = "darling"
+version = "0.20.11"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "fc7f46116c46ff9ab3eb1597a45688b6715c6e628b5c133e288e709a29bcb4ee"
+dependencies = [
+ "darling_core",
+ "darling_macro",
+]
+
+[[package]]
+name = "darling_core"
+version = "0.20.11"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0d00b9596d185e565c2207a0b01f8bd1a135483d02d9b7b0a54b11da8d53412e"
+dependencies = [
+ "fnv",
+ "ident_case",
+ "proc-macro2",
+ "quote",
+ "strsim",
+ "syn",
+]
+
+[[package]]
+name = "darling_macro"
+version = "0.20.11"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "fc34b93ccb385b40dc71c6fceac4b2ad23662c7eeb248cf10d529b7e055b6ead"
+dependencies = [
+ "darling_core",
+ "quote",
+ "syn",
+]
+
+[[package]]
 name = "defmt"
 version = "0.3.100"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -77,6 +112,36 @@ dependencies = [
 ]
 
 [[package]]
+name = "embassy-executor"
+version = "0.9.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "06070468370195e0e86f241c8e5004356d696590a678d47d6676795b2e439c6b"
+dependencies = [
+ "critical-section",
+ "document-features",
+ "embassy-executor-macros",
+ "embassy-executor-timer-queue",
+]
+
+[[package]]
+name = "embassy-executor-macros"
+version = "0.7.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "dfdddc3a04226828316bf31393b6903ee162238576b1584ee2669af215d55472"
+dependencies = [
+ "darling",
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
+[[package]]
+name = "embassy-executor-timer-queue"
+version = "0.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2fc328bf943af66b80b98755db9106bf7e7471b0cf47dc8559cd9a6be504cc9c"
+
+[[package]]
 name = "embassy-futures"
 version = "0.1.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -86,7 +151,9 @@ checksum = "dc2d050bdc5c21e0862a89256ed8029ae6c290a93aecefc73084b3002cdebb01"
 name = "embassy-ha"
 version = "0.1.0"
 dependencies = [
+ "critical-section",
  "defmt 1.0.1",
+ "embassy-executor",
  "embassy-futures",
  "embassy-net",
  "embassy-sync",
@@ -96,6 +163,7 @@ dependencies = [
  "heapless 0.9.2",
  "serde",
  "serde-json-core",
+ "static_cell",
 ]
 
 [[package]]
@@ -151,6 +219,7 @@ dependencies = [
  "defmt 1.0.1",
  "document-features",
  "embassy-time-driver",
+ "embassy-time-queue-utils",
  "embedded-hal 0.2.7",
  "embedded-hal 1.0.0",
  "embedded-hal-async",
@@ -167,6 +236,16 @@ dependencies = [
 ]
 
 [[package]]
+name = "embassy-time-queue-utils"
+version = "0.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "80e2ee86063bd028a420a5fb5898c18c87a8898026da1d4c852af2c443d0a454"
+dependencies = [
+ "embassy-executor-timer-queue",
+ "heapless 0.8.0",
+]
+
+[[package]]
 name = "embedded-hal"
 version = "0.2.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -234,6 +313,12 @@ dependencies = [
 ]
 
 [[package]]
+name = "fnv"
+version = "1.0.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1"
+
+[[package]]
 name = "futures-core"
 version = "0.3.31"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -277,6 +362,12 @@ dependencies = [
 ]
 
 [[package]]
+name = "ident_case"
+version = "1.0.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b9e0384b61958566e926dc50660321d12159025e767c18e043daf26b70104c39"
+
+[[package]]
 name = "litrs"
 version = "1.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -304,6 +395,12 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8d5439c4ad607c3c23abf66de8c8bf57ba8adcd1f129e699851a6e43935d339d"
 
 [[package]]
+name = "portable-atomic"
+version = "1.11.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f84267b20a16ea918e43c6a88433c2d54fa145c92a811b5b047ccbe153674483"
+
+[[package]]
 name = "proc-macro-error-attr2"
 version = "2.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -411,6 +508,21 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "6ce2be8dc25455e1f91df71bfa12ad37d7af1092ae736f3a6cd0e37bc7810596"
 
 [[package]]
+name = "static_cell"
+version = "2.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0530892bb4fa575ee0da4b86f86c667132a94b74bb72160f58ee5a4afec74c23"
+dependencies = [
+ "portable-atomic",
+]
+
+[[package]]
+name = "strsim"
+version = "0.11.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7da8b5736845d9f2fcb837ea5d9e2628564b3b043a70948a3f0b778838c5fb4f"
+
+[[package]]
 name = "syn"
 version = "2.0.111"
 source = "registry+https://github.com/rust-lang/crates.io-index"
diff --git a/Cargo.toml b/Cargo.toml
index 908e183..d37e957 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -14,3 +14,11 @@ defmt = "1.0.1"
 embassy-sync = { version = "0.7.2", features = ["defmt"] }
 embassy-futures = "0.1.2"
 embedded-io-async = "0.6"
+
+[dev-dependencies]
+embassy-time = { version = "0.5.0", features = ["std"] }
+embassy-executor = { version = "0.9.1", features = ["arch-std", "executor-thread"] }
+embassy-sync = { version = "0.7.2" }
+static_cell = "2.1.1"
+embedded-io-async = { version = "0.6", features = ["std"] }
+critical-section = { version = "1", features = ["std"] }
diff --git a/examples/constant-temperature.rs b/examples/constant-temperature.rs
new file mode 100644
index 0000000..27e1076
--- /dev/null
+++ b/examples/constant-temperature.rs
@@ -0,0 +1,171 @@
+use std::{
+    io::{Read, Write},
+    net::{TcpStream, ToSocketAddrs},
+    sync::{Arc, Mutex},
+    thread::JoinHandle,
+};
+
+use embassy_executor::{Executor, Spawner};
+use embassy_sync::waitqueue::AtomicWaker;
+use embassy_time::Timer;
+use static_cell::StaticCell;
+
+static EXECUTOR: StaticCell<Executor> = StaticCell::new();
+static RESOURCES: StaticCell<embassy_ha::DeviceResources> = StaticCell::new();
+
+struct AsyncTcp {
+    write_handle: JoinHandle<()>,
+    write_buffer: Arc<Mutex<Vec<u8>>>,
+    read_buffer: Arc<Mutex<Vec<u8>>>,
+    waker: Arc<AtomicWaker>,
+}
+
+impl AsyncTcp {
+    fn connect(addr: impl ToSocketAddrs) -> Self {
+        let stream = TcpStream::connect(addr).expect("failed to connect to remote");
+        let mut read_stream = stream.try_clone().unwrap();
+        let mut write_stream = stream;
+
+        let read_buffer: Arc<Mutex<Vec<u8>>> = Default::default();
+        let write_buffer: Arc<Mutex<Vec<u8>>> = Default::default();
+
+        let waker = Arc::new(AtomicWaker::new());
+
+        let write_handle = std::thread::spawn({
+            let write_buffer = write_buffer.clone();
+            move || {
+                loop {
+                    let buffer = {
+                        let mut buffer = write_buffer.lock().unwrap();
+                        std::mem::take(&mut *buffer)
+                    };
+                    if !buffer.is_empty() {
+                        println!("writing {} bytes", buffer.len());
+                        write_stream.write_all(&buffer).unwrap();
+                        write_stream.flush().unwrap();
+                    } else {
+                        std::thread::park();
+                    }
+                }
+            }
+        });
+
+        std::thread::spawn({
+            let read_buffer = read_buffer.clone();
+            let waker = waker.clone();
+            move || {
+                let mut scratch = [0u8; 1024];
+                loop {
+                    let n = read_stream.read(&mut scratch).unwrap();
+                    if n == 0 {
+                        panic!("EOF");
+                    }
+
+                    {
+                        let mut buffer = read_buffer.lock().unwrap();
+                        buffer.extend_from_slice(&scratch[..n]);
+                        waker.wake();
+                    }
+                }
+            }
+        });
+
+        Self {
+            write_handle,
+            write_buffer,
+            read_buffer,
+            waker,
+        }
+    }
+}
+
+impl embedded_io_async::ErrorType for AsyncTcp {
+    type Error = std::io::Error;
+}
+
+impl embedded_io_async::Write for AsyncTcp {
+    async fn write(&mut self, buf: &[u8]) -> Result<usize, Self::Error> {
+        {
+            let mut buffer = self.write_buffer.lock().unwrap();
+            buffer.extend_from_slice(buf);
+        }
+        self.write_handle.thread().unpark();
+        Ok(buf.len())
+    }
+}
+
+impl embedded_io_async::Read for AsyncTcp {
+    async fn read(&mut self, buf: &mut [u8]) -> Result<usize, Self::Error> {
+        struct WaitForWaker<'a>(&'a AtomicWaker, bool);
+
+        impl<'a> Future for WaitForWaker<'a> {
+            type Output = ();
+
+            fn poll(
+                mut self: std::pin::Pin<&mut Self>,
+                cx: &mut std::task::Context<'_>,
+            ) -> std::task::Poll<Self::Output> {
+                if self.1 {
+                    std::task::Poll::Ready(())
+                } else {
+                    self.as_mut().1 = true;
+                    self.0.register(cx.waker());
+                    std::task::Poll::Pending
+                }
+            }
+        }
+
+        loop {
+            {
+                let mut buffer = self.read_buffer.lock().unwrap();
+                if !buffer.is_empty() {
+                    let copy_n = buf.len().min(buffer.len());
+                    buf[..copy_n].copy_from_slice(&buffer[..copy_n]);
+                    buffer.drain(..copy_n);
+                    return Ok(copy_n);
+                }
+            }
+            WaitForWaker(&self.waker, false).await
+        }
+    }
+}
+
+#[embassy_executor::task]
+async fn main_task(spawner: Spawner) {
+    let mut stream = AsyncTcp::connect("mqtt.d464.sh:1883");
+
+    let mut device = embassy_ha::Device::new(
+        RESOURCES.init(Default::default()),
+        embassy_ha::DeviceConfig {
+            device_id: "example-device-id",
+            device_name: "Example Device Name",
+            manufacturer: "Example Device Manufacturer",
+            model: "Example Device Model",
+        },
+    );
+
+    let temperature_sensor = device.create_temperature_sensor(
+        "temperature-sensor-id",
+        "Temperature Sensor Name",
+        embassy_ha::TemperatureUnit::Celcius,
+    );
+
+    spawner.must_spawn(temperature(temperature_sensor));
+
+    device.run(&mut stream).await;
+}
+
+#[embassy_executor::task]
+async fn temperature(mut sensor: embassy_ha::TemperatureSensor<'static>) {
+    loop {
+        sensor.publish(42.0);
+        Timer::after_secs(1).await;
+    }
+}
+
+fn main() {
+    let executor = EXECUTOR.init(Executor::new());
+    executor.run(|spawner| {
+        spawner.must_spawn(main_task(spawner));
+    });
+}