Skip to content

Add bounded split application messages #984

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
colonelpanic8 wants to merge 3 commits into
base: main
Choose a base branch
from
Open

Add bounded split application messages #984

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
0195a61
split: add a bounded application-message side channel
colonelpanic8 Jul 19, 2026
3bdbd25
fix(split): satisfy collapsible match lint
colonelpanic8 Aug 5, 2026
c4b4cc0
refactor(split): dedupe link guard and flatten peripheral handling
colonelpanic8 Aug 5, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions rmk/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
9 changes: 9 additions & 0 deletions rmk/src/split/driver.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -186,6 +186,12 @@ impl<const ROW: usize, const COL: usize, const ROW_OFFSET: usize, const COL_OFFS
#[cfg(feature = "display")]
let mut sleep_sub = crate::event::SleepStateEvent::subscriber();

// This manager runs exactly while the peripheral session is up. On
// connection loss the `select3` in `split/ble/central.rs` cancels this
// future, so the guard's `Drop` is what sends the link-down edge.
let mut app_link = crate::split_app::LinkGuard::new();
app_link.mark_up();

// Send the current state once on startup so the peripheral matches us
// even when no transition has happened since the central booted.
if self
Expand Down Expand Up @@ -227,6 +233,8 @@ impl<const ROW: usize, const COL: usize, const ROW_OFFSET: usize, const COL_OFFS
with_feature("display"): e = wpm_sub.next_event().fuse() => 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),
}
};

Expand Down Expand Up @@ -285,6 +293,7 @@ impl<const ROW: usize, const COL: usize, const ROW_OFFSET: usize, const COL_OFFS
},
// Non-key events are drop-on-full to keep the split read loop responsive.
SplitMessage::Pointing(e) => 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")]
Expand Down
5 changes: 5 additions & 0 deletions rmk/src/split/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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),
}

// -----------------------------------------------------------------------
Expand Down
252 changes: 137 additions & 115 deletions rmk/src/split/peripheral.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -78,6 +78,15 @@ impl<S: SplitWriter + SplitReader> SplitPeripheral<S> {
/// 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")]
Expand Down Expand Up @@ -108,127 +117,20 @@ impl<S: SplitWriter + SplitReader> SplitPeripheral<S> {
},
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 {
Expand All @@ -243,4 +145,124 @@ impl<S: SplitWriter + SplitReader> SplitPeripheral<S> {
}
}
}

/// 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),
_ => (),
}
}
}
Loading