diff --git a/rmk/src/lib.rs b/rmk/src/lib.rs index 1be3ed35a..2c48d76ff 100644 --- a/rmk/src/lib.rs +++ b/rmk/src/lib.rs @@ -105,6 +105,8 @@ pub mod matrix; pub mod processor; #[cfg(feature = "split")] pub mod split; +#[cfg(feature = "split")] +pub mod split_app; pub mod state; #[cfg(feature = "storage")] pub mod storage; diff --git a/rmk/src/split/driver.rs b/rmk/src/split/driver.rs index 857574716..10ef5e8f7 100644 --- a/rmk/src/split/driver.rs +++ b/rmk/src/split/driver.rs @@ -186,6 +186,12 @@ impl SplitMessage::Wpm(e.0), with_feature("display"): e = modifier_sub.next_event().fuse() => SplitMessage::Modifier(e.modifier.into_bits()), with_feature("display"): e = sleep_sub.next_event().fuse() => SplitMessage::SleepState(e.0), + // Deliberately the last (lowest-priority) outgoing arm. + m = crate::split_app::SPLIT_APP_TX.receive().fuse() => SplitMessage::Application(m), } }; @@ -285,6 +293,7 @@ impl publish_event(e), + SplitMessage::Application(data) => crate::split_app::deliver_received(data), #[cfg(feature = "_ble")] SplitMessage::BatteryStatus(state) => set_peripheral_battery(self.id, state.0), #[cfg(feature = "dfu_split")] diff --git a/rmk/src/split/mod.rs b/rmk/src/split/mod.rs index de5ad13b6..d057676b2 100644 --- a/rmk/src/split/mod.rs +++ b/rmk/src/split/mod.rs @@ -92,6 +92,11 @@ pub(crate) enum SplitMessage { /// Peripheral → Central: confirm mark_updated succeeded, about to reset. #[cfg(feature = "dfu_split")] FirmwareUpdateConfirm, + + /// Opaque application payload, either direction (see `crate::split_app`). + /// Kept as the last variant so the postcard discriminants of existing + /// messages stay stable across halves flashed at different revisions. + Application(crate::split_app::SplitAppData), } // ----------------------------------------------------------------------- diff --git a/rmk/src/split/peripheral.rs b/rmk/src/split/peripheral.rs index f1a84104e..0809af523 100644 --- a/rmk/src/split/peripheral.rs +++ b/rmk/src/split/peripheral.rs @@ -78,6 +78,15 @@ impl SplitPeripheral { /// The peripheral uses the general matrix, does scanning and sends key events through `SplitWriter`. /// It also receives split messages from the central through `SplitReader`. pub(crate) async fn run(&mut self) { + // `run` executes exactly while a central session is up; the guard's + // `Drop` sends the link-down edge even when this future is cancelled + // from outside. Link-up waits for the first message from the central: + // a bare BLE connection is not enough, since trouble-host silently + // drops notifications until the central subscribes (CCCD write), and + // the central's `ConnectionStatus` snapshot arrives right after it + // subscribes — so first message ⇒ bidirectionally live. + let mut app_link = crate::split_app::LinkGuard::new(); + // Proactively announce our firmware hash so the central can detect // us even when it booted first and already gave up waiting for a query response. #[cfg(feature = "dfu_split")] @@ -108,127 +117,20 @@ impl SplitPeripheral { }, e = pointing_sub.next_message_pure().fuse() => SplitMessage::Pointing(e), with_feature("_ble"): e = battery_sub.next_event().fuse() => SplitMessage::BatteryStatus(e), + // Deliberately the last (lowest-priority) outgoing arm, behind key events. + m = crate::split_app::SPLIT_APP_PERIPH_TX.receive().fuse() => SplitMessage::Application(m), } }; match select(self.split_driver.read(), read_message_to_send).await { Either::First(m) => match m { // Process split messages from the central - Ok(split_message) => match split_message { - SplitMessage::ConnectionStatus(status) => { - trace!("Received central connection status: {:?}", status); - update_status(|c| *c = status); - } - #[cfg(all(feature = "_ble", feature = "storage"))] - SplitMessage::ClearPeer => { - // Clear the peer address - FLASH_CHANNEL - .send(crate::storage::FlashOperationMessage::PeerAddress(PeerAddress::new( - 0, false, [0; 6], - ))) - .await; - } - SplitMessage::KeyboardIndicator(indicator) => { - // Publish KeyboardIndicator event - publish_event(LedIndicatorEvent::new( - rmk_types::led_indicator::LedIndicator::from_bits(indicator), - )); - } - SplitMessage::Layer(layer) => { - // Publish Layer event - publish_event(LayerChangeEvent::new(layer)); - } - #[cfg(feature = "display")] - SplitMessage::Wpm(wpm) => publish_event(WpmUpdateEvent::new(wpm)), - #[cfg(feature = "display")] - SplitMessage::Modifier(bits) => { - publish_event(ModifierEvent { - modifier: rmk_types::modifier::ModifierCombination::from_bits(bits), - }); - } - #[cfg(feature = "display")] - SplitMessage::SleepState(sleeping) => { - publish_event(SleepStateEvent::new(sleeping)); - } - // --- dfu_split: firmware update handlers --- - #[cfg(feature = "dfu_split")] - SplitMessage::FirmwareHashQuery => { - let hash = crate::dfu::read_embedded_firmware_hash(); - info!("dfu_split: hash query, responding with {:#x}", hash); - self.split_driver - .write(&SplitMessage::FirmwareHashResponse(hash)) - .await - .ok(); - } - #[cfg(feature = "dfu_split")] - SplitMessage::FirmwareChunk { offset, len, data } => { - if self.dfu_handler.is_none() { - self.dfu_handler = crate::dfu::SplitDfuHandler::new(); - if self.dfu_handler.is_none() { - error!("dfu_split: FlashManager not initialized, skipping chunk"); - continue; - } - } - let handler = self.dfu_handler.as_mut().unwrap(); - let actual_len = len as usize; - let chunk_data = &data.0[..actual_len]; - match handler.write_chunk(offset as u32, chunk_data) { - Ok(()) => { - debug!("dfu_split: wrote {} bytes at offset {}", actual_len, offset); - let ack = SplitMessage::FirmwareChunkAck { - offset, - crc: crate::crc32::crc32(chunk_data), - }; - self.split_driver.write(&ack).await.ok(); - } - Err(()) => error!("dfu_split: write error at offset {}", offset), - } - } - #[cfg(feature = "dfu_split")] - SplitMessage::FirmwareUpdateComplete => { - if let Some(ref mut handler) = self.dfu_handler { - let dfu_crc = handler.compute_dfu_crc(); - info!("dfu_split: DFU partition CRC: {:#010x}", dfu_crc); - let crc_msg = SplitMessage::FirmwareCrcReport(dfu_crc); - self.split_driver.write(&crc_msg).await.ok(); - info!("dfu_split: CRC report sent"); - - let deadline = embassy_time::Instant::now() + embassy_time::Duration::from_secs(5); - let ok = loop { - match select(self.split_driver.read(), embassy_time::Timer::at(deadline)).await { - Either::First(Ok(SplitMessage::FirmwareCrcOk)) => { - info!("dfu_split: central confirmed CRC, resetting"); - break true; - } - Either::First(Ok(SplitMessage::FirmwareCrcFail)) => { - warn!("dfu_split: central rejected CRC, stopping update"); - break false; - } - Either::First(Ok(_)) => {} - Either::First(Err(e)) => { - error!("read error: {:?}", e); - break false; - } - Either::Second(_) => { - error!("timeout"); - break false; - } - } - }; - - if ok { - self.split_driver.write(&SplitMessage::FirmwareUpdateConfirm).await.ok(); - embassy_time::Timer::after_millis(50).await; - handler.mark_updated_and_reset().ok(); - } else { - self.dfu_handler = None; - } - } else { - error!("dfu_split: no active DFU session"); - } - } - _ => (), - }, + Ok(split_message) => { + // First traffic from the central ⇒ the session is + // bidirectionally live. + app_link.mark_up(); + self.handle_central_message(split_message).await; + } Err(e) => { error!("Split message read error: {:?}", e); if let crate::split::driver::SplitDriverError::Disconnected = e { @@ -243,4 +145,124 @@ impl SplitPeripheral { } } } + + /// Process a single message from the central. + async fn handle_central_message(&mut self, split_message: SplitMessage) { + match split_message { + SplitMessage::ConnectionStatus(status) => { + trace!("Received central connection status: {:?}", status); + update_status(|c| *c = status); + } + #[cfg(all(feature = "_ble", feature = "storage"))] + SplitMessage::ClearPeer => { + // Clear the peer address + FLASH_CHANNEL + .send(crate::storage::FlashOperationMessage::PeerAddress(PeerAddress::new( + 0, false, [0; 6], + ))) + .await; + } + SplitMessage::KeyboardIndicator(indicator) => { + // Publish KeyboardIndicator event + publish_event(LedIndicatorEvent::new( + rmk_types::led_indicator::LedIndicator::from_bits(indicator), + )); + } + SplitMessage::Layer(layer) => { + // Publish Layer event + publish_event(LayerChangeEvent::new(layer)); + } + #[cfg(feature = "display")] + SplitMessage::Wpm(wpm) => publish_event(WpmUpdateEvent::new(wpm)), + #[cfg(feature = "display")] + SplitMessage::Modifier(bits) => { + publish_event(ModifierEvent { + modifier: rmk_types::modifier::ModifierCombination::from_bits(bits), + }); + } + #[cfg(feature = "display")] + SplitMessage::SleepState(sleeping) => { + publish_event(SleepStateEvent::new(sleeping)); + } + // --- dfu_split: firmware update handlers --- + #[cfg(feature = "dfu_split")] + SplitMessage::FirmwareHashQuery => { + let hash = crate::dfu::read_embedded_firmware_hash(); + info!("dfu_split: hash query, responding with {:#x}", hash); + self.split_driver + .write(&SplitMessage::FirmwareHashResponse(hash)) + .await + .ok(); + } + #[cfg(feature = "dfu_split")] + SplitMessage::FirmwareChunk { offset, len, data } => { + if self.dfu_handler.is_none() { + self.dfu_handler = crate::dfu::SplitDfuHandler::new(); + if self.dfu_handler.is_none() { + error!("dfu_split: FlashManager not initialized, skipping chunk"); + return; + } + } + let handler = self.dfu_handler.as_mut().unwrap(); + let actual_len = len as usize; + let chunk_data = &data.0[..actual_len]; + match handler.write_chunk(offset as u32, chunk_data) { + Ok(()) => { + debug!("dfu_split: wrote {} bytes at offset {}", actual_len, offset); + let ack = SplitMessage::FirmwareChunkAck { + offset, + crc: crate::crc32::crc32(chunk_data), + }; + self.split_driver.write(&ack).await.ok(); + } + Err(()) => error!("dfu_split: write error at offset {}", offset), + } + } + #[cfg(feature = "dfu_split")] + SplitMessage::FirmwareUpdateComplete => { + if let Some(ref mut handler) = self.dfu_handler { + let dfu_crc = handler.compute_dfu_crc(); + info!("dfu_split: DFU partition CRC: {:#010x}", dfu_crc); + let crc_msg = SplitMessage::FirmwareCrcReport(dfu_crc); + self.split_driver.write(&crc_msg).await.ok(); + info!("dfu_split: CRC report sent"); + + let deadline = embassy_time::Instant::now() + embassy_time::Duration::from_secs(5); + let ok = loop { + match select(self.split_driver.read(), embassy_time::Timer::at(deadline)).await { + Either::First(Ok(SplitMessage::FirmwareCrcOk)) => { + info!("dfu_split: central confirmed CRC, resetting"); + break true; + } + Either::First(Ok(SplitMessage::FirmwareCrcFail)) => { + warn!("dfu_split: central rejected CRC, stopping update"); + break false; + } + Either::First(Ok(_)) => {} + Either::First(Err(e)) => { + error!("read error: {:?}", e); + break false; + } + Either::Second(_) => { + error!("timeout"); + break false; + } + } + }; + + if ok { + self.split_driver.write(&SplitMessage::FirmwareUpdateConfirm).await.ok(); + embassy_time::Timer::after_millis(50).await; + handler.mark_updated_and_reset().ok(); + } else { + self.dfu_handler = None; + } + } else { + error!("dfu_split: no active DFU session"); + } + } + SplitMessage::Application(data) => crate::split_app::deliver_received(data), + _ => (), + } + } } diff --git a/rmk/src/split_app.rs b/rmk/src/split_app.rs new file mode 100644 index 000000000..b69271083 --- /dev/null +++ b/rmk/src/split_app.rs @@ -0,0 +1,131 @@ +//! Bounded application-message hook for the split protocol. +//! +//! An application built on RMK can exchange small opaque payloads +//! ([`SplitAppData`]) between the split halves alongside — but never in front +//! of — normal split traffic. RMK attaches no meaning to the bytes. +//! +//! Each side queues outgoing payloads with `try_send` ([`SPLIT_APP_TX`] on +//! the central, [`SPLIT_APP_PERIPH_TX`] on the peripheral). The split loops +//! drain them as their lowest-priority outgoing arm and deliver received +//! payloads into [`SPLIT_APP_RX`], dropping on full so application traffic +//! can never delay key events. [`SPLIT_APP_LINK`] carries the link state; its +//! `false → true` edge is the application's cue to resync, which also covers +//! messages lost to a full queue. +//! +//! The queues assume a single split peripheral. + +use embassy_sync::channel::Channel; +use embassy_sync::watch::Watch; +use postcard::experimental::max_size::MaxSize; +use serde::{Deserialize, Deserializer, Serialize, Serializer}; + +use crate::RawMutex; + +/// Maximum payload of one application split message. Every split transfer is +/// `SPLIT_MESSAGE_MAX_SIZE` bytes on the wire, and trouble's `gatt_service` +/// macro caps that at 32 (it initializes characteristic arrays via `Default`, +/// which arrays only implement up to 32 elements), so this plus postcard +/// overhead must keep `SPLIT_MESSAGE_MAX_SIZE` ≤ 32. +pub const SPLIT_APP_MSG_MAX: usize = 26; + +/// One opaque application payload. Only `data[..len]` is meaningful. +#[derive(Debug, Clone, Copy)] +#[cfg_attr(feature = "defmt", derive(defmt::Format))] +pub struct SplitAppData { + pub len: u8, + pub data: [u8; SPLIT_APP_MSG_MAX], +} + +impl SplitAppData { + /// Wrap `payload`; `None` if it exceeds [`SPLIT_APP_MSG_MAX`]. + pub fn new(payload: &[u8]) -> Option { + if payload.len() > SPLIT_APP_MSG_MAX { + return None; + } + let mut data = [0u8; SPLIT_APP_MSG_MAX]; + data[..payload.len()].copy_from_slice(payload); + Some(Self { + len: payload.len() as u8, + data, + }) + } + + pub fn payload(&self) -> &[u8] { + &self.data[..(self.len as usize).min(SPLIT_APP_MSG_MAX)] + } +} + +// Postcard stores the payload as `&[u8]` (varint length prefix + bytes), so +// only the used bytes travel inside the (fixed-size) split transfer buffer. +impl Serialize for SplitAppData { + fn serialize(&self, serializer: S) -> Result { + self.payload().serialize(serializer) + } +} + +impl<'de> Deserialize<'de> for SplitAppData { + fn deserialize>(deserializer: D) -> Result { + use serde::de::Error; + let buf: &[u8] = Deserialize::deserialize(deserializer)?; + SplitAppData::new(buf).ok_or_else(|| D::Error::custom("split app message too long")) + } +} + +impl MaxSize for SplitAppData { + // 1-byte varint length prefix + payload. + const POSTCARD_MAX_SIZE: usize = SPLIT_APP_MSG_MAX + 1; +} + +/// Central → peripheral queue. Producers must use `try_send` (never await a +/// full queue); capacity fits one full application resync burst. +pub static SPLIT_APP_TX: Channel = Channel::new(); + +/// This side's inbox of received application messages. Filled with `try_send` +/// (drop-on-full) by the split read loops. +pub static SPLIT_APP_RX: Channel = Channel::new(); + +/// Peripheral → central queue. Producers must use `try_send`. Small: meant +/// for tiny, rare state such as a build identity announced once per link-up. +pub static SPLIT_APP_PERIPH_TX: Channel = Channel::new(); + +/// Deliver a received payload into [`SPLIT_APP_RX`]. Drop-on-full: a slow +/// application consumer must never stall the split read loops, and the +/// application resyncs on reconnect anyway. +pub(crate) fn deliver_received(data: SplitAppData) { + if SPLIT_APP_RX.try_send(data).is_err() { + warn!("split app message dropped (inbox full)"); + } +} + +/// Split-link state for the application, written by the split loops via +/// [`LinkGuard`]. State-based, so a late receiver still observes the current +/// value. +pub static SPLIT_APP_LINK: Watch = Watch::new(); + +/// Owns one split session's [`SPLIT_APP_LINK`] state: `mark_up` sends the +/// link-up edge once, and `Drop` sends the link-down edge. The down edge must +/// come from `Drop` because the session futures holding a guard are cancelled +/// from outside on connection loss. +pub(crate) struct LinkGuard { + up: bool, +} + +impl LinkGuard { + pub(crate) fn new() -> Self { + Self { up: false } + } + + /// Send the link-up edge; idempotent. + pub(crate) fn mark_up(&mut self) { + if !self.up { + SPLIT_APP_LINK.sender().send(true); + self.up = true; + } + } +} + +impl Drop for LinkGuard { + fn drop(&mut self) { + SPLIT_APP_LINK.sender().send(false); + } +}