spotify_cli/rpc/
events.rs

1//! Event polling and broadcasting
2//!
3//! Polls Spotify for playback state changes and broadcasts events to subscribers.
4
5use std::time::Duration;
6
7use serde_json::Value;
8use tokio::sync::broadcast;
9use tokio::time::interval;
10use tracing::{debug, trace, warn};
11
12use crate::cli::commands::get_authenticated_client;
13use crate::endpoints::player::get_playback_state;
14
15use super::protocol::RpcNotification;
16
17/// Event types that clients can subscribe to
18#[derive(Debug, Clone, PartialEq, Eq)]
19pub enum EventType {
20    TrackChanged,
21    PlaybackStateChanged,
22    VolumeChanged,
23    ShuffleChanged,
24    RepeatChanged,
25    DeviceChanged,
26}
27
28/// Playback state snapshot for change detection
29#[derive(Debug, Clone, Default)]
30struct PlaybackSnapshot {
31    track_id: Option<String>,
32    is_playing: bool,
33    volume: u8,
34    shuffle: bool,
35    repeat: String,
36    device_id: Option<String>,
37}
38
39impl PlaybackSnapshot {
40    fn from_json(state: &Value) -> Self {
41        Self {
42            track_id: state
43                .get("item")
44                .and_then(|i| i.get("id"))
45                .and_then(|v| v.as_str())
46                .map(String::from),
47            is_playing: state
48                .get("is_playing")
49                .and_then(|v| v.as_bool())
50                .unwrap_or(false),
51            volume: state
52                .get("device")
53                .and_then(|d| d.get("volume_percent"))
54                .and_then(|v| v.as_u64())
55                .unwrap_or(0) as u8,
56            shuffle: state
57                .get("shuffle_state")
58                .and_then(|v| v.as_bool())
59                .unwrap_or(false),
60            repeat: state
61                .get("repeat_state")
62                .and_then(|v| v.as_str())
63                .unwrap_or("off")
64                .to_string(),
65            device_id: state
66                .get("device")
67                .and_then(|d| d.get("id"))
68                .and_then(|v| v.as_str())
69                .map(String::from),
70        }
71    }
72}
73
74/// Event poller that monitors Spotify playback state
75pub struct EventPoller {
76    event_tx: broadcast::Sender<RpcNotification>,
77    poll_interval: Duration,
78}
79
80impl EventPoller {
81    pub fn new(event_tx: broadcast::Sender<RpcNotification>) -> Self {
82        Self {
83            event_tx,
84            poll_interval: Duration::from_secs(2),
85        }
86    }
87
88    /// Set the polling interval
89    pub fn with_interval(mut self, interval: Duration) -> Self {
90        self.poll_interval = interval;
91        self
92    }
93
94    /// Run the event polling loop
95    pub async fn run(&self) {
96        let mut interval = interval(self.poll_interval);
97        let mut last_state = PlaybackSnapshot::default();
98
99        loop {
100            interval.tick().await;
101
102            match self.poll_playback_state().await {
103                Some(current) => {
104                    self.detect_and_broadcast_changes(&last_state, &current)
105                        .await;
106                    last_state = current;
107                }
108                None => {
109                    trace!("No playback state available");
110                }
111            }
112        }
113    }
114
115    async fn poll_playback_state(&self) -> Option<PlaybackSnapshot> {
116        let client = match get_authenticated_client().await {
117            Ok(c) => c,
118            Err(_) => {
119                trace!("Not authenticated, skipping poll");
120                return None;
121            }
122        };
123
124        match get_playback_state::get_playback_state(&client).await {
125            Ok(Some(state)) => Some(PlaybackSnapshot::from_json(&state)),
126            Ok(None) => None,
127            Err(e) => {
128                warn!(error = %e, "Failed to poll playback state");
129                None
130            }
131        }
132    }
133
134    async fn detect_and_broadcast_changes(&self, old: &PlaybackSnapshot, new: &PlaybackSnapshot) {
135        // Track changed
136        if old.track_id != new.track_id {
137            debug!(old = ?old.track_id, new = ?new.track_id, "Track changed");
138            self.broadcast(
139                "event.trackChanged",
140                serde_json::json!({
141                    "track_id": new.track_id,
142                }),
143            );
144        }
145
146        // Playback state changed (play/pause)
147        if old.is_playing != new.is_playing {
148            debug!(is_playing = new.is_playing, "Playback state changed");
149            self.broadcast(
150                "event.playbackStateChanged",
151                serde_json::json!({
152                    "is_playing": new.is_playing,
153                }),
154            );
155        }
156
157        // Volume changed
158        if old.volume != new.volume {
159            debug!(volume = new.volume, "Volume changed");
160            self.broadcast(
161                "event.volumeChanged",
162                serde_json::json!({
163                    "volume": new.volume,
164                }),
165            );
166        }
167
168        // Shuffle changed
169        if old.shuffle != new.shuffle {
170            debug!(shuffle = new.shuffle, "Shuffle changed");
171            self.broadcast(
172                "event.shuffleChanged",
173                serde_json::json!({
174                    "shuffle": new.shuffle,
175                }),
176            );
177        }
178
179        // Repeat changed
180        if old.repeat != new.repeat {
181            debug!(repeat = %new.repeat, "Repeat changed");
182            self.broadcast(
183                "event.repeatChanged",
184                serde_json::json!({
185                    "repeat": new.repeat,
186                }),
187            );
188        }
189
190        // Device changed
191        if old.device_id != new.device_id {
192            debug!(device = ?new.device_id, "Device changed");
193            self.broadcast(
194                "event.deviceChanged",
195                serde_json::json!({
196                    "device_id": new.device_id,
197                }),
198            );
199        }
200    }
201
202    fn broadcast(&self, method: &str, params: Value) {
203        let notification = RpcNotification::new(method, Some(params));
204        // Ignore send errors - no subscribers is fine
205        let _ = self.event_tx.send(notification);
206    }
207}
spotify_cli/rpc/events.rs

spotify_cli/rpc/
events.rs
