Virtual gearing, trainer-speed blend, and cadence decode
Gears are expressed as an offset to the commanded gradient, leaving the physics on the route's true gradient so shifting changes effort, not speed. Neutral gear commands exactly the route gradient, so an un-shifted ride is unchanged. Cadence is not in FTMS on this trainer but is on its Zwift channel, decoded against captured frames. The undeclared FTMS trailing bytes were ruled out: wheel RPM restated at a fixed 73.8x speed. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5
@@ -0,0 +1,499 @@
|
||||
//! The Zwift Click client: an async actor that owns a controller peripheral.
|
||||
//!
|
||||
//! Structurally a smaller sibling of [`crate::client`]. One task owns the
|
||||
//! peripheral; callers hold a cheap handle and receive [`ClickEvent`]s on a
|
||||
//! broadcast channel.
|
||||
//!
|
||||
//! ```text
|
||||
//! caller ◄──broadcast── ClickEvent ◄── actor ◄─notify── Click pod
|
||||
//! ```
|
||||
//!
|
||||
//! The protocol itself lives in [`crate::zwift`] as pure functions over bytes;
|
||||
//! this module is only the radio and the reconnect loop. See REQUIREMENTS.md
|
||||
//! §2.3.1 for what was confirmed against the hardware — notably that a Click v2
|
||||
//! needs **no encryption**, so there is no key exchange here to get wrong.
|
||||
//!
|
||||
//! Unlike the trainer, a controller has no safety story: it never commands
|
||||
//! load, and the only bytes ever written to it are the `RideOn` handshake.
|
||||
|
||||
use std::future::Future;
|
||||
use std::time::Duration;
|
||||
|
||||
use btleplug::api::{CharPropFlags, Characteristic, Peripheral as _, WriteType};
|
||||
use btleplug::platform::{Adapter, Peripheral};
|
||||
use futures::{Stream, StreamExt};
|
||||
use tokio::sync::{broadcast, mpsc, oneshot};
|
||||
|
||||
use crate::client::{Backoff, InFlight, DISCONNECT_TIMEOUT};
|
||||
use crate::error::FtmsError;
|
||||
use crate::scan::{self, TrainerSelector};
|
||||
use crate::zwift::{self, Button, ButtonTracker};
|
||||
|
||||
/// Tunables for [`ClickClient`].
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct ClickConfig {
|
||||
/// How long to look for the pod before giving up. A Click sleeps quickly
|
||||
/// and only advertises after a button press (A-4), so this is generous.
|
||||
pub scan_timeout: Duration,
|
||||
pub backoff: Backoff,
|
||||
pub channel_capacity: usize,
|
||||
}
|
||||
|
||||
impl Default for ClickConfig {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
scan_timeout: Duration::from_secs(20),
|
||||
backoff: Backoff::default(),
|
||||
channel_capacity: 64,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Everything the app learns from a controller.
|
||||
#[derive(Debug, Clone, PartialEq)]
|
||||
pub enum ClickEvent {
|
||||
Connected {
|
||||
address: String,
|
||||
name: Option<String>,
|
||||
},
|
||||
/// The link dropped. Any held button has already been reported as released.
|
||||
Disconnected,
|
||||
/// A press or release edge. Repeats while held are filtered out here, not
|
||||
/// by the consumer (see [`ButtonTracker`]).
|
||||
Button { button: Button, pressed: bool },
|
||||
Battery { percent: u8 },
|
||||
/// A frame we could not interpret. Surfaced rather than dropped: the
|
||||
/// protocol is only partly documented, and silence would hide the parts we
|
||||
/// have not met yet.
|
||||
Unknown { kind: u8, raw: Vec<u8> },
|
||||
}
|
||||
|
||||
enum Cmd {
|
||||
Shutdown { reply: oneshot::Sender<()> },
|
||||
}
|
||||
|
||||
/// Cheap, cloneable handle to a controller session.
|
||||
#[derive(Clone)]
|
||||
pub struct ClickClient {
|
||||
cmd_tx: mpsc::Sender<Cmd>,
|
||||
events_tx: broadcast::Sender<ClickEvent>,
|
||||
}
|
||||
|
||||
impl ClickClient {
|
||||
/// Connect to a Click and start streaming events.
|
||||
///
|
||||
/// Returns once the pod has answered the handshake, so a caller that gets
|
||||
/// an `Ok` knows the controller is genuinely talking — not merely that a
|
||||
/// BLE link exists.
|
||||
pub async fn connect(
|
||||
selector: TrainerSelector,
|
||||
config: ClickConfig,
|
||||
) -> Result<Self, FtmsError> {
|
||||
let adapter = scan::default_adapter().await?;
|
||||
Self::connect_with_adapter(adapter, selector, config).await
|
||||
}
|
||||
|
||||
/// As [`ClickClient::connect`], but abandoned as soon as `cancel` resolves.
|
||||
///
|
||||
/// Returns `Ok(None)` when it was cancelled. A Click only advertises after a
|
||||
/// button press (A-4), so a connect routinely runs the full `scan_timeout` —
|
||||
/// twenty seconds, which is long enough that a quit waiting for it reads as
|
||||
/// a hang (FR-1.10). Any link the abandoned attempt had opened is closed
|
||||
/// before this returns (SAF-9).
|
||||
pub async fn connect_cancellable(
|
||||
selector: TrainerSelector,
|
||||
config: ClickConfig,
|
||||
cancel: impl Future<Output = ()>,
|
||||
) -> Result<Option<Self>, FtmsError> {
|
||||
let in_flight = InFlight::default();
|
||||
let outcome = {
|
||||
let attempt = async {
|
||||
let adapter = scan::default_adapter().await?;
|
||||
Self::connect_on(adapter, selector, config, &in_flight).await
|
||||
};
|
||||
tokio::pin!(attempt, cancel);
|
||||
tokio::select! {
|
||||
result = &mut attempt => Some(result),
|
||||
() = &mut cancel => None,
|
||||
}
|
||||
};
|
||||
match outcome {
|
||||
Some(result) => result.map(Some),
|
||||
None => {
|
||||
tracing::info!("click: connect attempt cancelled");
|
||||
in_flight.abandon().await;
|
||||
Ok(None)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// As [`ClickClient::connect`], but on a caller-supplied adapter.
|
||||
pub async fn connect_with_adapter(
|
||||
adapter: Adapter,
|
||||
selector: TrainerSelector,
|
||||
config: ClickConfig,
|
||||
) -> Result<Self, FtmsError> {
|
||||
Self::connect_on(adapter, selector, config, &InFlight::default()).await
|
||||
}
|
||||
|
||||
async fn connect_on(
|
||||
adapter: Adapter,
|
||||
selector: TrainerSelector,
|
||||
config: ClickConfig,
|
||||
in_flight: &InFlight,
|
||||
) -> Result<Self, FtmsError> {
|
||||
let (events_tx, _) = broadcast::channel(config.channel_capacity);
|
||||
let (cmd_tx, cmd_rx) = mpsc::channel(8);
|
||||
|
||||
let (session, notifications) = open_session(&adapter, &selector, &config, in_flight).await?;
|
||||
let _ = events_tx.send(ClickEvent::Connected {
|
||||
address: session.address.clone(),
|
||||
name: session.name.clone(),
|
||||
});
|
||||
|
||||
let actor = Actor {
|
||||
adapter,
|
||||
selector,
|
||||
config,
|
||||
events_tx: events_tx.clone(),
|
||||
tracker: ButtonTracker::new(),
|
||||
};
|
||||
tokio::spawn(actor.run(cmd_rx, session, Box::pin(notifications)));
|
||||
|
||||
Ok(Self { cmd_tx, events_tx })
|
||||
}
|
||||
|
||||
/// Subscribe to controller events. Late subscribers see only what arrives
|
||||
/// after they subscribe.
|
||||
pub fn events(&self) -> broadcast::Receiver<ClickEvent> {
|
||||
self.events_tx.subscribe()
|
||||
}
|
||||
|
||||
/// Close the link, and wait for it to actually be closed.
|
||||
///
|
||||
/// SAF-9: a pod whose link the process merely abandons can stay held by
|
||||
/// BlueZ and be unreachable on the next launch. Fire-and-forget would return
|
||||
/// before the disconnect had even been issued, which at app exit means it
|
||||
/// never is. Idempotent — shutting down a stopped client is a no-op.
|
||||
pub async fn shutdown(&self) {
|
||||
let (reply, done) = oneshot::channel();
|
||||
if self.cmd_tx.send(Cmd::Shutdown { reply }).await.is_err() {
|
||||
// The actor is already gone; it tore the link down on its way out.
|
||||
return;
|
||||
}
|
||||
let _ = done.await;
|
||||
}
|
||||
}
|
||||
|
||||
/// Boxed so that the initial connection and every reconnect share one type —
|
||||
/// `impl Stream` would mint a fresh opaque type per call site and the two could
|
||||
/// not be assigned to the same variable.
|
||||
type Notifications = std::pin::Pin<
|
||||
Box<dyn Stream<Item = btleplug::api::ValueNotification> + Send>,
|
||||
>;
|
||||
|
||||
/// A live connection: the peripheral plus the characteristics we care about.
|
||||
struct Session {
|
||||
peripheral: Peripheral,
|
||||
address: String,
|
||||
name: Option<String>,
|
||||
subscribed: Vec<Characteristic>,
|
||||
}
|
||||
|
||||
/// Find the pod, connect, subscribe, and complete the `RideOn` handshake.
|
||||
async fn open_session(
|
||||
adapter: &Adapter,
|
||||
selector: &TrainerSelector,
|
||||
config: &ClickConfig,
|
||||
in_flight: &InFlight,
|
||||
) -> Result<(Session, Notifications), FtmsError> {
|
||||
let peripheral = scan::find_peripheral(adapter, selector, config.scan_timeout).await?;
|
||||
// Cancelling past this point would otherwise strand the link (FR-1.10).
|
||||
in_flight.hold(peripheral.clone());
|
||||
|
||||
match setup_session(peripheral.clone()).await {
|
||||
Ok(session) => {
|
||||
in_flight.released();
|
||||
Ok(session)
|
||||
}
|
||||
Err(e) => {
|
||||
// Every failure after `connect()` leaves a live GATT link behind,
|
||||
// and a pod that is still held will not advertise for the next
|
||||
// attempt — so the retry loop would never succeed.
|
||||
tracing::debug!(error = %e, "click: session setup failed; disconnecting");
|
||||
let _ = peripheral.disconnect().await;
|
||||
in_flight.released();
|
||||
Err(e)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn setup_session(peripheral: Peripheral) -> Result<(Session, Notifications), FtmsError> {
|
||||
if !peripheral.is_connected().await.unwrap_or(false) {
|
||||
peripheral.connect().await?;
|
||||
}
|
||||
peripheral.discover_services().await?;
|
||||
|
||||
// A Click v2 carries 0xFC82; the trainer carries 00000001-19CA-…. Both hold
|
||||
// the same characteristics, so take whichever is present.
|
||||
let service = zwift::SERVICES
|
||||
.iter()
|
||||
.find_map(|want| peripheral.services().into_iter().find(|s| s.uuid == *want))
|
||||
.ok_or(FtmsError::MissingCharacteristic("Zwift custom service"))?;
|
||||
|
||||
let notifications = peripheral.notifications().await?;
|
||||
|
||||
// Subscribe before writing, so the handshake reply cannot outrun us.
|
||||
let mut subscribed = Vec::new();
|
||||
for ch in &service.characteristics {
|
||||
if ch
|
||||
.properties
|
||||
.intersects(CharPropFlags::NOTIFY | CharPropFlags::INDICATE)
|
||||
{
|
||||
match peripheral.subscribe(ch).await {
|
||||
Ok(()) => subscribed.push(ch.clone()),
|
||||
Err(e) => tracing::debug!(uuid = %ch.uuid, error = %e, "click: subscribe failed"),
|
||||
}
|
||||
}
|
||||
}
|
||||
if subscribed.is_empty() {
|
||||
return Err(FtmsError::MissingCharacteristic(
|
||||
"a notifying Zwift characteristic",
|
||||
));
|
||||
}
|
||||
|
||||
let sync_rx = service
|
||||
.characteristics
|
||||
.iter()
|
||||
.find(|c| c.uuid == zwift::SYNC_RX)
|
||||
.ok_or(FtmsError::MissingCharacteristic("Zwift sync RX"))?;
|
||||
let write_type = if sync_rx
|
||||
.properties
|
||||
.contains(CharPropFlags::WRITE_WITHOUT_RESPONSE)
|
||||
{
|
||||
WriteType::WithoutResponse
|
||||
} else {
|
||||
WriteType::WithResponse
|
||||
};
|
||||
peripheral
|
||||
.write(sync_rx, &zwift::handshake(&zwift::REQUEST_START), write_type)
|
||||
.await?;
|
||||
|
||||
let described = scan::describe(&peripheral).await;
|
||||
Ok((
|
||||
Session {
|
||||
address: described
|
||||
.as_ref()
|
||||
.map(|d| d.address.clone())
|
||||
.unwrap_or_default(),
|
||||
name: described.and_then(|d| d.name),
|
||||
peripheral,
|
||||
subscribed,
|
||||
},
|
||||
Box::pin(notifications),
|
||||
))
|
||||
}
|
||||
|
||||
struct Actor {
|
||||
adapter: Adapter,
|
||||
selector: TrainerSelector,
|
||||
config: ClickConfig,
|
||||
events_tx: broadcast::Sender<ClickEvent>,
|
||||
tracker: ButtonTracker,
|
||||
}
|
||||
|
||||
impl Actor {
|
||||
async fn run(
|
||||
mut self,
|
||||
mut cmd_rx: mpsc::Receiver<Cmd>,
|
||||
mut current: Session,
|
||||
mut notifications: Notifications,
|
||||
) {
|
||||
let mut attempt = 0u32;
|
||||
|
||||
loop {
|
||||
let dropped = tokio::select! {
|
||||
cmd = cmd_rx.recv() => {
|
||||
match cmd {
|
||||
Some(Cmd::Shutdown { reply }) => {
|
||||
self.teardown(¤t).await;
|
||||
// Answered only once the link is genuinely closed,
|
||||
// so a caller blocking the app's exit on this knows
|
||||
// what it waited for (SAF-9).
|
||||
let _ = reply.send(());
|
||||
return;
|
||||
}
|
||||
None => {
|
||||
self.teardown(¤t).await;
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
frame = notifications.next() => match frame {
|
||||
Some(n) => { self.handle(&n.value); false }
|
||||
// The stream ending is how btleplug reports a dropped link.
|
||||
None => true,
|
||||
},
|
||||
};
|
||||
|
||||
if !dropped {
|
||||
continue;
|
||||
}
|
||||
|
||||
// Release anything still held, so a paddle held through a dropout
|
||||
// cannot latch (see ButtonTracker::reset).
|
||||
self.release_held();
|
||||
let _ = self.events_tx.send(ClickEvent::Disconnected);
|
||||
|
||||
// FR-1.6: reconnect with backoff. A Click sleeps aggressively, so
|
||||
// "not found" is the normal case, not an error worth giving up on.
|
||||
loop {
|
||||
if self.config.backoff.exhausted(attempt) {
|
||||
tracing::warn!("click: giving up after {attempt} reconnect attempts");
|
||||
return;
|
||||
}
|
||||
let delay = self.config.backoff.delay(attempt);
|
||||
attempt += 1;
|
||||
|
||||
// Both the backoff *and* the attempt stay answerable to
|
||||
// shutdown. `open_session` runs for as long as `scan_timeout`
|
||||
// — twenty seconds against a pod that only advertises after a
|
||||
// button press — and a quit that waits for that is a quit that
|
||||
// leaves the link open (FR-1.10, SAF-9).
|
||||
let in_flight = InFlight::default();
|
||||
let outcome = {
|
||||
let attempting = async {
|
||||
tokio::time::sleep(delay).await;
|
||||
open_session(&self.adapter, &self.selector, &self.config, &in_flight).await
|
||||
};
|
||||
tokio::pin!(attempting);
|
||||
tokio::select! {
|
||||
biased;
|
||||
cmd = cmd_rx.recv() => Err(cmd),
|
||||
result = &mut attempting => Ok(result),
|
||||
}
|
||||
};
|
||||
|
||||
let result = match outcome {
|
||||
Ok(result) => result,
|
||||
Err(cmd) => {
|
||||
// There is no session to tear down — the attempt owns
|
||||
// whatever link exists, so closing that is the whole job.
|
||||
in_flight.abandon().await;
|
||||
if let Some(Cmd::Shutdown { reply }) = cmd {
|
||||
let _ = reply.send(());
|
||||
}
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
match result {
|
||||
Ok((session, stream)) => {
|
||||
let _ = self.events_tx.send(ClickEvent::Connected {
|
||||
address: session.address.clone(),
|
||||
name: session.name.clone(),
|
||||
});
|
||||
current = session;
|
||||
notifications = stream;
|
||||
attempt = 0;
|
||||
break;
|
||||
}
|
||||
Err(e) => tracing::debug!(error = %e, "click: reconnect failed"),
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Decode one notification into events.
|
||||
fn handle(&mut self, raw: &[u8]) {
|
||||
if zwift::is_ride_on_reply(raw) {
|
||||
tracing::debug!("click: RideOn acknowledged");
|
||||
return;
|
||||
}
|
||||
let Some(frame) = zwift::parse_frame(raw) else {
|
||||
return;
|
||||
};
|
||||
|
||||
match frame.kind {
|
||||
zwift::MessageType::ButtonBitmask => {
|
||||
match zwift::decode_button_bitmask(frame.payload) {
|
||||
Ok(mask) => {
|
||||
for edge in self.tracker.update(mask) {
|
||||
let _ = self.events_tx.send(ClickEvent::Button {
|
||||
button: edge.button,
|
||||
pressed: edge.pressed,
|
||||
});
|
||||
}
|
||||
}
|
||||
Err(e) => tracing::debug!(error = %e, "click: bad button frame"),
|
||||
}
|
||||
}
|
||||
zwift::MessageType::Battery => {
|
||||
if let Ok(Some(percent)) = zwift::decode_battery(frame.payload) {
|
||||
let _ = self.events_tx.send(ClickEvent::Battery { percent });
|
||||
}
|
||||
}
|
||||
zwift::MessageType::KeepAlive => {}
|
||||
zwift::MessageType::Unknown(kind) => {
|
||||
let _ = self.events_tx.send(ClickEvent::Unknown {
|
||||
kind,
|
||||
raw: raw.to_vec(),
|
||||
});
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
|
||||
/// Emit a release for every button the tracker still believes is held.
|
||||
fn release_held(&mut self) {
|
||||
for button in self.tracker.held() {
|
||||
let _ = self.events_tx.send(ClickEvent::Button {
|
||||
button,
|
||||
pressed: false,
|
||||
});
|
||||
}
|
||||
self.tracker.reset();
|
||||
}
|
||||
|
||||
/// Close the link. Every step is bounded: this runs on the app's exit path,
|
||||
/// where a radio operation that never returns is a window that never shuts
|
||||
/// (NFR-9).
|
||||
async fn teardown(&mut self, session: &Session) {
|
||||
self.release_held();
|
||||
for ch in &session.subscribed {
|
||||
let unsubscribe = session.peripheral.unsubscribe(ch);
|
||||
let _ = tokio::time::timeout(DISCONNECT_TIMEOUT, unsubscribe).await;
|
||||
}
|
||||
match tokio::time::timeout(DISCONNECT_TIMEOUT, session.peripheral.disconnect()).await {
|
||||
Ok(Ok(())) => tracing::info!("click: disconnected"),
|
||||
Ok(Err(e)) => tracing::debug!(error = %e, "click: disconnect failed"),
|
||||
Err(_) => tracing::warn!("click: disconnect timed out"),
|
||||
}
|
||||
let _ = self.events_tx.send(ClickEvent::Disconnected);
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn config_defaults_are_patient_enough_for_a_sleeping_pod() {
|
||||
let c = ClickConfig::default();
|
||||
assert!(c.scan_timeout >= Duration::from_secs(10));
|
||||
// Retry forever: a Click that has gone to sleep is the normal case.
|
||||
assert_eq!(c.backoff.max_attempts, None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn events_compare_by_value() {
|
||||
assert_eq!(
|
||||
ClickEvent::Button { button: Button::Plus, pressed: true },
|
||||
ClickEvent::Button { button: Button::Plus, pressed: true }
|
||||
);
|
||||
assert_ne!(
|
||||
ClickEvent::Button { button: Button::Plus, pressed: true },
|
||||
ClickEvent::Button { button: Button::Plus, pressed: false }
|
||||
);
|
||||
}
|
||||
}
|
||||
+239
-13
@@ -17,6 +17,8 @@
|
||||
//! MIT licensed, Copyright (c) 2025 Ogi. See REQUIREMENTS.md §3.2.
|
||||
|
||||
use std::collections::VecDeque;
|
||||
use std::future::Future;
|
||||
use std::sync::Arc;
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use bikecontrol_core::types::{ConnectionState, ControlTarget, SafetyLimits, Telemetry};
|
||||
@@ -232,11 +234,55 @@ impl FtmsClient {
|
||||
Self::connect_with_adapter(adapter, selector, config).await
|
||||
}
|
||||
|
||||
/// As [`FtmsClient::connect`], but abandoned as soon as `cancel` resolves.
|
||||
///
|
||||
/// Returns `Ok(None)` when it was cancelled. A connect runs for up to
|
||||
/// [`FtmsConfig::scan_timeout`] before the trainer has even been found, and
|
||||
/// a caller that needs to stop — the app is quitting, the rider pressed
|
||||
/// disconnect — cannot wait that out (FR-1.10, SAF-8). Any link the
|
||||
/// abandoned attempt had already opened is closed before this returns, so
|
||||
/// the trainer is free for the next launch (A-3).
|
||||
pub async fn connect_cancellable(
|
||||
selector: TrainerSelector,
|
||||
config: FtmsConfig,
|
||||
cancel: impl Future<Output = ()>,
|
||||
) -> Result<Option<Self>, FtmsError> {
|
||||
let in_flight = InFlight::default();
|
||||
let outcome = {
|
||||
let attempt = async {
|
||||
let adapter = scan::default_adapter().await?;
|
||||
Self::connect_on(adapter, selector, config, &in_flight).await
|
||||
};
|
||||
tokio::pin!(attempt, cancel);
|
||||
tokio::select! {
|
||||
result = &mut attempt => Some(result),
|
||||
() = &mut cancel => None,
|
||||
}
|
||||
};
|
||||
match outcome {
|
||||
Some(result) => result.map(Some),
|
||||
None => {
|
||||
tracing::info!("connect attempt cancelled");
|
||||
in_flight.abandon().await;
|
||||
Ok(None)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// As [`FtmsClient::connect`], but on a caller-supplied adapter.
|
||||
pub async fn connect_with_adapter(
|
||||
adapter: Adapter,
|
||||
selector: TrainerSelector,
|
||||
config: FtmsConfig,
|
||||
) -> Result<Self, FtmsError> {
|
||||
Self::connect_on(adapter, selector, config, &InFlight::default()).await
|
||||
}
|
||||
|
||||
async fn connect_on(
|
||||
adapter: Adapter,
|
||||
selector: TrainerSelector,
|
||||
config: FtmsConfig,
|
||||
in_flight: &InFlight,
|
||||
) -> Result<Self, FtmsError> {
|
||||
let (state_tx, state_rx) = watch::channel(ConnectionState::Scanning);
|
||||
let (caps_tx, caps_rx) = watch::channel(TrainerCapabilities::default());
|
||||
@@ -244,7 +290,7 @@ impl FtmsClient {
|
||||
let (events_tx, _) = broadcast::channel(config.channel_capacity);
|
||||
let (cmd_tx, cmd_rx) = mpsc::channel(64);
|
||||
|
||||
let connected = connect_session(&adapter, &selector, &config, &state_tx).await?;
|
||||
let connected = connect_session(&adapter, &selector, &config, &state_tx, in_flight).await?;
|
||||
|
||||
let address = connected.address.clone();
|
||||
let name = connected.name.clone();
|
||||
@@ -385,6 +431,16 @@ impl FtmsClient {
|
||||
/// on application exit: it is the difference between "the trainer is left
|
||||
/// at zero" and "probably".
|
||||
pub async fn shutdown(self) -> Result<(), FtmsError> {
|
||||
self.shutdown_ref().await
|
||||
}
|
||||
|
||||
/// As [`FtmsClient::shutdown`], but by reference.
|
||||
///
|
||||
/// A caller that shares the client — an `Arc<FtmsClient>` behind a
|
||||
/// supervisor, so control writes can be spawned without stalling the
|
||||
/// telemetry pump — cannot consume it, and SAF-2 must not depend on being
|
||||
/// able to. Idempotent: shutting an already-stopped client down is a no-op.
|
||||
pub async fn shutdown_ref(&self) -> Result<(), FtmsError> {
|
||||
let (reply, rx) = oneshot::channel();
|
||||
if self
|
||||
.cmd_tx
|
||||
@@ -419,6 +475,60 @@ enum Command {
|
||||
},
|
||||
}
|
||||
|
||||
/// Upper bound on one write in the SAF-2 reset sequence.
|
||||
const SAFETY_WRITE_TIMEOUT: Duration = Duration::from_millis(750);
|
||||
/// Total budget for the SAF-2 *writes*, leaving room inside NFR-9's eight
|
||||
/// seconds for the disconnect that follows. Individually bounded writes are not
|
||||
/// enough on their own: five of them plus their pacing can still outlast a
|
||||
/// caller holding the app's exit open. The sequence zeroes the active targets
|
||||
/// first (see [`safety_reset_commands`]), so a truncated run has still left the
|
||||
/// trainer at minimum load — which is the part SAF-2 actually cares about.
|
||||
const SAFETY_SEQUENCE_BUDGET: Duration = Duration::from_secs(4);
|
||||
/// Upper bound on closing a link, whether deliberately or after a cancelled
|
||||
/// attempt. A disconnect that hangs must not be the reason the app will not
|
||||
/// quit (NFR-9).
|
||||
pub(crate) const DISCONNECT_TIMEOUT: Duration = Duration::from_secs(2);
|
||||
|
||||
/// The peripheral a connect attempt is working on right now.
|
||||
///
|
||||
/// A cancelled attempt (FR-1.10) may already have opened a GATT link, and a
|
||||
/// cancelled future cannot hand anything back to its caller — hence the shared
|
||||
/// slot. Walking away from that link without closing it matters because the
|
||||
/// D100 accepts one host (A-3): the difference is between "quit and relaunch
|
||||
/// fixed it" and "power-cycle the trainer".
|
||||
#[derive(Clone, Default)]
|
||||
pub(crate) struct InFlight(Arc<std::sync::Mutex<Option<Peripheral>>>);
|
||||
|
||||
impl InFlight {
|
||||
fn lock(&self) -> std::sync::MutexGuard<'_, Option<Peripheral>> {
|
||||
self.0.lock().unwrap_or_else(|e| e.into_inner())
|
||||
}
|
||||
|
||||
/// Take responsibility for a link the attempt is about to open.
|
||||
pub(crate) fn hold(&self, peripheral: Peripheral) {
|
||||
*self.lock() = Some(peripheral);
|
||||
}
|
||||
|
||||
/// The attempt finished and its outcome owns the link now — successfully,
|
||||
/// or by having disconnected on its own error path.
|
||||
pub(crate) fn released(&self) {
|
||||
let _ = self.lock().take();
|
||||
}
|
||||
|
||||
/// Close a link left half-open by a cancelled attempt (SAF-8).
|
||||
pub(crate) async fn abandon(&self) {
|
||||
let Some(peripheral) = self.lock().take() else {
|
||||
return;
|
||||
};
|
||||
tracing::info!("closing the half-open link from a cancelled connect attempt");
|
||||
match tokio::time::timeout(DISCONNECT_TIMEOUT, peripheral.disconnect()).await {
|
||||
Ok(Ok(())) => {}
|
||||
Ok(Err(e)) => tracing::warn!(error = %e, "abandoned link would not close"),
|
||||
Err(_) => tracing::warn!("abandoned link did not close in time"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// The characteristics we need on a connected peripheral.
|
||||
struct Session {
|
||||
peripheral: Peripheral,
|
||||
@@ -884,8 +994,49 @@ impl Actor {
|
||||
|
||||
attempt += 1;
|
||||
self.set_state(ConnectionState::Connecting);
|
||||
match connect_session(&self.adapter, &self.selector, &self.config, &self.state_tx).await
|
||||
{
|
||||
|
||||
// The *attempt* has to be interruptible too, not just the backoff
|
||||
// sleep before it. It runs for up to `scan_timeout` plus connect,
|
||||
// discovery and handshake — far longer than the caller's shutdown
|
||||
// budget — and a shutdown that waits for it to finish is a shutdown
|
||||
// that skips SAF-2 (FR-1.10, SAF-8).
|
||||
let in_flight = InFlight::default();
|
||||
let outcome = {
|
||||
let connect = connect_session(
|
||||
&self.adapter,
|
||||
&self.selector,
|
||||
&self.config,
|
||||
&self.state_tx,
|
||||
&in_flight,
|
||||
);
|
||||
tokio::pin!(connect);
|
||||
loop {
|
||||
tokio::select! {
|
||||
biased;
|
||||
cmd = cmd_rx.recv() => match cmd {
|
||||
Some(Command::Shutdown { reply }) => break Attempt::Abandoned(Some(reply)),
|
||||
Some(Command::SetTarget { reply, .. }) => {
|
||||
let _ = reply.send(Err(FtmsError::NotConnected));
|
||||
}
|
||||
Some(Command::Procedure { reply, .. }) => {
|
||||
let _ = reply.send(Err(FtmsError::NotConnected));
|
||||
}
|
||||
None => break Attempt::Abandoned(None),
|
||||
},
|
||||
result = &mut connect => break Attempt::Finished(Box::new(result)),
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
let result = match outcome {
|
||||
Attempt::Finished(result) => *result,
|
||||
Attempt::Abandoned(reply) => {
|
||||
in_flight.abandon().await;
|
||||
return Reconnected::Shutdown(reply);
|
||||
}
|
||||
};
|
||||
|
||||
match result {
|
||||
Ok(connected) => {
|
||||
self.session = Some(connected.session);
|
||||
self.capabilities = connected.capabilities;
|
||||
@@ -918,23 +1069,36 @@ impl Actor {
|
||||
}
|
||||
|
||||
let Some(session) = self.session.take() else {
|
||||
self.set_state(ConnectionState::Idle);
|
||||
// FR-1.11: a give-up has already published `Lost` with the reason
|
||||
// it gave up for. `Idle` here would erase that and read as "never
|
||||
// connected", which is the one thing the rider must not be told
|
||||
// after the app spent minutes trying to get back.
|
||||
if !matches!(*self.state_tx.borrow(), ConnectionState::Lost { .. }) {
|
||||
self.set_state(ConnectionState::Idle);
|
||||
}
|
||||
return;
|
||||
};
|
||||
|
||||
tracing::info!("running the SAF-2 shutdown sequence");
|
||||
let deadline = Instant::now() + SAFETY_SEQUENCE_BUDGET;
|
||||
for (op, bytes) in safety_reset_commands(
|
||||
&self.config.limits,
|
||||
&self.capabilities,
|
||||
self.config.use_simulation_mode,
|
||||
self.config.simulation_template,
|
||||
) {
|
||||
if Instant::now() >= deadline {
|
||||
// The zeroing writes go first, so what is being skipped here is
|
||||
// `Reset`/`Stop` on a trainer that is already at minimum load.
|
||||
tracing::warn!(%op, "SAF-2 budget spent; skipping the rest of the reset sequence");
|
||||
break;
|
||||
}
|
||||
let write = session.peripheral.write(
|
||||
&session.control_point,
|
||||
&bytes,
|
||||
WriteType::WithResponse,
|
||||
);
|
||||
match tokio::time::timeout(Duration::from_millis(750), write).await {
|
||||
match tokio::time::timeout(SAFETY_WRITE_TIMEOUT, write).await {
|
||||
Ok(Ok(())) => tracing::debug!(%op, raw = %hex(&bytes), "shutdown write"),
|
||||
Ok(Err(e)) => tracing::warn!(%op, error = %e, "shutdown write failed"),
|
||||
Err(_) => tracing::warn!(%op, "shutdown write timed out"),
|
||||
@@ -943,7 +1107,7 @@ impl Actor {
|
||||
tokio::time::sleep(Duration::from_millis(60)).await;
|
||||
}
|
||||
|
||||
match tokio::time::timeout(Duration::from_secs(2), session.peripheral.disconnect()).await {
|
||||
match tokio::time::timeout(DISCONNECT_TIMEOUT, session.peripheral.disconnect()).await {
|
||||
Ok(Ok(())) => tracing::info!("disconnected from the trainer"),
|
||||
Ok(Err(e)) => tracing::warn!(error = %e, "disconnect failed"),
|
||||
Err(_) => tracing::warn!("disconnect timed out"),
|
||||
@@ -981,6 +1145,15 @@ enum Reconnected {
|
||||
GaveUp,
|
||||
}
|
||||
|
||||
/// How one reconnect attempt ended: on its own, or because the caller stopped
|
||||
/// waiting for it (FR-1.10).
|
||||
enum Attempt {
|
||||
/// Boxed only to keep the two variants a similar size; a `ConnectedTrainer`
|
||||
/// carries the whole session.
|
||||
Finished(Box<Result<ConnectedTrainer, FtmsError>>),
|
||||
Abandoned(Option<oneshot::Sender<()>>),
|
||||
}
|
||||
|
||||
fn complete(reply: Reply, result: Result<(), FtmsError>) {
|
||||
match reply {
|
||||
Reply::Target { sent, tx } => {
|
||||
@@ -1071,6 +1244,7 @@ async fn connect_session(
|
||||
selector: &TrainerSelector,
|
||||
config: &FtmsConfig,
|
||||
state_tx: &watch::Sender<ConnectionState>,
|
||||
in_flight: &InFlight,
|
||||
) -> Result<ConnectedTrainer, FtmsError> {
|
||||
let _ = state_tx.send_if_modified(|s| {
|
||||
if *s == ConnectionState::Scanning {
|
||||
@@ -1082,6 +1256,11 @@ async fn connect_session(
|
||||
});
|
||||
|
||||
let peripheral = scan::find_peripheral(adapter, selector, config.scan_timeout).await?;
|
||||
// From here until this function returns, cancelling the caller is the only
|
||||
// thing that can leave a link open with nobody to close it. Hand the
|
||||
// peripheral over now, before `connect()` — a cancellation lands wherever it
|
||||
// lands, including halfway through the connect itself (FR-1.10).
|
||||
in_flight.hold(peripheral.clone());
|
||||
let described = scan::describe(&peripheral).await;
|
||||
let address = described
|
||||
.as_ref()
|
||||
@@ -1099,16 +1278,20 @@ async fn connect_session(
|
||||
// that accepts one connection at a time (A-3) would stay unavailable to the
|
||||
// next attempt. Set it up separately so every error path disconnects.
|
||||
match setup_session(peripheral.clone(), config, state_tx).await {
|
||||
Ok((session, capabilities, notifications)) => Ok(ConnectedTrainer {
|
||||
session,
|
||||
capabilities,
|
||||
notifications,
|
||||
address,
|
||||
name,
|
||||
}),
|
||||
Ok((session, capabilities, notifications)) => {
|
||||
in_flight.released();
|
||||
Ok(ConnectedTrainer {
|
||||
session,
|
||||
capabilities,
|
||||
notifications,
|
||||
address,
|
||||
name,
|
||||
})
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::warn!(error = %e, "connection setup failed; disconnecting");
|
||||
let _ = peripheral.disconnect().await;
|
||||
in_flight.released();
|
||||
Err(e)
|
||||
}
|
||||
}
|
||||
@@ -1604,6 +1787,49 @@ mod tests {
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn the_shutdown_sequence_cannot_outlast_the_apps_exit() {
|
||||
// SAF-8 / NFR-9. Per-write timeouts alone are not a bound: five of them
|
||||
// plus their pacing and the disconnect can still outlast the caller
|
||||
// holding the window open, and a caller that gives up waiting leaves the
|
||||
// trainer loaded. Hence a deadline across the whole sequence.
|
||||
assert!(SAFETY_WRITE_TIMEOUT < SAFETY_SEQUENCE_BUDGET);
|
||||
assert!(SAFETY_SEQUENCE_BUDGET + DISCONNECT_TIMEOUT < Duration::from_secs(8));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_truncated_shutdown_sequence_has_still_zeroed_the_load() {
|
||||
// What the budget above may cut short is the tail. That is only
|
||||
// acceptable because the zeroing writes come first — the trainer is at
|
||||
// minimum load before `Reset` and `Stop` are even attempted, which is
|
||||
// the part SAF-2 exists for.
|
||||
for use_sim in [false, true] {
|
||||
let cmds = safety_reset_commands(
|
||||
&SafetyLimits::default(),
|
||||
&TrainerCapabilities::default(),
|
||||
use_sim,
|
||||
SimulationParameters::default(),
|
||||
);
|
||||
let ops: Vec<OpCode> = cmds.iter().map(|(op, _)| *op).collect();
|
||||
let first_zeroing = ops
|
||||
.iter()
|
||||
.position(|op| {
|
||||
matches!(
|
||||
op,
|
||||
OpCode::SetTargetInclination
|
||||
| OpCode::SetTargetResistanceLevel
|
||||
| OpCode::SetIndoorBikeSimulationParameters
|
||||
)
|
||||
})
|
||||
.expect("the sequence must zero something");
|
||||
let reset = ops
|
||||
.iter()
|
||||
.position(|op| *op == OpCode::Reset)
|
||||
.expect("the sequence must reset");
|
||||
assert!(first_zeroing < reset, "{ops:?}");
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn safety_reset_in_simulation_mode_sends_zero_grade_and_zero_wind() {
|
||||
let caps = TrainerCapabilities {
|
||||
|
||||
@@ -12,6 +12,7 @@
|
||||
//! | [`indoor_bike_data`] | `0x2AD2` decoder | no |
|
||||
//! | [`control_point`] | `0x2AD9` encoders and response decoding | no |
|
||||
//! | [`capabilities`] | `0x2ACC`/`0x2AD5`/`0x2AD6`/`0x2AD8` decoding, and the safety gate | no |
|
||||
//! | [`zwift`] | Zwift's proprietary protocol (§2.3.1) | no |
|
||||
//! | [`scan`] | discovery | yes |
|
||||
//! | [`client`] | the connection actor | yes |
|
||||
//!
|
||||
@@ -47,13 +48,16 @@
|
||||
//! (REQUIREMENTS.md §3.2). Per-module attribution notes mark where.
|
||||
|
||||
pub mod capabilities;
|
||||
pub mod click;
|
||||
pub mod client;
|
||||
pub mod control_point;
|
||||
pub mod error;
|
||||
pub mod indoor_bike_data;
|
||||
pub mod scan;
|
||||
pub mod uuids;
|
||||
pub mod zwift;
|
||||
|
||||
pub use click::{ClickClient, ClickConfig, ClickEvent};
|
||||
pub use capabilities::{
|
||||
FitnessMachineFeature, InclinationRange, PowerRange, ResistanceLevelRange, TrainerCapabilities,
|
||||
UnsupportedTarget,
|
||||
@@ -68,3 +72,7 @@ pub use scan::{
|
||||
default_adapter, scan, scan_trainers, DiscoveredDevice, ScanKind, TrainerSelector,
|
||||
};
|
||||
pub use uuids::FITNESS_MACHINE_SERVICE;
|
||||
pub use zwift::{
|
||||
Button, ButtonBitmask, ClickButtons, DeviceKind as ZwiftDeviceKind,
|
||||
MessageType as ZwiftMessageType,
|
||||
};
|
||||
|
||||
+16
-10
@@ -8,20 +8,14 @@ use btleplug::platform::{Adapter, Manager, Peripheral, PeripheralId};
|
||||
use uuid::Uuid;
|
||||
|
||||
use crate::error::FtmsError;
|
||||
use crate::uuids;
|
||||
use crate::{uuids, zwift};
|
||||
|
||||
/// Zwift's custom service UUID, used to recognise Click pods during a scan
|
||||
/// (FR-1.2). The Click *client* is Phase 3 and lives elsewhere; discovery only
|
||||
/// needs the UUID so `probe scan` can label them.
|
||||
pub const ZWIFT_SERVICE: Uuid = Uuid::from_fields(
|
||||
0x0000_0001,
|
||||
0x19CA,
|
||||
0x4651,
|
||||
&[0x86, 0xE5, 0xFA, 0x29, 0xDC, 0xDD, 0x09, 0xD1],
|
||||
);
|
||||
/// (FR-1.2). Re-exported from [`crate::zwift`], which owns the protocol.
|
||||
pub use crate::zwift::SERVICE as ZWIFT_SERVICE;
|
||||
|
||||
/// Zwift's Bluetooth SIG manufacturer ID (2378).
|
||||
pub const ZWIFT_MANUFACTURER_ID: u16 = 0x094A;
|
||||
pub use crate::zwift::MANUFACTURER_ID as ZWIFT_MANUFACTURER_ID;
|
||||
|
||||
/// A peripheral seen during a scan.
|
||||
#[derive(Debug, Clone)]
|
||||
@@ -48,11 +42,23 @@ impl DiscoveredDevice {
|
||||
}
|
||||
|
||||
/// True when the peripheral looks like a Zwift controller.
|
||||
///
|
||||
/// Note that the Van Rysel D100 also advertises Zwift's custom service, so
|
||||
/// this is "speaks the Zwift protocol", not "is a Click".
|
||||
pub fn is_zwift_device(&self) -> bool {
|
||||
self.services.contains(&ZWIFT_SERVICE)
|
||||
|| self.manufacturer_data.contains_key(&ZWIFT_MANUFACTURER_ID)
|
||||
}
|
||||
|
||||
/// Which Zwift device this is, from the type byte in its manufacturer data
|
||||
/// (§2.3.1). `None` when it advertises no Zwift manufacturer data at all —
|
||||
/// which is the case for a trainer that merely exposes the service.
|
||||
pub fn zwift_kind(&self) -> Option<zwift::DeviceKind> {
|
||||
self.manufacturer_data
|
||||
.get(&ZWIFT_MANUFACTURER_ID)
|
||||
.and_then(|d| zwift::DeviceKind::from_manufacturer_data(d))
|
||||
}
|
||||
|
||||
/// Best-effort human label.
|
||||
pub fn label(&self) -> String {
|
||||
match &self.name {
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user