feat(lib): dual connections for the D1301 (Sleep A30)

lib/src/devices/soundcore/common/device.rs

@@ -267,6 +267,10 @@ where
         self.change_notify.clone()
     }
 
+    pub fn state_sender(&self) -> &watch::Sender<StateType> {
+        &self.state_sender
+    }
+
     pub fn sound_modes(&mut self, available_sound_modes: AvailableSoundModes)
     where
         StateType: Has<SoundModes>,
@@ -836,13 +840,42 @@ pub mod test_utils {
             config: SoundcoreDeviceConfig,
             delays: HashMap<packet::Command, Duration>,
         ) -> Self
+        where
+            StateType: Clone + Send + Sync + 'static,
+            SoundcoreDeviceRegistry<StateType>: BuildDevice<StateType>,
+        {
+            let database = Arc::new(OpenSCQ30Database::new_in_memory().await.unwrap());
+            Self::new_with_database(
+                constructor,
+                device_model,
+                packet_responses,
+                config,
+                delays,
+                database,
+            )
+            .await
+        }
+
+        /// Allows sharing a database between multiple connections, for testing things that are
+        /// persisted across reconnects.
+        pub async fn new_with_database<StateType>(
+            constructor: fn(
+                Arc<dyn RfcommBackend + Send + Sync>,
+                Arc<OpenSCQ30Database>,
+                DeviceModel,
+            ) -> SoundcoreDeviceRegistry<StateType>,
+            device_model: DeviceModel,
+            packet_responses: HashMap<Command, packet::Inbound>,
+            config: SoundcoreDeviceConfig,
+            delays: HashMap<packet::Command, Duration>,
+            database: Arc<OpenSCQ30Database>,
+        ) -> Self
         where
             StateType: Clone + Send + Sync + 'static,
             SoundcoreDeviceRegistry<StateType>: BuildDevice<StateType>,
         {
             let (inbound_sender, inbound_receiver) = mpsc::channel(100);
             let (outbound_sender, mut outbound_receiver) = mpsc::channel(100);
-            let database = Arc::new(OpenSCQ30Database::new_in_memory().await.unwrap());
 
             let registry = constructor(
                 Arc::new(MockRfcommBackend::new(inbound_receiver, outbound_sender)),

lib/src/devices/soundcore/d1301.rs

@@ -3,6 +3,7 @@ use std::collections::HashMap;
 use crate::{
     devices::soundcore::{
         common::{
+            self,
             macros::soundcore_device,
             modules::{
                 button_configuration::{
@@ -11,7 +12,7 @@ use crate::{
                 dual_battery_level::DualBatteryLevelConfiguration,
             },
             packet::{
-                inbound::TryToPacket,
+                inbound::{DualConnectionsDevicePacket, TryToPacket},
                 outbound::{RequestState, ToPacket},
             },
             structures::button_configuration::{
@@ -43,10 +44,15 @@ soundcore_device!(
             .send_with_response(&packets::outbound::request_alarms())
             .await?
             .try_to_packet()?;
+        // Unlike other models, the D1301 answers this even while dual connections is disabled, and the
+        // state update packet has no enabled flag, so always ask.
+        let dual_connections_devices =
+            common::modules::dual_connections::take_dual_connection_devices(&packet_io).await?;
         Ok(state::D1301State::new(
             state_update_packet,
             auto_stop_timer_packet.0,
             alarms_packet.0,
+            modules::dual_connections::initial_dual_connections(dual_connections_devices),
         ))
     },
     async |builder| {
@@ -60,6 +66,8 @@ soundcore_device!(
 
         builder.d1301_alarms();
 
+        builder.d1301_dual_connections().await;
+
         // misc
         builder.d1301_auto_stop_timer();
         builder.d1301_auto_switch_once_asleep();
@@ -96,6 +104,10 @@ soundcore_device!(
                 packets::inbound::AlarmsPacket::COMMAND,
                 packets::inbound::AlarmsPacket(Vec::new()).to_packet(),
             ),
+            (
+                DualConnectionsDevicePacket::COMMAND,
+                DualConnectionsDevicePacket::demo().to_packet(),
+            ),
         ])
     },
 );
@@ -337,4 +349,145 @@ mod tests {
 
         device.assert_setting_values(vec![(SettingId::Alarm1Volume, 51.into())]);
     }
+
+    mod dual_connections {
+        use std::{sync::Arc, time::Duration};
+
+        use macaddr::MacAddr6;
+
+        use super::*;
+        use crate::{
+            devices::soundcore::common::structures::DualConnectionsDevice,
+            storage::OpenSCQ30Database,
+        };
+
+        fn packet_responses(
+            dual_connections_devices: Vec<DualConnectionsDevice>,
+        ) -> HashMap<packet::Command, packet::Inbound> {
+            HashMap::from([
+                (
+                    RequestState::COMMAND,
+                    D1301StateUpdatePacket::default().to_packet(),
+                ),
+                (
+                    packets::inbound::AutoStopTimerPacket::COMMAND,
+                    packets::inbound::AutoStopTimerPacket(structures::AutoStopTimer::default())
+                        .to_packet(),
+                ),
+                (
+                    packets::inbound::AlarmsPacket::COMMAND,
+                    packets::inbound::AlarmsPacket(Vec::new()).to_packet(),
+                ),
+                (
+                    DualConnectionsDevicePacket::COMMAND,
+                    DualConnectionsDevicePacket {
+                        total_packets: 1,
+                        current_packet_index: 1,
+                        devices: dual_connections_devices,
+                    }
+                    .to_packet(),
+                ),
+                // captured from a real device: echoes the value followed by a 0 status byte
+                (
+                    packet::Command([0x0b, 0x84]),
+                    packet::Inbound::new(packet::Command([0x0b, 0x84]), vec![1, 0]),
+                ),
+            ])
+        }
+
+        fn host(last_octet: u8, is_connected: bool) -> DualConnectionsDevice {
+            DualConnectionsDevice {
+                is_connected,
+                mac_address: MacAddr6::new(0, 0, 0, 0, 0, last_octet),
+                name: format!("Device {last_octet}"),
+            }
+        }
+
+        async fn connect(
+            database: Arc<OpenSCQ30Database>,
+            dual_connections_devices: Vec<DualConnectionsDevice>,
+        ) -> TestSoundcoreDevice {
+            TestSoundcoreDevice::new_with_database(
+                super::super::device_registry,
+                DeviceModel::SoundcoreD1301,
+                packet_responses(dual_connections_devices),
+                SoundcoreDeviceConfig::default(),
+                HashMap::<packet::Command, Duration>::new(),
+                database,
+            )
+            .await
+        }
+
+        async fn database() -> Arc<OpenSCQ30Database> {
+            Arc::new(OpenSCQ30Database::new_in_memory().await.unwrap())
+        }
+
+        #[tokio::test(start_paused = true)]
+        async fn is_off_when_unknown() {
+            let device = connect(database().await, vec![host(1, true), host(2, false)]).await;
+            device.assert_setting_values([(SettingId::DualConnections, false.into())]);
+        }
+
+        #[tokio::test(start_paused = true)]
+        async fn is_on_when_two_devices_are_connected() {
+            let device = connect(database().await, vec![host(1, true), host(2, true)]).await;
+            device.assert_setting_values([(SettingId::DualConnections, true.into())]);
+        }
+
+        #[tokio::test(start_paused = true)]
+        async fn enabling_sends_packet() {
+            let mut device = connect(database().await, vec![host(1, true)]).await;
+            device
+                .assert_set_settings_response(
+                    vec![(SettingId::DualConnections, true.into())],
+                    vec![packet::outbound::set_dual_connections_enabled(true)],
+                )
+                .await;
+            device.assert_setting_values([(SettingId::DualConnections, true.into())]);
+        }
+
+        #[tokio::test(start_paused = true)]
+        async fn enabled_is_reapplied_on_reconnect() {
+            let database = database().await;
+            let mut device = connect(database.clone(), vec![host(1, true)]).await;
+            device
+                .set_settings(vec![(SettingId::DualConnections, true.into())])
+                .await;
+            drop(device);
+
+            // simulate the earbuds having been powered off, which disables dual connections
+            let device = connect(database, vec![host(1, true)]).await;
+            device.assert_setting_values([(SettingId::DualConnections, true.into())]);
+        }
+
+        #[tokio::test(start_paused = true)]
+        async fn enabling_while_already_on_is_reapplied_on_reconnect() {
+            let database = database().await;
+            // two connected devices: already shown as enabled, but nothing has been saved yet
+            let mut device = connect(database.clone(), vec![host(1, true), host(2, true)]).await;
+            device
+                .set_settings(vec![(SettingId::DualConnections, true.into())])
+                .await;
+            drop(device);
+
+            let device = connect(database, vec![host(1, true)]).await;
+            device.assert_setting_values([(SettingId::DualConnections, true.into())]);
+        }
+
+        #[tokio::test(start_paused = true)]
+        async fn disabled_is_not_reapplied_on_reconnect() {
+            let database = database().await;
+            let mut device = connect(database.clone(), vec![host(1, true)]).await;
+            device
+                .set_settings(vec![(SettingId::DualConnections, true.into())])
+                .await;
+            device
+                .set_settings(vec![(SettingId::DualConnections, false.into())])
+                .await;
+            drop(device);
+
+            let device = connect(database, vec![host(1, true)]).await;
+            device.assert_setting_values([(SettingId::DualConnections, false.into())]);
+        }
+    }
 }

lib/src/devices/soundcore/d1301/modules.rs

@@ -3,6 +3,7 @@ use crate::devices::soundcore::{common::device::SoundcoreDeviceBuilder, d1301::s
 mod alarms;
 mod auto_stop_timer;
 mod auto_switch_once_asleep;
+pub mod dual_connections;
 mod flags;
 mod listening_mode;
 

lib/src/devices/soundcore/d1301/modules/dual_connections.rs

@@ -0,0 +1,134 @@
+//! The D1301 supports the standard dual connections commands (0x0b family), but unlike other models
+//! it neither reports whether dual connections is enabled nor remembers it after powering off. To
+//! make up for that, the last value set through openscq30 is stored and reapplied on connect.
+
+use std::sync::{Arc, Mutex};
+
+use async_trait::async_trait;
+use openscq30_lib_has::Has;
+use tokio::sync::watch;
+
+use crate::{
+    api::{device, settings::SettingId},
+    devices::{
+        DeviceModel,
+        soundcore::{
+            common::{
+                device::SoundcoreDeviceBuilder,
+                modules::ModuleCollection,
+                packet,
+                state_modifier::StateModifier,
+                structures::{DualConnections, DualConnectionsDevice},
+            },
+            d1301::state::D1301State,
+        },
+    },
+    storage::OpenSCQ30Database,
+};
+
+/// The D1301 does not report whether dual connections is enabled, but if more than one device is
+/// connected, it must be.
+pub fn initial_dual_connections(devices: Vec<DualConnectionsDevice>) -> DualConnections {
+    DualConnections {
+        is_enabled: devices.iter().filter(|device| device.is_connected).count() > 1,
+        devices,
+    }
+}
+
+impl SoundcoreDeviceBuilder<D1301State> {
+    pub async fn d1301_dual_connections(&mut self) {
+        let database = self.database();
+        let device_model = self.device_model();
+
+        let saved = match database
+            .fetch_device_model_setting(device_model, SettingId::DualConnections)
+            .await
+        {
+            Ok(value) => value.and_then(|value| value.as_bool()),
+            Err(err) => {
+                tracing::warn!("failed to fetch saved dual connections setting: {err:?}");
+                None
+            }
+        };
+        let is_enabled = Has::<DualConnections>::get(&*self.state_sender().borrow()).is_enabled;
+        if saved == Some(true) && !is_enabled {
+            match self
+                .packet_io_controller()
+                .send_with_response(&packet::outbound::set_dual_connections_enabled(true))
+                .await
+            {
+                Ok(_) => self.state_sender().send_modify(|state| {
+                    Has::<DualConnections>::get_mut(state).is_enabled = true;
+                }),
+                Err(err) => tracing::warn!("failed to reapply dual connections: {err:?}"),
+            }
+        }
+
+        self.dual_connections();
+        self.module_collection()
+            .add_d1301_dual_connections_persistence(database, device_model, saved);
+    }
+}
+
+impl<T> ModuleCollection<T>
+where
+    T: Has<DualConnections> + Clone + Send + Sync,
+{
+    /// Must be added after the dual connections module so that only values that were successfully
+    /// sent to the device get saved.
+    fn add_d1301_dual_connections_persistence(
+        &mut self,
+        database: Arc<OpenSCQ30Database>,
+        device_model: DeviceModel,
+        saved: Option<bool>,
+    ) {
+        self.state_modifiers
+            .push(Box::new(PersistDualConnectionsStateModifier {
+                database,
+                device_model,
+                saved: Mutex::new(saved),
+            }));
+    }
+}
+
+struct PersistDualConnectionsStateModifier {
+    database: Arc<OpenSCQ30Database>,
+    device_model: DeviceModel,
+    /// Compared against the database's value rather than the device's state, since the state can
+    /// already be enabled (two devices connected) without that ever having been saved.
+    saved: Mutex<Option<bool>>,
+}
+
+#[async_trait]
+impl<StateT> StateModifier<StateT> for PersistDualConnectionsStateModifier
+where
+    StateT: Has<DualConnections> + Send + Sync,
+{
+    async fn move_to_state(
+        &self,
+        state_sender: &watch::Sender<StateT>,
+        target_state: &StateT,
+    ) -> device::Result<()> {
+        let is_enabled = target_state.get().is_enabled;
+        // Only save what the dual connections module actually applied to the device.
+        if state_sender.borrow().get().is_enabled != is_enabled
+            || *self.saved.lock().unwrap() == Some(is_enabled)
+        {
+            return Ok(());
+        }
+        // Failing to save shouldn't fail setting the value on the device, which already happened.
+        match self
+            .database
+            .upsert_device_model_setting(
+                self.device_model,
+                SettingId::DualConnections,
+                is_enabled.into(),
+            )
+            .await
+        {
+            Ok(()) => *self.saved.lock().unwrap() = Some(is_enabled),
+            Err(err) => tracing::warn!("failed to save dual connections setting: {err:?}"),
+        }
+        Ok(())
+    }
+}