, DeskSubscription>,
+ cancel_subscription_tx: oneshot::Sender<()>,
+ disconnect_on_drop: bool,
+ },
+ Disconnected {
+ peripheral: P,
+ },
+}
+
+impl UpliftDesk {
+ pub fn new(peripheral: P) -> Self {
+ Self::Disconnected { peripheral }
+ }
+
+ pub async fn connect(&mut self) -> Result<()> {
+ if !matches!(self, UpliftDesk::Connected { peripheral, .. } if peripheral.is_connected().await?)
+ {
+ let peripheral = self.peripheral().clone();
+
+ log::debug!("{{{}}} - Connecting to desk", peripheral.id());
+
+ peripheral.connect().await?;
+ peripheral.discover_services().await?;
+
+ let (service, desk_uuids) = peripheral.services().iter().find_map(|service| {
+ for desk in DESK_UUIDS {
+ if desk.service_uuid == service.uuid {
+ return Some((service.clone(), desk));
+ }
+ }
+ None
+ }).ok_or_else(|| {
+ log::warn!("{{{}}} - Attempted to connect to desk that isn't advertising the necessary Service or Characteristics", peripheral.id());
+ Error::DeviceNotFound
+ })?;
+
+ let (data_in_characteristic, data_out_characteristic, name_characteristic) =
+ get_characteristics(desk_uuids, service.characteristics)?;
+
+ let (cancel_subscription_tx, cancel_subscription_rx) = oneshot::channel();
+
+ // we're about to drop our non-connected self, we don't want to drop our connection as we're establishing it
+ if let UpliftDesk::Connected {
+ disconnect_on_drop, ..
+ } = self
+ {
+ *disconnect_on_drop = true;
+ }
+
+ *self = Self::Connected {
+ peripheral,
+ disconnect_on_drop: false,
+ data_in_characteristic,
+ data_out_characteristic,
+ name_characteristic,
+ subscription: Either::Left(cancel_subscription_rx),
+ cancel_subscription_tx,
+ };
+ }
+
+ Ok(())
+ }
+
+ pub fn id(&self) -> UpliftDeskId {
+ UpliftDeskId(self.peripheral().id())
+ }
+
+ pub async fn rssi(&self) -> Result> {
+ match self {
+ UpliftDesk::Connected { peripheral, .. } => {
+ Ok(peripheral.properties().await?.and_then(|p| p.rssi))
+ }
+ UpliftDesk::Disconnected { .. } => Err(Error::NotConnected),
+ }
+ }
+
+ pub async fn name(&self) -> Result> {
+ match self {
+ UpliftDesk::Connected { peripheral, .. } => {
+ Ok(peripheral.properties().await?.and_then(|p| p.local_name))
+ }
+ UpliftDesk::Disconnected { .. } => Err(Error::NotConnected),
+ }
+ }
+
+ pub async fn command(&self, command: Command) -> Result<()> {
+ match self {
+ UpliftDesk::Connected {
+ peripheral,
+ data_in_characteristic,
+ ..
+ } if peripheral.is_connected().await? => {
+ peripheral
+ .command(command, data_in_characteristic, WriteType::WithoutResponse)
+ .await
+ }
+ _ => Err(Error::NotConnected),
+ }
+ }
+
+ pub async fn command_retry(&self, command: Command) -> Result<()> {
+ self.command(command).await?;
+
+ sleep(RETRY_DELAY).await;
+
+ self.command(command).await?;
+
+ Ok(())
+ }
+
+ pub async fn height(&mut self) -> Result {
+ self.get_subscription().await?.height().await
+ }
+
+ pub async fn physical_limits(&mut self) -> Result> {
+ Ok(self.get_subscription().await?.physical_limits())
+ }
+
+ pub async fn height_unit(&mut self) -> Result {
+ Ok(self.get_subscription().await?.height_unit())
+ }
+
+ pub async fn limited_status(&mut self) -> Result {
+ Ok(self.get_subscription().await?.limited_status())
+ }
+
+ pub async fn height_stream(&mut self) -> Result + use> {
+ self.get_subscription().await?.height_stream().await
+ }
+
+ /// It would be nice to return our desk here but btleplug drops any knowledge of this peripheral on disconnect
+ pub async fn disconnect(&mut self) -> Result<()> {
+ if let UpliftDesk::Connected {
+ peripheral,
+ cancel_subscription_tx,
+ disconnect_on_drop: dont_disconnect,
+ ..
+ } = self
+ {
+ disconnect(
+ peripheral,
+ mem::replace(cancel_subscription_tx, oneshot::channel().0),
+ )
+ .await?;
+ // we've already disconnected, we don't need a background job when we drop
+ *dont_disconnect = true;
+ }
+ let peripheral = self.peripheral().clone();
+ *self = UpliftDesk::Disconnected { peripheral };
+
+ Ok(())
+ }
+
+ pub async fn get_subscription(&mut self) -> Result<&DeskSubscription
> {
+ match self {
+ UpliftDesk::Connected {
+ peripheral,
+ data_in_characteristic,
+ data_out_characteristic,
+ subscription,
+ ..
+ } if peripheral.is_connected().await? => {
+ // subscription.as_ref()
+
+ if let Either::Left(cancel_rx) = subscription.as_mut() {
+ let cancel_rx = mem::replace(cancel_rx, oneshot::channel().1);
+
+ *subscription = Either::Right(
+ DeskSubscription::new(
+ peripheral.id(),
+ data_in_characteristic.clone(),
+ data_out_characteristic.clone(),
+ peripheral.clone(),
+ cancel_rx,
+ )
+ .await?,
+ )
+ }
+
+ let subscription = subscription.as_ref().right().unwrap();
+ if !subscription.is_live() {
+ // our subscription exists but the main loop isn't running
+ Err(Error::NotConnected)?
+ }
+
+ Ok(subscription)
+ }
+ _ => Err(Error::NotConnected),
+ }
+ }
+
+ fn peripheral(&self) -> &P {
+ match self {
+ UpliftDesk::Connected { peripheral, .. } => peripheral,
+ UpliftDesk::Disconnected { peripheral } => peripheral,
+ }
+ }
+}
+
+impl Debug for UpliftDesk {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ f.debug_struct("UpliftDesk")
+ .field("id", &self.id())
+ .finish()
+ }
+}
+
+impl Drop for UpliftDesk {
+ fn drop(&mut self) {
+ if let UpliftDesk::Connected {
+ peripheral,
+ cancel_subscription_tx,
+ disconnect_on_drop: false,
+ ..
+ } = self
+ {
+ let peripheral = peripheral.clone();
+ let cancel = mem::replace(cancel_subscription_tx, oneshot::channel().0);
+ tokio::spawn(async move {
+ if let Err(e) = disconnect(&peripheral, cancel).await {
+ log::warn!("{{{}}} - Error while dropping desk: {e:?}", peripheral.id());
+ }
+ });
+ }
+ }
+}
+
+async fn disconnect(
+ peripheral: &P,
+ cancel_subscription_tx: oneshot::Sender<()>,
+) -> Result<()> {
+ log::debug!("{{{}}} - Disconnecting from Desk", peripheral.id());
+ if cancel_subscription_tx.send(()).is_err() {
+ log::error!(
+ "{{{}}} - We have already cancelled our subscription",
+ peripheral.id()
+ );
+ }
+ peripheral.disconnect().await
+}
+
+fn get_characteristics(
+ desk_characteristic: DeskUuids,
+ characteristics: BTreeSet,
+) -> Result<(Characteristic, Characteristic, Characteristic)> {
+ let mut data_in_characteristic = None;
+ let mut data_out_characteristic = None;
+ let mut name_characteristic = None;
+
+ for characteristic in characteristics.into_iter() {
+ if desk_characteristic.data_in == characteristic.uuid {
+ data_in_characteristic = Some(characteristic);
+ } else if desk_characteristic.data_out == characteristic.uuid {
+ data_out_characteristic = Some(characteristic);
+ } else if desk_characteristic.name_uuid == characteristic.uuid {
+ name_characteristic = Some(characteristic);
+ }
+ }
+
+ Ok((
+ data_in_characteristic.ok_or_else(|| Error::NoSuchCharacteristic)?,
+ data_out_characteristic.ok_or_else(|| Error::NoSuchCharacteristic)?,
+ name_characteristic.ok_or_else(|| Error::NoSuchCharacteristic)?,
+ ))
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::mock_btle::*;
+ use btleplug::api::{CharPropFlags, Characteristic, Service};
+ use tokio::runtime::Runtime;
+
+ fn valid_characteristics() -> BTreeSet {
+ [
+ DESK_UUIDS[0].data_in,
+ DESK_UUIDS[0].data_out,
+ DESK_UUIDS[0].name_uuid,
+ ]
+ .into_iter()
+ .map(|uuid| Characteristic {
+ uuid,
+ service_uuid: DESK_UUIDS[0].service_uuid,
+ properties: CharPropFlags::empty(),
+ descriptors: BTreeSet::new(),
+ })
+ .collect()
+ }
+
+ #[test]
+ fn test_new_desk() {
+ let runtime = Runtime::new().unwrap();
+
+ runtime.block_on(async {
+ let mut peripheral_a = MockPeripheral::new();
+
+ let mut disconnect_peripheral = MockPeripheral::new();
+ disconnect_peripheral
+ .expect_disconnect()
+ .returning(|| Ok(()));
+
+ let mut peripheral_b = MockPeripheral::new();
+ peripheral_b.expect_connect().returning(|| Ok(()));
+ peripheral_b.expect_discover_services().returning(|| Ok(()));
+ peripheral_b.expect_services().returning(|| {
+ [Service {
+ uuid: DESK_UUIDS[0].service_uuid,
+ primary: true,
+ characteristics: valid_characteristics(),
+ }]
+ .into_iter()
+ .collect()
+ });
+ peripheral_b
+ .expect_clone()
+ .return_once(move || disconnect_peripheral);
+ peripheral_a
+ .expect_clone()
+ .return_once(move || peripheral_b);
+
+ let mut desk = UpliftDesk::new(peripheral_a);
+ desk.connect().await.unwrap();
+ drop(desk);
+ });
+
+ // This will log panics from tokio threads, but it won't fail our test :( we need
+ // https://docs.rs/tokio/latest/tokio/runtime/struct.Builder.html#method.unhandled_panic
+ drop(runtime);
+ }
+
+ #[tokio::test]
+ async fn test_invalid_peripheral() {
+ let mut peripheral = MockPeripheral::new();
+
+ let mut drop_peripheral = MockPeripheral::new();
+ drop_peripheral.expect_connect().returning(|| Ok(()));
+ drop_peripheral
+ .expect_discover_services()
+ .returning(|| Ok(()));
+ drop_peripheral.expect_services().returning(BTreeSet::new);
+
+ peripheral
+ .expect_clone()
+ .return_once(move || drop_peripheral);
+
+ assert!(matches!(
+ UpliftDesk::new(peripheral).connect().await.err().unwrap(),
+ Error::DeviceNotFound
+ ));
+ }
+
+ // fn mock_central(setup_peripheral: F) -> MockCentral
+ // where
+ // F: FnOnce(&mut MockPeripheral),
+ // {
+ // let mut central = MockCentral::new();
+ // let mut peripheral = MockPeripheral::new();
+ //
+ // peripheral.expect_connect().returning(|| Ok(()));
+ // peripheral.expect_discover_services().returning(|| Ok(()));
+ // peripheral.expect_properties().returning(|| {
+ // Ok(Some(PeripheralProperties {
+ // services: vec![DESK_UUIDS[0].service_uuid],
+ // ..Default::default()
+ // }))
+ // });
+ // peripheral.expect_services().returning(move || {
+ // let mut services = BTreeSet::new();
+ // services.insert(Service {
+ // uuid: DESK_UUIDS[0].service_uuid,
+ // primary: true,
+ // characteristics: valid_characteristics(),
+ // });
+ //
+ // services
+ // });
+ //
+ // let mut drop_peripheral = MockPeripheral::new();
+ // drop_peripheral
+ // .expect_disconnect()
+ // .once()
+ // .returning(|| Ok(()));
+ // peripheral
+ // .expect_clone()
+ // .return_once(move || drop_peripheral);
+ //
+ // setup_peripheral(&mut peripheral);
+ //
+ // central.expect_peripheral().return_once(|_| Ok(peripheral));
+ //
+ // central
+ // }
+ //
+ // fn valid_characteristics() -> BTreeSet {
+ // let mut characteristics = BTreeSet::new();
+ //
+ // for uuid in [
+ // DESK_UUIDS[0].data_in,
+ // DESK_UUIDS[0].data_out,
+ // DESK_UUIDS[0].name_uuid,
+ // ] {
+ // characteristics.insert(Characteristic {
+ // uuid,
+ // service_uuid: DESK_UUIDS[0].service_uuid,
+ // properties: CharPropFlags::empty(),
+ // descriptors: BTreeSet::new(),
+ // });
+ // }
+ //
+ // characteristics
+ // }
+}
diff --git a/crates/uplift-lib/src/desk/subscription.rs b/crates/uplift-lib/src/desk/subscription.rs
new file mode 100644
index 0000000..6befd64
--- /dev/null
+++ b/crates/uplift-lib/src/desk/subscription.rs
@@ -0,0 +1,305 @@
+use crate::api::{Command, EnhancedPeripheral as _, HeightUnit, Message};
+use crate::{LimitedDirection, TouchMode};
+use btleplug::Error;
+use btleplug::api::{Characteristic, Peripheral, ValueNotification, WriteType};
+use btleplug::platform::PeripheralId;
+use std::future::Future;
+use std::ops::Deref;
+use std::time::Duration;
+use tokio::sync::{oneshot, watch};
+use tokio::task::JoinHandle;
+use tokio::time::timeout;
+use tokio::{select, time};
+use tokio_stream::wrappers::WatchStream;
+use tokio_stream::{Stream, StreamExt};
+use uom::si::f32::Length;
+use uom::si::length::{centimeter, inch};
+
+pub struct DeskSubscription {
+ height_rx: watch::Receiver>,
+ physical_limits_rx: watch::Receiver>,
+ height_unit_rx: watch::Receiver,
+ touch_mode_rx: watch::Receiver,
+ limited_status_rx: watch::Receiver,
+ data_in_characteristic: Characteristic,
+ peripheral: P,
+ subscription_live: JoinHandle>,
+}
+
+/// The app starts up by sending
+///
+/// t:400 [Command::FetchHighLowLimit]
+/// t:500 [Command::FetchHighLowLimit]
+/// t:600 [Command::FetchHeightValue]
+/// t:700 [Command::FetchHeightRange]
+/// t:? [Command::LockInfo]
+/// t:? [Command::Handshake]
+/// t:1100 [Command::Handshake]
+impl DeskSubscription {
+ pub(super) async fn new(
+ id: PeripheralId,
+ data_in_characteristic: Characteristic,
+ data_out_characteristic: Characteristic,
+ peripheral: P,
+ cancel_rx: oneshot::Receiver<()>,
+ ) -> btleplug::Result {
+ let (height_tx, height_rx) = watch::channel(None);
+ let (physical_limits_tx, physical_limits_rx) = watch::channel(None);
+ // The app seems to default to inches
+ let (height_unit_tx, height_unit_rx) = watch::channel(HeightUnit::Inch);
+ let (touch_mode_tx, touch_mode_rx) = watch::channel(TouchMode::Continuous);
+ let (limited_status_tx, limited_status_rx) = watch::channel(LimitedDirection::None);
+
+ let receiver = peripheral.notifications().await?;
+ peripheral.subscribe(&data_out_characteristic).await?;
+
+ // spawn the initial commands
+ tokio::spawn({
+ let peripheral = peripheral.clone();
+ let data_in_characteristic = data_in_characteristic.clone();
+ async move {
+ // The app specifically sleeps for 100ms between commands
+ let mut interval = time::interval(Duration::from_millis(100));
+ for cmd in [
+ Command::Handshake,
+ Command::FetchHighLowLimit,
+ Command::FetchHighLowLimit,
+ Command::FetchHeightRange,
+ Command::FetchHeightValue,
+ Command::Handshake,
+ ] {
+ interval.tick().await;
+ if let Err(e) = peripheral
+ .command(cmd, &data_in_characteristic, WriteType::WithoutResponse)
+ .await
+ {
+ log::warn!("Failed to write `{cmd:?}`: {e:?}");
+ break;
+ }
+ }
+ }
+ });
+
+ let subscription_live = tokio::spawn({
+ let peripheral = peripheral.clone();
+ async move {
+ let result = subscription_main_loop(
+ receiver,
+ height_tx,
+ physical_limits_tx,
+ height_unit_tx,
+ touch_mode_tx,
+ limited_status_tx,
+ cancel_rx,
+ data_out_characteristic,
+ peripheral,
+ )
+ .await;
+
+ if let Err(e) = &result {
+ log::error!("{{{id}}} - Error in our subscription task: {e:?}")
+ }
+
+ result
+ }
+ });
+
+ Ok(Self {
+ height_rx,
+ physical_limits_rx,
+ height_unit_rx,
+ touch_mode_rx,
+ limited_status_rx,
+ data_in_characteristic,
+ peripheral,
+ subscription_live,
+ })
+ }
+
+ pub fn height(&self) -> impl Future