Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
34153cca9d | ||
|
|
d4c219837b | ||
|
|
1c721b68f5 |
@@ -89,6 +89,20 @@ joining another user's trusted-device group.
|
|||||||
Keeping these layers separate prevents discovery convenience from silently
|
Keeping these layers separate prevents discovery convenience from silently
|
||||||
becoming a synchronization trust decision.
|
becoming a synchronization trust decision.
|
||||||
|
|
||||||
|
### Federation Jam control
|
||||||
|
|
||||||
|
Jam is a third, deliberately narrow authority boundary. A host creates an
|
||||||
|
opaque `frid://j/...` runtime capability and remains the only node producing
|
||||||
|
audio. Other TUI peers use a dedicated Jam ALPN to submit the same portable
|
||||||
|
playback commands used by connected-device control and receive the host's
|
||||||
|
playback snapshot. They receive queue metadata, not audio.
|
||||||
|
|
||||||
|
Jam never exchanges trusted membership, likes, playlists, or listening
|
||||||
|
history. Volume remains local. Commands carry unique IDs and are retried until
|
||||||
|
the host acknowledges them, while inactive participants expire from the
|
||||||
|
runtime session. Regenerating the capability or restarting the host invalidates
|
||||||
|
the previous link.
|
||||||
|
|
||||||
## Discovery and direct communication
|
## Discovery and direct communication
|
||||||
|
|
||||||
Furumi separates finding content from transferring it.
|
Furumi separates finding content from transferring it.
|
||||||
|
|||||||
Generated
+1
-1
@@ -1566,7 +1566,7 @@ dependencies = [
|
|||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "furumi_tui"
|
name = "furumi_tui"
|
||||||
version = "0.2.1"
|
version = "0.2.2"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"anyhow",
|
"anyhow",
|
||||||
"blake3",
|
"blake3",
|
||||||
|
|||||||
+1
-1
@@ -1,6 +1,6 @@
|
|||||||
[package]
|
[package]
|
||||||
name = "furumi_tui"
|
name = "furumi_tui"
|
||||||
version = "0.2.1"
|
version = "0.2.2"
|
||||||
edition = "2024"
|
edition = "2024"
|
||||||
rust-version = "1.97"
|
rust-version = "1.97"
|
||||||
description = "A federated P2P player for personal music libraries"
|
description = "A federated P2P player for personal music libraries"
|
||||||
|
|||||||
@@ -166,4 +166,14 @@ pub enum AppEvent {
|
|||||||
DevicePlayback(crate::devices::PlaybackSnapshot),
|
DevicePlayback(crate::devices::PlaybackSnapshot),
|
||||||
/// Playback command addressed to this device.
|
/// Playback command addressed to this device.
|
||||||
PlaybackCommand(crate::devices::PlaybackCommand),
|
PlaybackCommand(crate::devices::PlaybackCommand),
|
||||||
|
/// Current lifecycle/status of the federation Jam.
|
||||||
|
JamStatus(crate::jam::JamStatus),
|
||||||
|
/// Host playback snapshot received by a Jam participant.
|
||||||
|
JamPlayback(crate::devices::PlaybackSnapshot),
|
||||||
|
/// Participant command accepted by this Jam host.
|
||||||
|
JamCommand(crate::devices::PlaybackCommand),
|
||||||
|
/// Result of creating/regenerating a Jam capability.
|
||||||
|
JamInvite(Result<String, String>),
|
||||||
|
/// Result of joining a Jam capability.
|
||||||
|
JamJoined(Result<String, String>),
|
||||||
}
|
}
|
||||||
|
|||||||
+163
-12
@@ -38,6 +38,7 @@ pub struct Runtime {
|
|||||||
pub event_tx: mpsc::UnboundedSender<AppEvent>,
|
pub event_tx: mpsc::UnboundedSender<AppEvent>,
|
||||||
pub library: Arc<Library>,
|
pub library: Arc<Library>,
|
||||||
pub devices: Arc<crate::devices::DeviceSync>,
|
pub devices: Arc<crate::devices::DeviceSync>,
|
||||||
|
pub jam: Arc<crate::jam::JamManager>,
|
||||||
pub federation: Arc<crate::federation::Federation>,
|
pub federation: Arc<crate::federation::Federation>,
|
||||||
/// When the last Federation-tab status snapshot was requested.
|
/// When the last Federation-tab status snapshot was requested.
|
||||||
pub fed_status_at: Option<std::time::Instant>,
|
pub fed_status_at: Option<std::time::Instant>,
|
||||||
@@ -249,7 +250,12 @@ pub async fn run(
|
|||||||
|
|
||||||
let devices = crate::devices::DeviceSync::new(Arc::clone(&library))?;
|
let devices = crate::devices::DeviceSync::new(Arc::clone(&library))?;
|
||||||
devices.set_event_tx(event_tx.clone());
|
devices.set_event_tx(event_tx.clone());
|
||||||
let federation = crate::federation::Federation::new(Arc::clone(&library), Arc::clone(&devices));
|
let jam = crate::jam::JamManager::new(event_tx.clone());
|
||||||
|
let federation = crate::federation::Federation::new(
|
||||||
|
Arc::clone(&library),
|
||||||
|
Arc::clone(&devices),
|
||||||
|
Arc::clone(&jam),
|
||||||
|
);
|
||||||
state.federation.settings = federation.settings();
|
state.federation.settings = federation.settings();
|
||||||
state.federation.devices = Some(devices.status());
|
state.federation.devices = Some(devices.status());
|
||||||
if let Ok((device_id, device_name)) = devices.identity_summary() {
|
if let Ok((device_id, device_name)) = devices.identity_summary() {
|
||||||
@@ -263,6 +269,7 @@ pub async fn run(
|
|||||||
event_tx,
|
event_tx,
|
||||||
library,
|
library,
|
||||||
devices,
|
devices,
|
||||||
|
jam,
|
||||||
federation,
|
federation,
|
||||||
fed_status_at: None,
|
fed_status_at: None,
|
||||||
library_network_refresh_at: None,
|
library_network_refresh_at: None,
|
||||||
@@ -360,7 +367,7 @@ fn sync_player_shared(state: &mut AppState, runtime: &Runtime) {
|
|||||||
} else {
|
} else {
|
||||||
runtime.player.shared.audio_analysis()
|
runtime.player.shared.audio_analysis()
|
||||||
};
|
};
|
||||||
if state.device_playback.role == state::DevicePlaybackRole::Active {
|
if state.device_playback.is_audio_owner() {
|
||||||
publish_playback_snapshot(state, runtime);
|
publish_playback_snapshot(state, runtime);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -474,7 +481,7 @@ fn apply_playback_state_to_ui(
|
|||||||
}
|
}
|
||||||
|
|
||||||
fn publish_playback_snapshot(state: &mut AppState, runtime: &Runtime) {
|
fn publish_playback_snapshot(state: &mut AppState, runtime: &Runtime) {
|
||||||
if state.device_playback.role != state::DevicePlaybackRole::Active {
|
if !state.device_playback.is_audio_owner() {
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
publish_playback_snapshot_with_active(state, runtime, true);
|
publish_playback_snapshot_with_active(state, runtime, true);
|
||||||
@@ -503,6 +510,19 @@ fn publish_playback_snapshot_with_active(state: &mut AppState, runtime: &Runtime
|
|||||||
state: playback_state_from_ui(state),
|
state: playback_state_from_ui(state),
|
||||||
};
|
};
|
||||||
runtime.devices.publish_playback(snapshot);
|
runtime.devices.publish_playback(snapshot);
|
||||||
|
if state.device_playback.role == state::DevicePlaybackRole::Jam
|
||||||
|
&& state.device_playback.jam_host
|
||||||
|
{
|
||||||
|
runtime
|
||||||
|
.jam
|
||||||
|
.publish_host_playback(crate::devices::PlaybackSnapshot {
|
||||||
|
device_id: state.device_playback.self_device_id.clone(),
|
||||||
|
device_name: state.device_playback.self_device_name.clone(),
|
||||||
|
active: true,
|
||||||
|
updated_at_ms: unix_time_ms(),
|
||||||
|
state: playback_state_from_ui(state),
|
||||||
|
});
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
fn update_local_idle_since(state: &mut AppState) {
|
fn update_local_idle_since(state: &mut AppState) {
|
||||||
@@ -532,7 +552,7 @@ fn active_idle_lease_expired(snapshot: &crate::devices::PlaybackSnapshot, now: i
|
|||||||
}
|
}
|
||||||
|
|
||||||
fn local_active_lease_protected(state: &mut AppState, now: i64) -> bool {
|
fn local_active_lease_protected(state: &mut AppState, now: i64) -> bool {
|
||||||
if state.device_playback.role != state::DevicePlaybackRole::Active || !state.player.playing {
|
if !state.device_playback.is_audio_owner() || !state.player.playing {
|
||||||
return false;
|
return false;
|
||||||
}
|
}
|
||||||
if !state.player.paused {
|
if !state.player.paused {
|
||||||
@@ -576,7 +596,7 @@ pub(crate) fn become_control_device(
|
|||||||
runtime: &Runtime,
|
runtime: &Runtime,
|
||||||
snapshot: crate::devices::PlaybackSnapshot,
|
snapshot: crate::devices::PlaybackSnapshot,
|
||||||
) {
|
) {
|
||||||
if state.device_playback.role == state::DevicePlaybackRole::Active {
|
if state.device_playback.is_audio_owner() {
|
||||||
runtime.player.stop();
|
runtime.player.stop();
|
||||||
publish_inactive_playback_snapshot(state, runtime);
|
publish_inactive_playback_snapshot(state, runtime);
|
||||||
}
|
}
|
||||||
@@ -596,6 +616,7 @@ pub(crate) fn become_control_device(
|
|||||||
pub(crate) fn become_active_device(state: &mut AppState, runtime: &mut Runtime, start_audio: bool) {
|
pub(crate) fn become_active_device(state: &mut AppState, runtime: &mut Runtime, start_audio: bool) {
|
||||||
let was_control = state.device_playback.role == state::DevicePlaybackRole::Control;
|
let was_control = state.device_playback.role == state::DevicePlaybackRole::Control;
|
||||||
state.device_playback.role = state::DevicePlaybackRole::Active;
|
state.device_playback.role = state::DevicePlaybackRole::Active;
|
||||||
|
state.device_playback.jam_host = false;
|
||||||
let Ok((device_id, device_name)) = runtime.devices.identity_summary() else {
|
let Ok((device_id, device_name)) = runtime.devices.identity_summary() else {
|
||||||
return;
|
return;
|
||||||
};
|
};
|
||||||
@@ -617,7 +638,7 @@ pub(crate) fn become_active_device(state: &mut AppState, runtime: &mut Runtime,
|
|||||||
}
|
}
|
||||||
|
|
||||||
pub(crate) fn transfer_active_to_this_device(state: &mut AppState, runtime: &mut Runtime) {
|
pub(crate) fn transfer_active_to_this_device(state: &mut AppState, runtime: &mut Runtime) {
|
||||||
if state.device_playback.role == state::DevicePlaybackRole::Active {
|
if state.device_playback.is_audio_owner() {
|
||||||
publish_playback_snapshot(state, runtime);
|
publish_playback_snapshot(state, runtime);
|
||||||
request_urgent_device_sync(runtime);
|
request_urgent_device_sync(runtime);
|
||||||
return;
|
return;
|
||||||
@@ -722,6 +743,12 @@ fn record_control_playback_state(state: &mut AppState, runtime: &Runtime, seek:
|
|||||||
state: playback_state_from_ui(state),
|
state: playback_state_from_ui(state),
|
||||||
seek,
|
seek,
|
||||||
};
|
};
|
||||||
|
if state.device_playback.role == state::DevicePlaybackRole::Jam {
|
||||||
|
if let Err(err) = runtime.jam.submit_command(command) {
|
||||||
|
state.status_message = Some(format!("Jam command failed: {err:#}"));
|
||||||
|
}
|
||||||
|
return;
|
||||||
|
}
|
||||||
record_playback_command_async(runtime, target, command, "device command");
|
record_playback_command_async(runtime, target, command, "device command");
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -1479,10 +1506,41 @@ fn perform_effect(state: &mut AppState, runtime: &mut Runtime, effect: Effect) {
|
|||||||
| Effect::DeviceSetName(_)
|
| Effect::DeviceSetName(_)
|
||||||
| Effect::DeviceRevoke(_)
|
| Effect::DeviceRevoke(_)
|
||||||
| Effect::DeviceLeaveGroup
|
| Effect::DeviceLeaveGroup
|
||||||
|
| Effect::JamCreate
|
||||||
|
| Effect::JamJoin(_)
|
||||||
if !state.connected_devices_enabled() =>
|
if !state.connected_devices_enabled() =>
|
||||||
{
|
{
|
||||||
state.status_message = Some("enable federation before using connected devices".into());
|
state.status_message = Some("enable federation before using connected devices".into());
|
||||||
}
|
}
|
||||||
|
Effect::JamCreate => {
|
||||||
|
let federation = Arc::clone(&runtime.federation);
|
||||||
|
let tx = runtime.event_tx.clone();
|
||||||
|
tokio::spawn(async move {
|
||||||
|
let result = federation
|
||||||
|
.create_jam()
|
||||||
|
.await
|
||||||
|
.map_err(|err| format!("{err:#}"));
|
||||||
|
let _ = tx.send(AppEvent::JamInvite(result));
|
||||||
|
});
|
||||||
|
}
|
||||||
|
Effect::JamJoin(invite) => {
|
||||||
|
let result = runtime
|
||||||
|
.federation
|
||||||
|
.join_jam(&invite)
|
||||||
|
.map(|()| "joined Jam; waiting for host state".to_string())
|
||||||
|
.map_err(|err| format!("{err:#}"));
|
||||||
|
let _ = runtime.event_tx.send(AppEvent::JamJoined(result));
|
||||||
|
let _ = runtime
|
||||||
|
.event_tx
|
||||||
|
.send(AppEvent::JamStatus(runtime.jam.status()));
|
||||||
|
}
|
||||||
|
Effect::JamLeave => {
|
||||||
|
runtime.jam.leave();
|
||||||
|
state.jam = runtime.jam.status();
|
||||||
|
state.device_playback.role = state::DevicePlaybackRole::Active;
|
||||||
|
state.device_playback.jam_host = false;
|
||||||
|
state.status_message = Some("left Jam".into());
|
||||||
|
}
|
||||||
Effect::DeviceShowInvite => {
|
Effect::DeviceShowInvite => {
|
||||||
let fed = Arc::clone(&runtime.federation);
|
let fed = Arc::clone(&runtime.federation);
|
||||||
let devices = Arc::clone(&runtime.devices);
|
let devices = Arc::clone(&runtime.devices);
|
||||||
@@ -1679,6 +1737,8 @@ fn is_controlled_playback_effect(effect: &Effect) -> bool {
|
|||||||
|
|
||||||
fn perform_control_playback_effect(state: &mut AppState, runtime: &mut Runtime, effect: Effect) {
|
fn perform_control_playback_effect(state: &mut AppState, runtime: &mut Runtime, effect: Effect) {
|
||||||
let mut seek = false;
|
let mut seek = false;
|
||||||
|
let local_only_volume = state.device_playback.role == state::DevicePlaybackRole::Jam
|
||||||
|
&& matches!(effect, Effect::SetVolume(_));
|
||||||
match effect {
|
match effect {
|
||||||
Effect::PlayCurrent => {
|
Effect::PlayCurrent => {
|
||||||
state.player.current = state.player.queue.get(state.player.queue_pos).cloned();
|
state.player.current = state.player.queue.get(state.player.queue_pos).cloned();
|
||||||
@@ -1708,6 +1768,9 @@ fn perform_control_playback_effect(state: &mut AppState, runtime: &mut Runtime,
|
|||||||
| Effect::LoadListenHistory => {}
|
| Effect::LoadListenHistory => {}
|
||||||
_ => {}
|
_ => {}
|
||||||
}
|
}
|
||||||
|
if local_only_volume {
|
||||||
|
return;
|
||||||
|
}
|
||||||
record_control_playback_state(state, runtime, seek);
|
record_control_playback_state(state, runtime, seek);
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -2721,6 +2784,12 @@ fn handle_device_playback_snapshot(
|
|||||||
runtime: &mut Runtime,
|
runtime: &mut Runtime,
|
||||||
snapshot: crate::devices::PlaybackSnapshot,
|
snapshot: crate::devices::PlaybackSnapshot,
|
||||||
) {
|
) {
|
||||||
|
// Personal-device reconciliation must never change Jam ownership. Jam
|
||||||
|
// has its own authority and lifecycle even when the same TUI also belongs
|
||||||
|
// to a trusted-device group.
|
||||||
|
if state.device_playback.role == state::DevicePlaybackRole::Jam {
|
||||||
|
return;
|
||||||
|
}
|
||||||
if snapshot.device_id == state.device_playback.self_device_id {
|
if snapshot.device_id == state.device_playback.self_device_id {
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
@@ -2750,10 +2819,10 @@ fn handle_device_playback_snapshot(
|
|||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
let lease_expired = active_idle_lease_expired(&snapshot, now);
|
let lease_expired = active_idle_lease_expired(&snapshot, now);
|
||||||
let already_controls_this_device = state.device_playback.is_control()
|
let already_controls_this_device = state.device_playback.is_personal_control()
|
||||||
&& state.device_playback.active_device_id.as_deref() == Some(snapshot.device_id.as_str());
|
&& state.device_playback.active_device_id.as_deref() == Some(snapshot.device_id.as_str());
|
||||||
if !lease_expired || already_controls_this_device {
|
if !lease_expired || already_controls_this_device {
|
||||||
let was_active = state.device_playback.role == state::DevicePlaybackRole::Active;
|
let was_active = state.device_playback.is_audio_owner();
|
||||||
let was_paused = state.player.playing && state.player.paused;
|
let was_paused = state.player.playing && state.player.paused;
|
||||||
become_control_device(state, runtime, snapshot.clone());
|
become_control_device(state, runtime, snapshot.clone());
|
||||||
if was_active && was_paused {
|
if was_active && was_paused {
|
||||||
@@ -2765,7 +2834,7 @@ fn handle_device_playback_snapshot(
|
|||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
if state.device_playback.is_control() {
|
if state.device_playback.is_personal_control() {
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
become_active_device(state, runtime, false);
|
become_active_device(state, runtime, false);
|
||||||
@@ -2829,7 +2898,7 @@ fn handle_playback_command(
|
|||||||
if active_device_id == state.device_playback.self_device_id {
|
if active_device_id == state.device_playback.self_device_id {
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
let was_active = state.device_playback.role == state::DevicePlaybackRole::Active;
|
let was_active = state.device_playback.is_audio_owner();
|
||||||
let snapshot = crate::devices::PlaybackSnapshot {
|
let snapshot = crate::devices::PlaybackSnapshot {
|
||||||
device_id: active_device_id,
|
device_id: active_device_id,
|
||||||
device_name: active_device_name,
|
device_name: active_device_name,
|
||||||
@@ -2879,7 +2948,7 @@ fn handle_app_event(state: &mut AppState, runtime: &mut Runtime, event: AppEvent
|
|||||||
})
|
})
|
||||||
.unwrap_or(1)
|
.unwrap_or(1)
|
||||||
.max(1);
|
.max(1);
|
||||||
let active_revoked = state.device_playback.is_control()
|
let active_revoked = state.device_playback.is_personal_control()
|
||||||
&& state
|
&& state
|
||||||
.device_playback
|
.device_playback
|
||||||
.active_device_id
|
.active_device_id
|
||||||
@@ -2897,7 +2966,7 @@ fn handle_app_event(state: &mut AppState, runtime: &mut Runtime, event: AppEvent
|
|||||||
})
|
})
|
||||||
.is_some_and(|device| device.revoked)
|
.is_some_and(|device| device.revoked)
|
||||||
});
|
});
|
||||||
let active_missing = state.device_playback.is_control()
|
let active_missing = state.device_playback.is_personal_control()
|
||||||
&& state
|
&& state
|
||||||
.device_playback
|
.device_playback
|
||||||
.active_device_id
|
.active_device_id
|
||||||
@@ -2959,9 +3028,91 @@ fn handle_app_event(state: &mut AppState, runtime: &mut Runtime, event: AppEvent
|
|||||||
AppEvent::DevicePlayback(snapshot) => {
|
AppEvent::DevicePlayback(snapshot) => {
|
||||||
handle_device_playback_snapshot(state, runtime, snapshot);
|
handle_device_playback_snapshot(state, runtime, snapshot);
|
||||||
}
|
}
|
||||||
|
AppEvent::PlaybackCommand(_)
|
||||||
|
if state.device_playback.role == state::DevicePlaybackRole::Jam =>
|
||||||
|
{
|
||||||
|
tracing::debug!("ignored personal-device playback command while Jam is active");
|
||||||
|
}
|
||||||
AppEvent::PlaybackCommand(command) => {
|
AppEvent::PlaybackCommand(command) => {
|
||||||
handle_playback_command(state, runtime, command);
|
handle_playback_command(state, runtime, command);
|
||||||
}
|
}
|
||||||
|
AppEvent::JamStatus(status) => {
|
||||||
|
state.jam = status.clone();
|
||||||
|
match status.role {
|
||||||
|
crate::jam::JamRole::Host => {
|
||||||
|
state.device_playback.role = state::DevicePlaybackRole::Jam;
|
||||||
|
state.device_playback.jam_host = true;
|
||||||
|
publish_playback_snapshot(state, runtime);
|
||||||
|
}
|
||||||
|
crate::jam::JamRole::Participant => {
|
||||||
|
if state.device_playback.is_audio_owner() {
|
||||||
|
runtime.player.stop();
|
||||||
|
runtime.player_start_pending = false;
|
||||||
|
}
|
||||||
|
state.device_playback.role = state::DevicePlaybackRole::Jam;
|
||||||
|
state.device_playback.jam_host = false;
|
||||||
|
}
|
||||||
|
crate::jam::JamRole::None => {
|
||||||
|
if state.device_playback.role == state::DevicePlaybackRole::Jam {
|
||||||
|
state.device_playback.role = state::DevicePlaybackRole::Active;
|
||||||
|
state.device_playback.jam_host = false;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
AppEvent::JamPlayback(snapshot) => {
|
||||||
|
let local_volume = state.player.volume;
|
||||||
|
become_control_device(state, runtime, snapshot);
|
||||||
|
state.device_playback.role = state::DevicePlaybackRole::Jam;
|
||||||
|
state.device_playback.jam_host = false;
|
||||||
|
state.player.volume = local_volume;
|
||||||
|
state.status_message = Some(format!(
|
||||||
|
"Jam · controlling {}",
|
||||||
|
state.device_playback.active_label()
|
||||||
|
));
|
||||||
|
}
|
||||||
|
AppEvent::JamCommand(command) => {
|
||||||
|
let command = match command {
|
||||||
|
crate::devices::PlaybackCommand::SetState {
|
||||||
|
state: mut wire,
|
||||||
|
seek,
|
||||||
|
} => {
|
||||||
|
wire.volume = state.player.volume;
|
||||||
|
crate::devices::PlaybackCommand::SetState { state: wire, seek }
|
||||||
|
}
|
||||||
|
crate::devices::PlaybackCommand::ActiveChanged { .. } => {
|
||||||
|
state.status_message =
|
||||||
|
Some("Jam cannot transfer audio away from its host".into());
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
};
|
||||||
|
handle_playback_command(state, runtime, command);
|
||||||
|
state.device_playback.role = state::DevicePlaybackRole::Jam;
|
||||||
|
state.device_playback.jam_host = true;
|
||||||
|
publish_playback_snapshot(state, runtime);
|
||||||
|
}
|
||||||
|
AppEvent::JamInvite(result) => match result {
|
||||||
|
Ok(invite) => {
|
||||||
|
if !state.device_playback.is_audio_owner() {
|
||||||
|
transfer_active_to_this_device(state, runtime);
|
||||||
|
}
|
||||||
|
state.jam = runtime.jam.status();
|
||||||
|
state.device_playback.role = state::DevicePlaybackRole::Jam;
|
||||||
|
state.device_playback.jam_host = true;
|
||||||
|
publish_playback_snapshot(state, runtime);
|
||||||
|
state.popup = Some(state::Popup::FedCopyText {
|
||||||
|
title: "Jam invite".to_string(),
|
||||||
|
text: invite,
|
||||||
|
help: "Copied capability lets federation peers control this host player until restart or regeneration.".to_string(),
|
||||||
|
});
|
||||||
|
state.status_message = Some("Jam started".into());
|
||||||
|
}
|
||||||
|
Err(error) => state.status_message = Some(format!("Jam: {error}")),
|
||||||
|
},
|
||||||
|
AppEvent::JamJoined(result) => match result {
|
||||||
|
Ok(message) => state.status_message = Some(message),
|
||||||
|
Err(error) => state.status_message = Some(format!("Jam: {error}")),
|
||||||
|
},
|
||||||
AppEvent::FedSearchLoaded { seq, result } => {
|
AppEvent::FedSearchLoaded { seq, result } => {
|
||||||
if runtime.search_seq.load(std::sync::atomic::Ordering::SeqCst) != seq {
|
if runtime.search_seq.load(std::sync::atomic::Ordering::SeqCst) != seq {
|
||||||
return;
|
return;
|
||||||
|
|||||||
+44
-1
@@ -82,7 +82,7 @@ pub(crate) fn connected_device_rows(state: &AppState) -> Vec<ConnectedDevicePopu
|
|||||||
is_self: true,
|
is_self: true,
|
||||||
online: true,
|
online: true,
|
||||||
section: DevicePresenceSection::Online,
|
section: DevicePresenceSection::Online,
|
||||||
active: state.device_playback.role == crate::app::state::DevicePlaybackRole::Active,
|
active: state.device_playback.is_audio_owner(),
|
||||||
playing: state.player.playing,
|
playing: state.player.playing,
|
||||||
paused: state.player.paused,
|
paused: state.player.paused,
|
||||||
queue_len: state.player.queue.len(),
|
queue_len: state.player.queue.len(),
|
||||||
@@ -386,6 +386,32 @@ fn handle_connected_devices(
|
|||||||
let cursor = cursor.min(last);
|
let cursor = cursor.min(last);
|
||||||
match key.code {
|
match key.code {
|
||||||
KeyCode::Esc | KeyCode::Char('q') => {}
|
KeyCode::Esc | KeyCode::Char('q') => {}
|
||||||
|
KeyCode::Char('h') => {
|
||||||
|
super::perform_effect(state, runtime, crate::app::update::Effect::JamCreate);
|
||||||
|
}
|
||||||
|
KeyCode::Char('J') => {
|
||||||
|
state.popup = Some(Popup::FedInput {
|
||||||
|
field: FedInputField::JamInvite,
|
||||||
|
input: crate::app::input::LineEdit::default(),
|
||||||
|
});
|
||||||
|
}
|
||||||
|
KeyCode::Char('l') if state.jam.role != crate::jam::JamRole::None => {
|
||||||
|
super::perform_effect(state, runtime, crate::app::update::Effect::JamLeave);
|
||||||
|
}
|
||||||
|
KeyCode::Char('c') => {
|
||||||
|
if let Some(invite) = state.jam.invite.as_deref() {
|
||||||
|
match copy_to_clipboard(invite) {
|
||||||
|
Ok(()) => state.status_message = Some("Jam invite copied".into()),
|
||||||
|
Err(error) => {
|
||||||
|
state.status_message = Some(format!("clipboard: {error}"));
|
||||||
|
state.popup = Some(Popup::ConnectedDevices { cursor });
|
||||||
|
}
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
state.status_message = Some("only the Jam host has an invite".into());
|
||||||
|
state.popup = Some(Popup::ConnectedDevices { cursor });
|
||||||
|
}
|
||||||
|
}
|
||||||
KeyCode::Up | KeyCode::Char('k') => {
|
KeyCode::Up | KeyCode::Char('k') => {
|
||||||
state.popup = Some(Popup::ConnectedDevices {
|
state.popup = Some(Popup::ConnectedDevices {
|
||||||
cursor: cursor.saturating_sub(1),
|
cursor: cursor.saturating_sub(1),
|
||||||
@@ -397,6 +423,12 @@ fn handle_connected_devices(
|
|||||||
});
|
});
|
||||||
}
|
}
|
||||||
KeyCode::Enter => {
|
KeyCode::Enter => {
|
||||||
|
if state.device_playback.role == crate::app::state::DevicePlaybackRole::Jam {
|
||||||
|
state.status_message =
|
||||||
|
Some("leave the current Jam before switching personal devices".into());
|
||||||
|
state.popup = Some(Popup::ConnectedDevices { cursor });
|
||||||
|
return;
|
||||||
|
}
|
||||||
if cursor == 0 {
|
if cursor == 0 {
|
||||||
super::transfer_active_to_this_device(state, runtime);
|
super::transfer_active_to_this_device(state, runtime);
|
||||||
state.status_message = Some("active playback moved to this device".into());
|
state.status_message = Some("active playback moved to this device".into());
|
||||||
@@ -499,6 +531,17 @@ fn handle_fed_input(
|
|||||||
);
|
);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
FedInputField::JamInvite => {
|
||||||
|
if value.is_empty() {
|
||||||
|
state.status_message = Some("Jam invite is empty".into());
|
||||||
|
} else {
|
||||||
|
super::perform_effect(
|
||||||
|
state,
|
||||||
|
runtime,
|
||||||
|
crate::app::update::Effect::JamJoin(value),
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
_ => {
|
_ => {
|
||||||
|
|||||||
@@ -805,6 +805,7 @@ pub enum FedInputField {
|
|||||||
ConnectTicket,
|
ConnectTicket,
|
||||||
DeviceName,
|
DeviceName,
|
||||||
ConnectInvite,
|
ConnectInvite,
|
||||||
|
JamInvite,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl FedInputField {
|
impl FedInputField {
|
||||||
@@ -814,6 +815,7 @@ impl FedInputField {
|
|||||||
FedInputField::ConnectTicket => "Connect to peer (paste ticket)",
|
FedInputField::ConnectTicket => "Connect to peer (paste ticket)",
|
||||||
FedInputField::DeviceName => "Device name",
|
FedInputField::DeviceName => "Device name",
|
||||||
FedInputField::ConnectInvite => "Connect device (paste frid://i invite)",
|
FedInputField::ConnectInvite => "Connect device (paste frid://i invite)",
|
||||||
|
FedInputField::JamInvite => "Join Jam (paste frid://j invite)",
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -831,6 +833,9 @@ impl FedInputField {
|
|||||||
FedInputField::ConnectInvite => {
|
FedInputField::ConnectInvite => {
|
||||||
"Paste a frid:// invite generated by another client to add this device to its sync group."
|
"Paste a frid:// invite generated by another client to add this device to its sync group."
|
||||||
}
|
}
|
||||||
|
FedInputField::JamInvite => {
|
||||||
|
"Paste a frid://j capability to control the host player for this Jam only."
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -1210,6 +1215,7 @@ pub enum DevicePlaybackRole {
|
|||||||
#[default]
|
#[default]
|
||||||
Active,
|
Active,
|
||||||
Control,
|
Control,
|
||||||
|
Jam,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl DevicePlaybackRole {
|
impl DevicePlaybackRole {
|
||||||
@@ -1217,6 +1223,7 @@ impl DevicePlaybackRole {
|
|||||||
match self {
|
match self {
|
||||||
DevicePlaybackRole::Active => "active",
|
DevicePlaybackRole::Active => "active",
|
||||||
DevicePlaybackRole::Control => "control",
|
DevicePlaybackRole::Control => "control",
|
||||||
|
DevicePlaybackRole::Jam => "jam",
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -1232,11 +1239,22 @@ pub struct DevicePlaybackState {
|
|||||||
pub local_idle_since_ms: Option<i64>,
|
pub local_idle_since_ms: Option<i64>,
|
||||||
pub remote: BTreeMap<String, crate::devices::PlaybackSnapshot>,
|
pub remote: BTreeMap<String, crate::devices::PlaybackSnapshot>,
|
||||||
pub last_remote_snapshot: Option<crate::devices::PlaybackSnapshot>,
|
pub last_remote_snapshot: Option<crate::devices::PlaybackSnapshot>,
|
||||||
|
pub jam_host: bool,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl DevicePlaybackState {
|
impl DevicePlaybackState {
|
||||||
pub fn is_control(&self) -> bool {
|
pub fn is_control(&self) -> bool {
|
||||||
self.role == DevicePlaybackRole::Control
|
self.role == DevicePlaybackRole::Control
|
||||||
|
|| (self.role == DevicePlaybackRole::Jam && !self.jam_host)
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn is_personal_control(&self) -> bool {
|
||||||
|
self.role == DevicePlaybackRole::Control
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn is_audio_owner(&self) -> bool {
|
||||||
|
self.role == DevicePlaybackRole::Active
|
||||||
|
|| (self.role == DevicePlaybackRole::Jam && self.jam_host)
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn active_label(&self) -> String {
|
pub fn active_label(&self) -> String {
|
||||||
@@ -1264,6 +1282,7 @@ pub struct AppState {
|
|||||||
pub settings_cursor: usize,
|
pub settings_cursor: usize,
|
||||||
pub player: PlayerBar,
|
pub player: PlayerBar,
|
||||||
pub device_playback: DevicePlaybackState,
|
pub device_playback: DevicePlaybackState,
|
||||||
|
pub jam: crate::jam::JamStatus,
|
||||||
pub visualizer: crate::visualizer::VisualizerState,
|
pub visualizer: crate::visualizer::VisualizerState,
|
||||||
pub global: GlobalTab,
|
pub global: GlobalTab,
|
||||||
pub artist_views: HashMap<i64, Loadable<ArtistDetail>>,
|
pub artist_views: HashMap<i64, Loadable<ArtistDetail>>,
|
||||||
|
|||||||
@@ -68,6 +68,12 @@ pub enum Effect {
|
|||||||
DeviceRevoke(String),
|
DeviceRevoke(String),
|
||||||
/// Leave the current personal-device group after publishing self-revoke.
|
/// Leave the current personal-device group after publishing self-revoke.
|
||||||
DeviceLeaveGroup,
|
DeviceLeaveGroup,
|
||||||
|
/// Create or regenerate the runtime Jam capability.
|
||||||
|
JamCreate,
|
||||||
|
/// Join a federation Jam by capability.
|
||||||
|
JamJoin(String),
|
||||||
|
/// Leave the current Jam without changing personal-device state.
|
||||||
|
JamLeave,
|
||||||
/// Assemble the federated artist card (fan-out to the owning peers).
|
/// Assemble the federated artist card (fan-out to the owning peers).
|
||||||
FedOpenArtist(String),
|
FedOpenArtist(String),
|
||||||
/// Download federated tracks into the local library, one by one.
|
/// Download federated tracks into the local library, one by one.
|
||||||
|
|||||||
+41
-1
@@ -416,6 +416,7 @@ struct Running {
|
|||||||
pub struct Federation {
|
pub struct Federation {
|
||||||
library: Arc<Library>,
|
library: Arc<Library>,
|
||||||
devices: Arc<crate::devices::DeviceSync>,
|
devices: Arc<crate::devices::DeviceSync>,
|
||||||
|
jam: Arc<crate::jam::JamManager>,
|
||||||
data_dir: PathBuf,
|
data_dir: PathBuf,
|
||||||
cache_dir: PathBuf,
|
cache_dir: PathBuf,
|
||||||
media_dir: PathBuf,
|
media_dir: PathBuf,
|
||||||
@@ -515,7 +516,11 @@ async fn dht_record_payload_bytes(data_dir: PathBuf, now_ms: u64) -> Result<u64>
|
|||||||
}
|
}
|
||||||
|
|
||||||
impl Federation {
|
impl Federation {
|
||||||
pub fn new(library: Arc<Library>, devices: Arc<crate::devices::DeviceSync>) -> Arc<Self> {
|
pub fn new(
|
||||||
|
library: Arc<Library>,
|
||||||
|
devices: Arc<crate::devices::DeviceSync>,
|
||||||
|
jam: Arc<crate::jam::JamManager>,
|
||||||
|
) -> Arc<Self> {
|
||||||
let dirs = crate::config::project_dirs();
|
let dirs = crate::config::project_dirs();
|
||||||
let data_dir = dirs
|
let data_dir = dirs
|
||||||
.as_ref()
|
.as_ref()
|
||||||
@@ -539,6 +544,7 @@ impl Federation {
|
|||||||
Arc::new(Self {
|
Arc::new(Self {
|
||||||
library,
|
library,
|
||||||
devices,
|
devices,
|
||||||
|
jam,
|
||||||
data_dir,
|
data_dir,
|
||||||
cache_dir,
|
cache_dir,
|
||||||
media_dir,
|
media_dir,
|
||||||
@@ -555,6 +561,27 @@ impl Federation {
|
|||||||
lock(&self.settings).clone()
|
lock(&self.settings).clone()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
pub async fn create_jam(&self) -> Result<String> {
|
||||||
|
let service = {
|
||||||
|
let running = self.running.lock().await;
|
||||||
|
Arc::clone(
|
||||||
|
&running
|
||||||
|
.as_ref()
|
||||||
|
.context("enable federation before creating a Jam")?
|
||||||
|
.service,
|
||||||
|
)
|
||||||
|
};
|
||||||
|
let (device_id, device_name) = self.devices.identity_summary()?;
|
||||||
|
self.jam
|
||||||
|
.create_or_regenerate(service, device_id, device_name)
|
||||||
|
.await
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn join_jam(&self, invite: &str) -> Result<()> {
|
||||||
|
let (device_id, device_name) = self.devices.identity_summary()?;
|
||||||
|
self.jam.join(invite, device_id, device_name)
|
||||||
|
}
|
||||||
|
|
||||||
fn cached_metadata_snapshot(&self) -> Vec<CachedTrackMetadata> {
|
fn cached_metadata_snapshot(&self) -> Vec<CachedTrackMetadata> {
|
||||||
lock(&self.metadata_cache).values().cloned().collect()
|
lock(&self.metadata_cache).values().cloned().collect()
|
||||||
}
|
}
|
||||||
@@ -627,6 +654,8 @@ impl Federation {
|
|||||||
.stream_protocol(CATALOG_ALPN)
|
.stream_protocol(CATALOG_ALPN)
|
||||||
// Personal-device sync (likes, playlists, trusted devices).
|
// Personal-device sync (likes, playlists, trusted devices).
|
||||||
.stream_protocol(crate::devices::SYNC_ALPN)
|
.stream_protocol(crate::devices::SYNC_ALPN)
|
||||||
|
// Capability-scoped shared playback control.
|
||||||
|
.stream_protocol(crate::jam::JAM_ALPN)
|
||||||
.build()
|
.build()
|
||||||
.map_err(|err| anyhow::anyhow!("invalid federation config: {err}"))?;
|
.map_err(|err| anyhow::anyhow!("invalid federation config: {err}"))?;
|
||||||
let (service, mut events) = MusicDhtService::start(config)
|
let (service, mut events) = MusicDhtService::start(config)
|
||||||
@@ -690,6 +719,15 @@ impl Federation {
|
|||||||
let device_tick_task = tokio::spawn(async move {
|
let device_tick_task = tokio::spawn(async move {
|
||||||
crate::devices::sync_loop(device_sync, device_service, device_transport).await;
|
crate::devices::sync_loop(device_sync, device_service, device_transport).await;
|
||||||
});
|
});
|
||||||
|
let jam_acceptor = service
|
||||||
|
.stream_acceptor(crate::jam::JAM_ALPN)
|
||||||
|
.map_err(|err| anyhow::anyhow!("failed to take the Jam acceptor: {err}"))?;
|
||||||
|
let jam_serve_task =
|
||||||
|
tokio::spawn(crate::jam::serve_peers(jam_acceptor, Arc::clone(&self.jam)));
|
||||||
|
let jam_poll_task = tokio::spawn(crate::jam::poll_loop(
|
||||||
|
Arc::clone(&self.jam),
|
||||||
|
Arc::clone(&service),
|
||||||
|
));
|
||||||
|
|
||||||
*guard = Some(Running {
|
*guard = Some(Running {
|
||||||
service,
|
service,
|
||||||
@@ -702,6 +740,8 @@ impl Federation {
|
|||||||
catalog_task,
|
catalog_task,
|
||||||
device_sync_task,
|
device_sync_task,
|
||||||
device_tick_task,
|
device_tick_task,
|
||||||
|
jam_serve_task,
|
||||||
|
jam_poll_task,
|
||||||
],
|
],
|
||||||
});
|
});
|
||||||
self.set_error(None);
|
self.set_error(None);
|
||||||
|
|||||||
+732
@@ -0,0 +1,732 @@
|
|||||||
|
//! Capability-based Jam playback control for independent federation peers.
|
||||||
|
|
||||||
|
use std::collections::{HashMap, HashSet, VecDeque};
|
||||||
|
use std::sync::{Arc, Mutex, MutexGuard};
|
||||||
|
use std::time::Duration;
|
||||||
|
|
||||||
|
use anyhow::{Context, Result};
|
||||||
|
use music_dht::{ByteStream, MusicDhtService, PeerTicket, StreamAcceptor};
|
||||||
|
use serde::{Deserialize, Serialize};
|
||||||
|
use tokio::sync::mpsc;
|
||||||
|
|
||||||
|
use crate::app::event::AppEvent;
|
||||||
|
use crate::devices::{PlaybackCommand, PlaybackSnapshot};
|
||||||
|
|
||||||
|
pub const JAM_ALPN: &[u8] = b"furumi/jam/1";
|
||||||
|
const PROTOCOL_VERSION: u16 = 1;
|
||||||
|
const MAX_LINE: usize = 8 * 1024 * 1024;
|
||||||
|
const MAX_COMMANDS: usize = 128;
|
||||||
|
const PARTICIPANT_TTL_MS: i64 = 30 * 60 * 1_000;
|
||||||
|
const POLL_INTERVAL: Duration = Duration::from_millis(500);
|
||||||
|
|
||||||
|
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||||
|
pub enum JamRole {
|
||||||
|
None,
|
||||||
|
Host,
|
||||||
|
Participant,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, Clone)]
|
||||||
|
pub struct JamStatus {
|
||||||
|
pub role: JamRole,
|
||||||
|
pub jam_id: Option<String>,
|
||||||
|
pub host_name: Option<String>,
|
||||||
|
pub invite: Option<String>,
|
||||||
|
pub participants: Vec<JamParticipant>,
|
||||||
|
pub connected: bool,
|
||||||
|
pub last_error: Option<String>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl Default for JamStatus {
|
||||||
|
fn default() -> Self {
|
||||||
|
Self {
|
||||||
|
role: JamRole::None,
|
||||||
|
jam_id: None,
|
||||||
|
host_name: None,
|
||||||
|
invite: None,
|
||||||
|
participants: Vec::new(),
|
||||||
|
connected: false,
|
||||||
|
last_error: None,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||||
|
pub struct JamParticipant {
|
||||||
|
pub participant_id: String,
|
||||||
|
pub name: String,
|
||||||
|
pub last_seen_ms: i64,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||||
|
struct JamInvite {
|
||||||
|
v: u16,
|
||||||
|
#[serde(rename = "t")]
|
||||||
|
ticket: String,
|
||||||
|
#[serde(rename = "j")]
|
||||||
|
jam_id: String,
|
||||||
|
#[serde(rename = "s")]
|
||||||
|
secret: String,
|
||||||
|
#[serde(rename = "d")]
|
||||||
|
host_device_id: String,
|
||||||
|
#[serde(rename = "n")]
|
||||||
|
host_name: String,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||||
|
struct JamCommand {
|
||||||
|
command_id: String,
|
||||||
|
participant_id: String,
|
||||||
|
command: PlaybackCommand,
|
||||||
|
sent_at_ms: i64,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||||
|
#[serde(tag = "type", rename_all = "snake_case")]
|
||||||
|
enum WireMessage {
|
||||||
|
Poll {
|
||||||
|
version: u16,
|
||||||
|
jam_id: String,
|
||||||
|
secret: String,
|
||||||
|
participant: JamParticipant,
|
||||||
|
#[serde(default)]
|
||||||
|
commands: Vec<JamCommand>,
|
||||||
|
},
|
||||||
|
Snapshot {
|
||||||
|
accepted: bool,
|
||||||
|
#[serde(default)]
|
||||||
|
error: Option<String>,
|
||||||
|
#[serde(default)]
|
||||||
|
acknowledged_command_ids: Vec<String>,
|
||||||
|
#[serde(default)]
|
||||||
|
playback: Option<PlaybackSnapshot>,
|
||||||
|
#[serde(default)]
|
||||||
|
participants: Vec<JamParticipant>,
|
||||||
|
host_time_ms: i64,
|
||||||
|
},
|
||||||
|
Leave {
|
||||||
|
version: u16,
|
||||||
|
jam_id: String,
|
||||||
|
secret: String,
|
||||||
|
participant_id: String,
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Default)]
|
||||||
|
struct HostState {
|
||||||
|
invite: Option<JamInvite>,
|
||||||
|
invite_uri: Option<String>,
|
||||||
|
playback: Option<PlaybackSnapshot>,
|
||||||
|
participants: HashMap<String, JamParticipant>,
|
||||||
|
seen_commands: HashSet<String>,
|
||||||
|
seen_order: VecDeque<String>,
|
||||||
|
}
|
||||||
|
|
||||||
|
struct JoinedState {
|
||||||
|
invite: JamInvite,
|
||||||
|
participant: JamParticipant,
|
||||||
|
pending: VecDeque<JamCommand>,
|
||||||
|
participants: Vec<JamParticipant>,
|
||||||
|
connected: bool,
|
||||||
|
last_error: Option<String>,
|
||||||
|
last_activity_ms: i64,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Default)]
|
||||||
|
struct State {
|
||||||
|
host: HostState,
|
||||||
|
joined: Option<JoinedState>,
|
||||||
|
}
|
||||||
|
|
||||||
|
pub struct JamManager {
|
||||||
|
state: Mutex<State>,
|
||||||
|
event_tx: mpsc::UnboundedSender<AppEvent>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl JamManager {
|
||||||
|
pub fn new(event_tx: mpsc::UnboundedSender<AppEvent>) -> Arc<Self> {
|
||||||
|
Arc::new(Self {
|
||||||
|
state: Mutex::new(State::default()),
|
||||||
|
event_tx,
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
pub async fn create_or_regenerate(
|
||||||
|
&self,
|
||||||
|
service: Arc<MusicDhtService>,
|
||||||
|
host_device_id: String,
|
||||||
|
host_name: String,
|
||||||
|
) -> Result<String> {
|
||||||
|
let invite = JamInvite {
|
||||||
|
v: PROTOCOL_VERSION,
|
||||||
|
ticket: service.ticket().await?.to_string(),
|
||||||
|
jam_id: format!("jam_{}", random_hex(12)),
|
||||||
|
secret: random_hex(32),
|
||||||
|
host_device_id,
|
||||||
|
host_name,
|
||||||
|
};
|
||||||
|
let uri = encode_invite(&invite)?;
|
||||||
|
let mut state = lock(&self.state);
|
||||||
|
state.joined = None;
|
||||||
|
state.host = HostState {
|
||||||
|
invite: Some(invite),
|
||||||
|
invite_uri: Some(uri.clone()),
|
||||||
|
..HostState::default()
|
||||||
|
};
|
||||||
|
Ok(uri)
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn join(&self, uri: &str, participant_id: String, name: String) -> Result<()> {
|
||||||
|
let invite = parse_invite(uri)?;
|
||||||
|
let mut state = lock(&self.state);
|
||||||
|
state.host = HostState::default();
|
||||||
|
state.joined = Some(JoinedState {
|
||||||
|
invite,
|
||||||
|
participant: JamParticipant {
|
||||||
|
participant_id,
|
||||||
|
name,
|
||||||
|
last_seen_ms: now_ms(),
|
||||||
|
},
|
||||||
|
pending: VecDeque::new(),
|
||||||
|
participants: Vec::new(),
|
||||||
|
connected: false,
|
||||||
|
last_error: None,
|
||||||
|
last_activity_ms: now_ms(),
|
||||||
|
});
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn leave(&self) {
|
||||||
|
let mut state = lock(&self.state);
|
||||||
|
state.joined = None;
|
||||||
|
state.host = HostState::default();
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn status(&self) -> JamStatus {
|
||||||
|
let state = lock(&self.state);
|
||||||
|
if let Some(joined) = &state.joined {
|
||||||
|
return JamStatus {
|
||||||
|
role: JamRole::Participant,
|
||||||
|
jam_id: Some(joined.invite.jam_id.clone()),
|
||||||
|
host_name: Some(joined.invite.host_name.clone()),
|
||||||
|
invite: None,
|
||||||
|
participants: joined.participants.clone(),
|
||||||
|
connected: joined.connected,
|
||||||
|
last_error: joined.last_error.clone(),
|
||||||
|
};
|
||||||
|
}
|
||||||
|
if let Some(invite) = &state.host.invite {
|
||||||
|
return JamStatus {
|
||||||
|
role: JamRole::Host,
|
||||||
|
jam_id: Some(invite.jam_id.clone()),
|
||||||
|
host_name: Some(invite.host_name.clone()),
|
||||||
|
invite: state.host.invite_uri.clone(),
|
||||||
|
participants: state.host.participants.values().cloned().collect(),
|
||||||
|
connected: true,
|
||||||
|
last_error: None,
|
||||||
|
};
|
||||||
|
}
|
||||||
|
JamStatus::default()
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn publish_host_playback(&self, snapshot: PlaybackSnapshot) {
|
||||||
|
let mut state = lock(&self.state);
|
||||||
|
if state.host.invite.is_some() {
|
||||||
|
state.host.playback = Some(snapshot);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn submit_command(&self, command: PlaybackCommand) -> Result<()> {
|
||||||
|
let mut state = lock(&self.state);
|
||||||
|
let joined = state
|
||||||
|
.joined
|
||||||
|
.as_mut()
|
||||||
|
.context("this player is not controlling a Jam")?;
|
||||||
|
if joined.pending.len() >= MAX_COMMANDS {
|
||||||
|
joined.pending.pop_front();
|
||||||
|
}
|
||||||
|
joined.pending.push_back(JamCommand {
|
||||||
|
command_id: format!("cmd_{}", random_hex(16)),
|
||||||
|
participant_id: joined.participant.participant_id.clone(),
|
||||||
|
command,
|
||||||
|
sent_at_ms: now_ms(),
|
||||||
|
});
|
||||||
|
joined.last_activity_ms = now_ms();
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn poll_once(&self, service: Arc<MusicDhtService>) -> Result<()> {
|
||||||
|
let (invite, participant, commands) = {
|
||||||
|
let state = lock(&self.state);
|
||||||
|
let joined = state.joined.as_ref().context("not joined")?;
|
||||||
|
(
|
||||||
|
joined.invite.clone(),
|
||||||
|
joined.participant.clone(),
|
||||||
|
joined.pending.iter().cloned().collect::<Vec<_>>(),
|
||||||
|
)
|
||||||
|
};
|
||||||
|
let ticket: PeerTicket = invite.ticket.parse().context("invalid Jam host ticket")?;
|
||||||
|
let peer = service.connect(ticket).await?;
|
||||||
|
let mut stream = service.open_stream(peer, JAM_ALPN).await?;
|
||||||
|
write_message(
|
||||||
|
&mut stream,
|
||||||
|
&WireMessage::Poll {
|
||||||
|
version: PROTOCOL_VERSION,
|
||||||
|
jam_id: invite.jam_id.clone(),
|
||||||
|
secret: invite.secret.clone(),
|
||||||
|
participant,
|
||||||
|
commands,
|
||||||
|
},
|
||||||
|
)
|
||||||
|
.await?;
|
||||||
|
stream.send.finish()?;
|
||||||
|
let response = read_message(&mut stream).await?;
|
||||||
|
let WireMessage::Snapshot {
|
||||||
|
accepted,
|
||||||
|
error,
|
||||||
|
acknowledged_command_ids,
|
||||||
|
playback,
|
||||||
|
participants,
|
||||||
|
..
|
||||||
|
} = response
|
||||||
|
else {
|
||||||
|
anyhow::bail!("unexpected Jam response");
|
||||||
|
};
|
||||||
|
anyhow::ensure!(
|
||||||
|
accepted,
|
||||||
|
"{}",
|
||||||
|
error.unwrap_or_else(|| "Jam refused".into())
|
||||||
|
);
|
||||||
|
{
|
||||||
|
let mut state = lock(&self.state);
|
||||||
|
let Some(joined) = state.joined.as_mut() else {
|
||||||
|
return Ok(());
|
||||||
|
};
|
||||||
|
if joined.invite.jam_id != invite.jam_id {
|
||||||
|
return Ok(());
|
||||||
|
}
|
||||||
|
let acknowledged = acknowledged_command_ids.into_iter().collect::<HashSet<_>>();
|
||||||
|
joined
|
||||||
|
.pending
|
||||||
|
.retain(|command| !acknowledged.contains(&command.command_id));
|
||||||
|
joined.participants = participants;
|
||||||
|
joined.connected = true;
|
||||||
|
joined.last_error = None;
|
||||||
|
joined.participant.last_seen_ms = now_ms();
|
||||||
|
if playback
|
||||||
|
.as_ref()
|
||||||
|
.is_some_and(|snapshot| snapshot.state.playing && !snapshot.state.paused)
|
||||||
|
{
|
||||||
|
joined.last_activity_ms = now_ms();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if let Some(playback) = playback {
|
||||||
|
let _ = self.event_tx.send(AppEvent::JamPlayback(playback));
|
||||||
|
}
|
||||||
|
let _ = self.event_tx.send(AppEvent::JamStatus(self.status()));
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
pub async fn serve_peers(mut acceptor: StreamAcceptor, manager: Arc<JamManager>) {
|
||||||
|
while let Some(stream) = acceptor.accept().await {
|
||||||
|
let manager = Arc::clone(&manager);
|
||||||
|
tokio::spawn(async move {
|
||||||
|
if let Err(err) = serve_one(stream, manager).await {
|
||||||
|
tracing::debug!("Jam stream failed: {err:#}");
|
||||||
|
}
|
||||||
|
});
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn serve_one(mut stream: ByteStream, manager: Arc<JamManager>) -> Result<()> {
|
||||||
|
match read_message(&mut stream).await? {
|
||||||
|
WireMessage::Poll {
|
||||||
|
version,
|
||||||
|
jam_id,
|
||||||
|
secret,
|
||||||
|
mut participant,
|
||||||
|
commands,
|
||||||
|
} => {
|
||||||
|
let (accepted, error, acknowledged, playback, participants) = {
|
||||||
|
let mut state = lock(&manager.state);
|
||||||
|
let valid =
|
||||||
|
version == PROTOCOL_VERSION
|
||||||
|
&& state.host.invite.as_ref().is_some_and(|invite| {
|
||||||
|
invite.jam_id == jam_id && invite.secret == secret
|
||||||
|
});
|
||||||
|
if !valid {
|
||||||
|
(
|
||||||
|
false,
|
||||||
|
Some("invalid Jam capability".to_string()),
|
||||||
|
vec![],
|
||||||
|
None,
|
||||||
|
vec![],
|
||||||
|
)
|
||||||
|
} else {
|
||||||
|
let now = now_ms();
|
||||||
|
state.host.participants.retain(|_, row| {
|
||||||
|
now.saturating_sub(row.last_seen_ms) <= PARTICIPANT_TTL_MS
|
||||||
|
});
|
||||||
|
participant.last_seen_ms = now;
|
||||||
|
state
|
||||||
|
.host
|
||||||
|
.participants
|
||||||
|
.insert(participant.participant_id.clone(), participant);
|
||||||
|
let mut acknowledged = Vec::new();
|
||||||
|
for command in commands.into_iter().take(MAX_COMMANDS) {
|
||||||
|
acknowledged.push(command.command_id.clone());
|
||||||
|
if state.host.seen_commands.insert(command.command_id.clone()) {
|
||||||
|
state.host.seen_order.push_back(command.command_id.clone());
|
||||||
|
let _ = manager.event_tx.send(AppEvent::JamCommand(command.command));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
while state.host.seen_order.len() > 4096 {
|
||||||
|
if let Some(id) = state.host.seen_order.pop_front() {
|
||||||
|
state.host.seen_commands.remove(&id);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
(
|
||||||
|
true,
|
||||||
|
None,
|
||||||
|
acknowledged,
|
||||||
|
state.host.playback.clone(),
|
||||||
|
state.host.participants.values().cloned().collect(),
|
||||||
|
)
|
||||||
|
}
|
||||||
|
};
|
||||||
|
write_message(
|
||||||
|
&mut stream,
|
||||||
|
&WireMessage::Snapshot {
|
||||||
|
accepted,
|
||||||
|
error,
|
||||||
|
acknowledged_command_ids: acknowledged,
|
||||||
|
playback,
|
||||||
|
participants,
|
||||||
|
host_time_ms: now_ms(),
|
||||||
|
},
|
||||||
|
)
|
||||||
|
.await?;
|
||||||
|
stream.send.finish()?;
|
||||||
|
let _ = tokio::time::timeout(Duration::from_secs(2), stream.send.stopped()).await;
|
||||||
|
let _ = manager.event_tx.send(AppEvent::JamStatus(manager.status()));
|
||||||
|
}
|
||||||
|
WireMessage::Leave {
|
||||||
|
version,
|
||||||
|
jam_id,
|
||||||
|
secret,
|
||||||
|
participant_id,
|
||||||
|
} => {
|
||||||
|
let mut state = lock(&manager.state);
|
||||||
|
if version == PROTOCOL_VERSION
|
||||||
|
&& state
|
||||||
|
.host
|
||||||
|
.invite
|
||||||
|
.as_ref()
|
||||||
|
.is_some_and(|invite| invite.jam_id == jam_id && invite.secret == secret)
|
||||||
|
{
|
||||||
|
state.host.participants.remove(&participant_id);
|
||||||
|
}
|
||||||
|
drop(state);
|
||||||
|
let _ = manager.event_tx.send(AppEvent::JamStatus(manager.status()));
|
||||||
|
}
|
||||||
|
WireMessage::Snapshot { .. } => anyhow::bail!("unexpected Jam snapshot"),
|
||||||
|
}
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
pub async fn poll_loop(manager: Arc<JamManager>, service: Arc<MusicDhtService>) {
|
||||||
|
let mut interval = tokio::time::interval(POLL_INTERVAL);
|
||||||
|
loop {
|
||||||
|
interval.tick().await;
|
||||||
|
if manager.status().role != JamRole::Participant {
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
let expired = {
|
||||||
|
let state = lock(&manager.state);
|
||||||
|
state.joined.as_ref().is_some_and(|joined| {
|
||||||
|
now_ms().saturating_sub(joined.last_activity_ms) > PARTICIPANT_TTL_MS
|
||||||
|
})
|
||||||
|
};
|
||||||
|
if expired {
|
||||||
|
manager.leave();
|
||||||
|
let _ = manager.event_tx.send(AppEvent::JamStatus(manager.status()));
|
||||||
|
let _ = manager.event_tx.send(AppEvent::StatusMessage(
|
||||||
|
"Jam ended after 30 minutes without playback or commands".into(),
|
||||||
|
));
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
if let Err(err) = manager.poll_once(Arc::clone(&service)).await {
|
||||||
|
{
|
||||||
|
let mut state = lock(&manager.state);
|
||||||
|
if let Some(joined) = state.joined.as_mut() {
|
||||||
|
joined.connected = false;
|
||||||
|
joined.last_error = Some(format!("{err:#}"));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
let _ = manager.event_tx.send(AppEvent::JamStatus(manager.status()));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn write_message(stream: &mut ByteStream, message: &WireMessage) -> Result<()> {
|
||||||
|
let mut bytes = serde_json::to_vec(message)?;
|
||||||
|
anyhow::ensure!(bytes.len() <= MAX_LINE, "Jam message is too large");
|
||||||
|
bytes.push(b'\n');
|
||||||
|
stream.send.write_all(&bytes).await?;
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn read_message(stream: &mut ByteStream) -> Result<WireMessage> {
|
||||||
|
let mut bytes = Vec::new();
|
||||||
|
let mut byte = [0_u8; 1];
|
||||||
|
loop {
|
||||||
|
let read = stream.recv.read(&mut byte).await?;
|
||||||
|
if read.unwrap_or(0) == 0 || byte[0] == b'\n' {
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
bytes.push(byte[0]);
|
||||||
|
anyhow::ensure!(bytes.len() <= MAX_LINE, "Jam message is too large");
|
||||||
|
}
|
||||||
|
anyhow::ensure!(!bytes.is_empty(), "empty Jam message");
|
||||||
|
Ok(serde_json::from_slice(&bytes)?)
|
||||||
|
}
|
||||||
|
|
||||||
|
fn encode_invite(invite: &JamInvite) -> Result<String> {
|
||||||
|
Ok(format!(
|
||||||
|
"frid://j/{}",
|
||||||
|
base64url_encode(&serde_json::to_vec(invite)?)
|
||||||
|
))
|
||||||
|
}
|
||||||
|
|
||||||
|
fn parse_invite(uri: &str) -> Result<JamInvite> {
|
||||||
|
let encoded = uri
|
||||||
|
.trim()
|
||||||
|
.strip_prefix("frid://j/")
|
||||||
|
.context("expected frid://j invite")?;
|
||||||
|
let invite: JamInvite = serde_json::from_slice(&base64url_decode(encoded)?)?;
|
||||||
|
anyhow::ensure!(
|
||||||
|
invite.v == PROTOCOL_VERSION,
|
||||||
|
"unsupported Jam invite version"
|
||||||
|
);
|
||||||
|
anyhow::ensure!(
|
||||||
|
!invite.ticket.is_empty()
|
||||||
|
&& !invite.jam_id.is_empty()
|
||||||
|
&& invite.secret.len() >= 16
|
||||||
|
&& !invite.host_device_id.is_empty(),
|
||||||
|
"incomplete Jam invite"
|
||||||
|
);
|
||||||
|
Ok(invite)
|
||||||
|
}
|
||||||
|
|
||||||
|
fn now_ms() -> i64 {
|
||||||
|
std::time::SystemTime::now()
|
||||||
|
.duration_since(std::time::UNIX_EPOCH)
|
||||||
|
.map(|duration| duration.as_millis() as i64)
|
||||||
|
.unwrap_or(0)
|
||||||
|
}
|
||||||
|
|
||||||
|
fn random_hex(bytes: usize) -> String {
|
||||||
|
use std::sync::atomic::{AtomicU64, Ordering};
|
||||||
|
static COUNTER: AtomicU64 = AtomicU64::new(1);
|
||||||
|
let seed = format!(
|
||||||
|
"{}:{}:{}",
|
||||||
|
now_ms(),
|
||||||
|
std::process::id(),
|
||||||
|
COUNTER.fetch_add(1, Ordering::Relaxed)
|
||||||
|
);
|
||||||
|
let hash = blake3::hash(seed.as_bytes()).to_hex().to_string();
|
||||||
|
hash[..(bytes * 2).min(hash.len())].to_string()
|
||||||
|
}
|
||||||
|
|
||||||
|
fn lock<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
|
||||||
|
mutex
|
||||||
|
.lock()
|
||||||
|
.unwrap_or_else(std::sync::PoisonError::into_inner)
|
||||||
|
}
|
||||||
|
|
||||||
|
fn base64url_encode(bytes: &[u8]) -> String {
|
||||||
|
const TABLE: &[u8; 64] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_";
|
||||||
|
let mut out = String::new();
|
||||||
|
for chunk in bytes.chunks(3) {
|
||||||
|
let b0 = chunk[0];
|
||||||
|
let b1 = chunk.get(1).copied().unwrap_or(0);
|
||||||
|
let b2 = chunk.get(2).copied().unwrap_or(0);
|
||||||
|
out.push(TABLE[(b0 >> 2) as usize] as char);
|
||||||
|
out.push(TABLE[(((b0 & 3) << 4) | (b1 >> 4)) as usize] as char);
|
||||||
|
if chunk.len() > 1 {
|
||||||
|
out.push(TABLE[(((b1 & 15) << 2) | (b2 >> 6)) as usize] as char);
|
||||||
|
}
|
||||||
|
if chunk.len() > 2 {
|
||||||
|
out.push(TABLE[(b2 & 63) as usize] as char);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
out
|
||||||
|
}
|
||||||
|
|
||||||
|
fn base64url_decode(value: &str) -> Result<Vec<u8>> {
|
||||||
|
fn decode(byte: u8) -> Option<u8> {
|
||||||
|
match byte {
|
||||||
|
b'A'..=b'Z' => Some(byte - b'A'),
|
||||||
|
b'a'..=b'z' => Some(byte - b'a' + 26),
|
||||||
|
b'0'..=b'9' => Some(byte - b'0' + 52),
|
||||||
|
b'-' => Some(62),
|
||||||
|
b'_' => Some(63),
|
||||||
|
_ => None,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
let bytes = value.as_bytes();
|
||||||
|
anyhow::ensure!(bytes.len() % 4 != 1, "invalid base64url Jam invite");
|
||||||
|
let mut out = Vec::with_capacity(bytes.len() * 3 / 4);
|
||||||
|
let mut i = 0;
|
||||||
|
while i < bytes.len() {
|
||||||
|
let a = decode(bytes[i]).context("invalid base64url Jam invite")?;
|
||||||
|
let b = decode(*bytes.get(i + 1).context("truncated Jam invite")?)
|
||||||
|
.context("invalid base64url Jam invite")?;
|
||||||
|
let c = bytes.get(i + 2).and_then(|byte| decode(*byte));
|
||||||
|
let d = bytes.get(i + 3).and_then(|byte| decode(*byte));
|
||||||
|
out.push((a << 2) | (b >> 4));
|
||||||
|
if let Some(c) = c {
|
||||||
|
out.push((b << 4) | (c >> 2));
|
||||||
|
if let Some(d) = d {
|
||||||
|
out.push((c << 6) | d);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
i += 4;
|
||||||
|
}
|
||||||
|
Ok(out)
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use super::*;
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn invite_round_trips_and_is_separate_from_pairing() {
|
||||||
|
let invite = JamInvite {
|
||||||
|
v: PROTOCOL_VERSION,
|
||||||
|
ticket: "ticket".into(),
|
||||||
|
jam_id: "jam_test".into(),
|
||||||
|
secret: "0123456789abcdef".into(),
|
||||||
|
host_device_id: "dev_host".into(),
|
||||||
|
host_name: "Host".into(),
|
||||||
|
};
|
||||||
|
let uri = encode_invite(&invite).unwrap();
|
||||||
|
assert!(uri.starts_with("frid://j/"));
|
||||||
|
assert_eq!(parse_invite(&uri).unwrap().jam_id, "jam_test");
|
||||||
|
assert!(parse_invite("frid://i/abcd").is_err());
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn participant_receives_host_state_and_host_receives_command() {
|
||||||
|
let unique = random_hex(8);
|
||||||
|
let host_dir = std::env::temp_dir().join(format!("furumi-jam-host-{unique}"));
|
||||||
|
let guest_dir = std::env::temp_dir().join(format!("furumi-jam-guest-{unique}"));
|
||||||
|
std::fs::create_dir_all(&host_dir).unwrap();
|
||||||
|
std::fs::create_dir_all(&guest_dir).unwrap();
|
||||||
|
let network = music_dht::NetworkId::from_name(&format!("jam-test-{unique}"));
|
||||||
|
let host_config = music_dht::MusicDhtConfig::builder()
|
||||||
|
.data_dir(&host_dir)
|
||||||
|
.network_id(network)
|
||||||
|
.stream_protocol(JAM_ALPN)
|
||||||
|
.build()
|
||||||
|
.unwrap();
|
||||||
|
let guest_config = music_dht::MusicDhtConfig::builder()
|
||||||
|
.data_dir(&guest_dir)
|
||||||
|
.network_id(network)
|
||||||
|
.stream_protocol(JAM_ALPN)
|
||||||
|
.build()
|
||||||
|
.unwrap();
|
||||||
|
let (host_service, mut host_events) = music_dht::MusicDhtService::start(host_config)
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
let (guest_service, mut guest_events) = music_dht::MusicDhtService::start(guest_config)
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
let host_service = Arc::new(host_service);
|
||||||
|
let guest_service = Arc::new(guest_service);
|
||||||
|
let host_drain = tokio::spawn(async move { while host_events.recv().await.is_some() {} });
|
||||||
|
let guest_drain = tokio::spawn(async move { while guest_events.recv().await.is_some() {} });
|
||||||
|
|
||||||
|
let (host_tx, mut host_rx) = mpsc::unbounded_channel();
|
||||||
|
let (guest_tx, mut guest_rx) = mpsc::unbounded_channel();
|
||||||
|
let host = JamManager::new(host_tx);
|
||||||
|
let guest = JamManager::new(guest_tx);
|
||||||
|
let invite = host
|
||||||
|
.create_or_regenerate(Arc::clone(&host_service), "dev_host".into(), "Host".into())
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
guest
|
||||||
|
.join(&invite, "dev_guest".into(), "Guest".into())
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
let playback_state = crate::devices::PlaybackStateWire {
|
||||||
|
queue: Vec::new(),
|
||||||
|
queue_pos: 0,
|
||||||
|
playing: false,
|
||||||
|
paused: false,
|
||||||
|
idle_since_ms: Some(now_ms()),
|
||||||
|
position_secs: 0.0,
|
||||||
|
volume: 73,
|
||||||
|
shuffle: false,
|
||||||
|
repeat: crate::devices::PlaybackRepeat::Off,
|
||||||
|
};
|
||||||
|
host.publish_host_playback(PlaybackSnapshot {
|
||||||
|
device_id: "dev_host".into(),
|
||||||
|
device_name: "Host".into(),
|
||||||
|
active: true,
|
||||||
|
updated_at_ms: now_ms(),
|
||||||
|
state: playback_state.clone(),
|
||||||
|
});
|
||||||
|
let acceptor = host_service.stream_acceptor(JAM_ALPN).unwrap();
|
||||||
|
let server = tokio::spawn(serve_peers(acceptor, Arc::clone(&host)));
|
||||||
|
|
||||||
|
guest.poll_once(Arc::clone(&guest_service)).await.unwrap();
|
||||||
|
let playback = tokio::time::timeout(Duration::from_secs(3), async {
|
||||||
|
loop {
|
||||||
|
if let Some(AppEvent::JamPlayback(snapshot)) = guest_rx.recv().await {
|
||||||
|
break snapshot;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
})
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
assert_eq!(playback.device_id, "dev_host");
|
||||||
|
assert_eq!(playback.state.volume, 73);
|
||||||
|
|
||||||
|
guest
|
||||||
|
.submit_command(PlaybackCommand::SetState {
|
||||||
|
state: crate::devices::PlaybackStateWire {
|
||||||
|
paused: true,
|
||||||
|
..playback_state
|
||||||
|
},
|
||||||
|
seek: false,
|
||||||
|
})
|
||||||
|
.unwrap();
|
||||||
|
guest.poll_once(Arc::clone(&guest_service)).await.unwrap();
|
||||||
|
let command = tokio::time::timeout(Duration::from_secs(3), async {
|
||||||
|
loop {
|
||||||
|
if let Some(AppEvent::JamCommand(command)) = host_rx.recv().await {
|
||||||
|
break command;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
})
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
assert!(matches!(
|
||||||
|
command,
|
||||||
|
PlaybackCommand::SetState {
|
||||||
|
state: crate::devices::PlaybackStateWire { paused: true, .. },
|
||||||
|
seek: false
|
||||||
|
}
|
||||||
|
));
|
||||||
|
|
||||||
|
server.abort();
|
||||||
|
host_drain.abort();
|
||||||
|
guest_drain.abort();
|
||||||
|
host_service.shutdown().await.unwrap();
|
||||||
|
guest_service.shutdown().await.unwrap();
|
||||||
|
let _ = std::fs::remove_dir_all(host_dir);
|
||||||
|
let _ = std::fs::remove_dir_all(guest_dir);
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -3,6 +3,7 @@ mod art;
|
|||||||
mod config;
|
mod config;
|
||||||
mod devices;
|
mod devices;
|
||||||
mod federation;
|
mod federation;
|
||||||
|
mod jam;
|
||||||
mod library;
|
mod library;
|
||||||
mod media;
|
mod media;
|
||||||
mod player;
|
mod player;
|
||||||
|
|||||||
+52
-6
@@ -440,7 +440,7 @@ fn draw_connected_devices(frame: &mut Frame, state: &AppState, cursor: usize) {
|
|||||||
display_lines.push(DisplayLine::Row(index));
|
display_lines.push(DisplayLine::Row(index));
|
||||||
}
|
}
|
||||||
let height =
|
let height =
|
||||||
(display_lines.len() as u16 + 9).clamp(11, frame.area().height.saturating_sub(2).max(11));
|
(display_lines.len() as u16 + 14).clamp(16, frame.area().height.saturating_sub(2).max(16));
|
||||||
let area = centered(frame.area(), 76, height);
|
let area = centered(frame.area(), 76, height);
|
||||||
let block = Block::bordered()
|
let block = Block::bordered()
|
||||||
.title(" Connected devices ")
|
.title(" Connected devices ")
|
||||||
@@ -450,9 +450,10 @@ fn draw_connected_devices(frame: &mut Frame, state: &AppState, cursor: usize) {
|
|||||||
frame.render_widget(Clear, area);
|
frame.render_widget(Clear, area);
|
||||||
frame.render_widget(block, area);
|
frame.render_widget(block, area);
|
||||||
|
|
||||||
let [summary_area, this_area, other_area, hint_area] = Layout::vertical([
|
let [summary_area, this_area, jam_area, other_area, hint_area] = Layout::vertical([
|
||||||
Constraint::Length(1),
|
Constraint::Length(1),
|
||||||
Constraint::Length(3),
|
Constraint::Length(3),
|
||||||
|
Constraint::Length(4),
|
||||||
Constraint::Min(1),
|
Constraint::Min(1),
|
||||||
Constraint::Length(2),
|
Constraint::Length(2),
|
||||||
])
|
])
|
||||||
@@ -479,8 +480,7 @@ fn draw_connected_devices(frame: &mut Frame, state: &AppState, cursor: usize) {
|
|||||||
width: this_area.width,
|
width: this_area.width,
|
||||||
height: 1,
|
height: 1,
|
||||||
};
|
};
|
||||||
let action_label =
|
let action_label = if state.device_playback.is_audio_owner() {
|
||||||
if state.device_playback.role == crate::app::state::DevicePlaybackRole::Active {
|
|
||||||
"This device is active"
|
"This device is active"
|
||||||
} else {
|
} else {
|
||||||
"Make this device active"
|
"Make this device active"
|
||||||
@@ -501,7 +501,53 @@ fn draw_connected_devices(frame: &mut Frame, state: &AppState, cursor: usize) {
|
|||||||
state,
|
state,
|
||||||
);
|
);
|
||||||
|
|
||||||
render_subtitle(frame, other_area, state, "Other devices");
|
render_subtitle(frame, jam_area, state, "Jam");
|
||||||
|
let jam_status = match state.jam.role {
|
||||||
|
crate::jam::JamRole::None => "inactive · h host · J join".to_string(),
|
||||||
|
crate::jam::JamRole::Host => format!(
|
||||||
|
"HOST · {} peer(s) · c copy · h new · l leave",
|
||||||
|
state.jam.participants.len()
|
||||||
|
),
|
||||||
|
crate::jam::JamRole::Participant => format!(
|
||||||
|
"{} · {} · {} participant(s) · l leave",
|
||||||
|
if state.jam.connected {
|
||||||
|
"CONNECTED"
|
||||||
|
} else {
|
||||||
|
"RECONNECTING"
|
||||||
|
},
|
||||||
|
state.jam.host_name.as_deref().unwrap_or("Jam host"),
|
||||||
|
state.jam.participants.len()
|
||||||
|
),
|
||||||
|
};
|
||||||
|
frame.render_widget(
|
||||||
|
Paragraph::new(Line::styled(
|
||||||
|
format!(" {jam_status}"),
|
||||||
|
if state.jam.role == crate::jam::JamRole::None {
|
||||||
|
theme::dim()
|
||||||
|
} else {
|
||||||
|
theme::accent_for(state)
|
||||||
|
},
|
||||||
|
)),
|
||||||
|
Rect {
|
||||||
|
x: jam_area.x,
|
||||||
|
y: jam_area.y + 1,
|
||||||
|
width: jam_area.width,
|
||||||
|
height: 1,
|
||||||
|
},
|
||||||
|
);
|
||||||
|
if let Some(error) = state.jam.last_error.as_deref() {
|
||||||
|
frame.render_widget(
|
||||||
|
Paragraph::new(Line::styled(format!(" {error}"), theme::dim())),
|
||||||
|
Rect {
|
||||||
|
x: jam_area.x,
|
||||||
|
y: jam_area.y + 2,
|
||||||
|
width: jam_area.width,
|
||||||
|
height: 1,
|
||||||
|
},
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
render_subtitle(frame, other_area, state, "My devices");
|
||||||
let list_area = Rect {
|
let list_area = Rect {
|
||||||
x: other_area.x,
|
x: other_area.x,
|
||||||
y: other_area.y + 1,
|
y: other_area.y + 1,
|
||||||
@@ -560,7 +606,7 @@ fn draw_connected_devices(frame: &mut Frame, state: &AppState, cursor: usize) {
|
|||||||
);
|
);
|
||||||
frame.render_widget(
|
frame.render_widget(
|
||||||
Paragraph::new(Line::styled(
|
Paragraph::new(Line::styled(
|
||||||
"enter: activate selected device / control active selected · esc close",
|
"enter device · h host Jam · J join · c copy · l leave · esc close",
|
||||||
theme::dim(),
|
theme::dim(),
|
||||||
))
|
))
|
||||||
.alignment(Alignment::Center),
|
.alignment(Alignment::Center),
|
||||||
|
|||||||
+14
-12
@@ -4,6 +4,7 @@ use crate::app::state::{AppState, DevicePlaybackRole};
|
|||||||
|
|
||||||
pub const ACCENT: Color = Color::Cyan;
|
pub const ACCENT: Color = Color::Cyan;
|
||||||
pub const CONTROL_ACCENT: Color = Color::Yellow;
|
pub const CONTROL_ACCENT: Color = Color::Yellow;
|
||||||
|
pub const JAM_ACCENT: Color = Color::Magenta;
|
||||||
pub const DIM: Color = Color::DarkGray;
|
pub const DIM: Color = Color::DarkGray;
|
||||||
|
|
||||||
pub fn accent() -> Style {
|
pub fn accent() -> Style {
|
||||||
@@ -37,10 +38,10 @@ pub fn selection() -> Style {
|
|||||||
}
|
}
|
||||||
|
|
||||||
pub fn selection_for(state: &AppState) -> Style {
|
pub fn selection_for(state: &AppState) -> Style {
|
||||||
if state.device_playback.is_control() {
|
match state.device_playback.role {
|
||||||
Style::new().fg(Color::White).bg(Color::Rgb(92, 72, 0))
|
DevicePlaybackRole::Jam => Style::new().fg(Color::White).bg(Color::Rgb(80, 24, 96)),
|
||||||
} else {
|
DevicePlaybackRole::Control => Style::new().fg(Color::White).bg(Color::Rgb(92, 72, 0)),
|
||||||
selection()
|
DevicePlaybackRole::Active => selection(),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -49,10 +50,10 @@ pub fn header_for(state: &AppState) -> Style {
|
|||||||
}
|
}
|
||||||
|
|
||||||
pub fn border_for(state: &AppState) -> Style {
|
pub fn border_for(state: &AppState) -> Style {
|
||||||
if state.device_playback.is_control() {
|
match state.device_playback.role {
|
||||||
Style::new().fg(CONTROL_ACCENT)
|
DevicePlaybackRole::Jam => Style::new().fg(JAM_ACCENT),
|
||||||
} else {
|
DevicePlaybackRole::Control => Style::new().fg(CONTROL_ACCENT),
|
||||||
dim()
|
DevicePlaybackRole::Active => dim(),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -64,6 +65,7 @@ pub fn role_pill(role: DevicePlaybackRole) -> Style {
|
|||||||
let bg = match role {
|
let bg = match role {
|
||||||
DevicePlaybackRole::Active => Color::Green,
|
DevicePlaybackRole::Active => Color::Green,
|
||||||
DevicePlaybackRole::Control => CONTROL_ACCENT,
|
DevicePlaybackRole::Control => CONTROL_ACCENT,
|
||||||
|
DevicePlaybackRole::Jam => JAM_ACCENT,
|
||||||
};
|
};
|
||||||
Style::new()
|
Style::new()
|
||||||
.fg(Color::Black)
|
.fg(Color::Black)
|
||||||
@@ -72,9 +74,9 @@ pub fn role_pill(role: DevicePlaybackRole) -> Style {
|
|||||||
}
|
}
|
||||||
|
|
||||||
fn accent_color_for(state: &AppState) -> Color {
|
fn accent_color_for(state: &AppState) -> Color {
|
||||||
if state.device_playback.is_control() {
|
match state.device_playback.role {
|
||||||
CONTROL_ACCENT
|
DevicePlaybackRole::Jam => JAM_ACCENT,
|
||||||
} else {
|
DevicePlaybackRole::Control => CONTROL_ACCENT,
|
||||||
ACCENT
|
DevicePlaybackRole::Active => ACCENT,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user