feat(qconnect-app): add app adapter with pending timeout, resync and concurrency policy
-cx
This commit is contained in:
vicrodh committed
1 parent
c4ffbf4f57
commit
9959117439
7 files changed
+846
No files matched your search
@@ -0,0 +1,20 @@
|
||||
[package]
|
||||
name = "qconnect-app"
|
||||
version.workspace = true
|
||||
edition.workspace = true
|
||||
license.workspace = true
|
||||
description = "UI-agnostic Qobuz Connect application adapter"
|
||||
|
||||
[dependencies]
|
||||
qconnect-core = { path = "../qconnect-core" }
|
||||
qconnect-protocol = { path = "../qconnect-protocol" }
|
||||
qconnect-transport-ws = { path = "../qconnect-transport-ws" }
|
||||
async-trait = { workspace = true }
|
||||
tokio = { workspace = true }
|
||||
serde = { workspace = true }
|
||||
serde_json = { workspace = true }
|
||||
thiserror = { workspace = true }
|
||||
uuid = { version = "1", features = ["v4"] }
|
||||
|
||||
[dev-dependencies]
|
||||
tokio = { workspace = true, features = ["test-util"] }
|
||||
@@ -0,0 +1,740 @@
|
||||
use std::{
|
||||
sync::Arc,
|
||||
time::{Duration, SystemTime, UNIX_EPOCH},
|
||||
};
|
||||
|
||||
use qconnect_core::{
|
||||
apply_event, telemetry, PendingCorrelation, PendingQueueAction, QConnectQueueState, QueueEvent,
|
||||
QueueItem,
|
||||
};
|
||||
use qconnect_protocol::{
|
||||
build_qconnect_outbound_envelope, parse_inbound_event, InboundEnvelope, QueueCommand,
|
||||
QueueCommandType, QueueEventType, QueueServerEvent,
|
||||
};
|
||||
use qconnect_transport_ws::{TransportEvent, WsTransport, WsTransportConfig};
|
||||
use serde_json::Value;
|
||||
use tokio::sync::Mutex;
|
||||
use uuid::Uuid;
|
||||
|
||||
use crate::{QconnectAppError, QconnectAppEvent, QconnectEventSink, QconnectRuntimeState};
|
||||
|
||||
pub struct QconnectApp<TTransport, TSink>
|
||||
where
|
||||
TTransport: WsTransport,
|
||||
TSink: QconnectEventSink,
|
||||
{
|
||||
transport: Arc<TTransport>,
|
||||
sink: Arc<TSink>,
|
||||
state: Arc<Mutex<QconnectRuntimeState>>,
|
||||
}
|
||||
|
||||
impl<TTransport, TSink> Clone for QconnectApp<TTransport, TSink>
|
||||
where
|
||||
TTransport: WsTransport,
|
||||
TSink: QconnectEventSink,
|
||||
{
|
||||
fn clone(&self) -> Self {
|
||||
Self {
|
||||
transport: Arc::clone(&self.transport),
|
||||
sink: Arc::clone(&self.sink),
|
||||
state: Arc::clone(&self.state),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<TTransport, TSink> QconnectApp<TTransport, TSink>
|
||||
where
|
||||
TTransport: WsTransport + 'static,
|
||||
TSink: QconnectEventSink + 'static,
|
||||
{
|
||||
const PENDING_ACTION_TIMEOUT_MS: u64 = 10_000;
|
||||
|
||||
pub fn new(transport: Arc<TTransport>, sink: Arc<TSink>) -> Self {
|
||||
Self {
|
||||
transport,
|
||||
sink,
|
||||
state: Arc::new(Mutex::new(QconnectRuntimeState::default())),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn state_handle(&self) -> Arc<Mutex<QconnectRuntimeState>> {
|
||||
Arc::clone(&self.state)
|
||||
}
|
||||
|
||||
pub fn subscribe_transport_events(&self) -> tokio::sync::broadcast::Receiver<TransportEvent> {
|
||||
self.transport.subscribe()
|
||||
}
|
||||
|
||||
pub async fn connect(&self, config: WsTransportConfig) -> Result<(), QconnectAppError> {
|
||||
self.transport.connect(config).await?;
|
||||
{
|
||||
let mut state = self.state.lock().await;
|
||||
state.transport_connected = true;
|
||||
}
|
||||
self.sink
|
||||
.on_event(QconnectAppEvent::TransportConnected)
|
||||
.await;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn disconnect(&self) -> Result<(), QconnectAppError> {
|
||||
self.transport.disconnect().await?;
|
||||
{
|
||||
let mut state = self.state.lock().await;
|
||||
state.transport_connected = false;
|
||||
state.pending.clear();
|
||||
}
|
||||
self.sink
|
||||
.on_event(QconnectAppEvent::TransportDisconnected)
|
||||
.await;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn queue_state_snapshot(&self) -> QConnectQueueState {
|
||||
self.state.lock().await.queue.clone()
|
||||
}
|
||||
|
||||
pub async fn build_queue_command(
|
||||
&self,
|
||||
command_type: QueueCommandType,
|
||||
payload: Value,
|
||||
) -> QueueCommand {
|
||||
let version_ref = self.state.lock().await.queue.version;
|
||||
QueueCommand::new(command_type, self.next_action_uuid(), version_ref, payload)
|
||||
}
|
||||
|
||||
pub async fn send_queue_command(
|
||||
&self,
|
||||
command: QueueCommand,
|
||||
) -> Result<String, QconnectAppError> {
|
||||
let action_uuid = command.action_uuid.clone();
|
||||
let pending = PendingQueueAction {
|
||||
uuid: action_uuid.clone(),
|
||||
queue_version_ref: command.queue_version_ref,
|
||||
emit_result_event: true,
|
||||
is_ask_for_state_action: matches!(
|
||||
command.command_type,
|
||||
QueueCommandType::CtrlSrvrAskForQueueState
|
||||
),
|
||||
concurrency_error: false,
|
||||
sent_at_ms: now_ms(),
|
||||
};
|
||||
|
||||
{
|
||||
let mut state = self.state.lock().await;
|
||||
state.pending.start(pending)?;
|
||||
}
|
||||
|
||||
let envelope = build_qconnect_outbound_envelope(command)?;
|
||||
if let Err(err) = self.transport.send(envelope).await {
|
||||
let mut state = self.state.lock().await;
|
||||
state.pending.clear();
|
||||
return Err(err.into());
|
||||
}
|
||||
|
||||
self.sink
|
||||
.on_event(QconnectAppEvent::PendingActionStarted {
|
||||
uuid: action_uuid.clone(),
|
||||
})
|
||||
.await;
|
||||
self.spawn_pending_timeout_watch(action_uuid.clone());
|
||||
Ok(action_uuid)
|
||||
}
|
||||
|
||||
pub async fn handle_transport_event(
|
||||
&self,
|
||||
event: TransportEvent,
|
||||
) -> Result<(), QconnectAppError> {
|
||||
match event {
|
||||
TransportEvent::Connected => {
|
||||
{
|
||||
let mut state = self.state.lock().await;
|
||||
state.transport_connected = true;
|
||||
}
|
||||
self.sink
|
||||
.on_event(QconnectAppEvent::TransportConnected)
|
||||
.await;
|
||||
}
|
||||
TransportEvent::Disconnected => {
|
||||
{
|
||||
let mut state = self.state.lock().await;
|
||||
state.transport_connected = false;
|
||||
state.pending.clear();
|
||||
}
|
||||
self.sink
|
||||
.on_event(QconnectAppEvent::TransportDisconnected)
|
||||
.await;
|
||||
}
|
||||
TransportEvent::InboundReceived(inbound) => {
|
||||
self.handle_inbound_envelope(inbound).await?;
|
||||
}
|
||||
TransportEvent::InboundQueueServerEvent(event) => {
|
||||
self.apply_server_event(event).await?;
|
||||
}
|
||||
TransportEvent::Authenticated
|
||||
| TransportEvent::Subscribed
|
||||
| TransportEvent::ReconnectScheduled { .. }
|
||||
| TransportEvent::KeepalivePingSent
|
||||
| TransportEvent::KeepalivePongReceived
|
||||
| TransportEvent::TransportError { .. }
|
||||
| TransportEvent::InboundFrameDecoded { .. }
|
||||
| TransportEvent::InboundPayloadBytes { .. }
|
||||
| TransportEvent::OutboundSent { .. } => {}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn handle_inbound_envelope(
|
||||
&self,
|
||||
inbound: InboundEnvelope,
|
||||
) -> Result<(), QconnectAppError> {
|
||||
let event = parse_inbound_event(inbound)?;
|
||||
self.apply_server_event(event).await
|
||||
}
|
||||
|
||||
async fn apply_server_event(&self, event: QueueServerEvent) -> Result<(), QconnectAppError> {
|
||||
let mut completed_uuid: Option<String> = None;
|
||||
let mut canceled_uuid: Option<String> = None;
|
||||
let mut ignored_queue_error_uuid: Option<String> = None;
|
||||
let mut should_trigger_resync = false;
|
||||
let mut should_emit_queue_update = false;
|
||||
let remote_action_uuid = event.action_uuid.clone().unwrap_or_default();
|
||||
let snapshot: QConnectQueueState;
|
||||
|
||||
{
|
||||
let mut state = self.state.lock().await;
|
||||
match state.pending.correlate(event.action_uuid.as_deref()) {
|
||||
PendingCorrelation::Matched => {
|
||||
completed_uuid = state.pending.clear().map(|pending| pending.uuid);
|
||||
}
|
||||
PendingCorrelation::Concurrent => {
|
||||
state.pending.mark_concurrency_error();
|
||||
canceled_uuid = state.pending.clear().map(|pending| pending.uuid);
|
||||
state.concurrency_canceled_action_uuid = canceled_uuid.clone();
|
||||
should_trigger_resync = canceled_uuid.is_some();
|
||||
}
|
||||
PendingCorrelation::NoPending | PendingCorrelation::EventWithoutActionUuid => {}
|
||||
}
|
||||
|
||||
let ignore_queue_error =
|
||||
matches!(event.event_type, QueueEventType::SrvrCtrlQueueErrorMessage)
|
||||
&& event.action_uuid.is_some()
|
||||
&& state.concurrency_canceled_action_uuid.as_deref()
|
||||
== event.action_uuid.as_deref();
|
||||
|
||||
if ignore_queue_error {
|
||||
ignored_queue_error_uuid = event.action_uuid.clone();
|
||||
state.concurrency_canceled_action_uuid = None;
|
||||
} else {
|
||||
let queue_event = map_server_event(&event, &state.queue);
|
||||
let reducer_outcome = apply_event(&mut state.queue, &queue_event, now_ms());
|
||||
let _metric_name = telemetry::queue_reducer_event_name(reducer_outcome.event_name);
|
||||
should_emit_queue_update = true;
|
||||
}
|
||||
snapshot = state.queue.clone();
|
||||
}
|
||||
|
||||
if let Some(uuid) = completed_uuid {
|
||||
self.sink
|
||||
.on_event(QconnectAppEvent::PendingActionCompleted { uuid })
|
||||
.await;
|
||||
}
|
||||
|
||||
if let Some(pending_uuid) = canceled_uuid {
|
||||
self.sink
|
||||
.on_event(
|
||||
QconnectAppEvent::PendingActionCanceledByConcurrentRemoteEvent {
|
||||
pending_uuid,
|
||||
remote_action_uuid,
|
||||
},
|
||||
)
|
||||
.await;
|
||||
}
|
||||
|
||||
if let Some(action_uuid) = ignored_queue_error_uuid {
|
||||
self.sink
|
||||
.on_event(QconnectAppEvent::QueueErrorIgnoredByConcurrency { action_uuid })
|
||||
.await;
|
||||
}
|
||||
|
||||
if should_emit_queue_update {
|
||||
self.sink
|
||||
.on_event(QconnectAppEvent::QueueUpdated(snapshot))
|
||||
.await;
|
||||
}
|
||||
|
||||
if should_trigger_resync {
|
||||
self.trigger_queue_state_resync().await;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn spawn_pending_timeout_watch(&self, action_uuid: String) {
|
||||
let app = self.clone();
|
||||
tokio::spawn(async move {
|
||||
app.watch_pending_action_timeout(action_uuid).await;
|
||||
});
|
||||
}
|
||||
|
||||
async fn watch_pending_action_timeout(&self, action_uuid: String) {
|
||||
tokio::time::sleep(Duration::from_millis(Self::PENDING_ACTION_TIMEOUT_MS)).await;
|
||||
|
||||
let (timed_out, timed_out_ask_for_state) = {
|
||||
let mut state = self.state.lock().await;
|
||||
let (is_same_pending, is_ask_for_state_action) = state
|
||||
.pending
|
||||
.current()
|
||||
.map(|pending| {
|
||||
(
|
||||
pending.uuid.as_str() == action_uuid,
|
||||
pending.is_ask_for_state_action,
|
||||
)
|
||||
})
|
||||
.unwrap_or((false, false));
|
||||
if is_same_pending {
|
||||
state.pending.clear();
|
||||
(true, is_ask_for_state_action)
|
||||
} else {
|
||||
(false, false)
|
||||
}
|
||||
};
|
||||
|
||||
if !timed_out {
|
||||
return;
|
||||
}
|
||||
|
||||
self.sink
|
||||
.on_event(QconnectAppEvent::PendingActionTimedOut {
|
||||
uuid: action_uuid,
|
||||
timeout_ms: Self::PENDING_ACTION_TIMEOUT_MS,
|
||||
})
|
||||
.await;
|
||||
|
||||
if !timed_out_ask_for_state {
|
||||
self.trigger_queue_state_resync().await;
|
||||
}
|
||||
}
|
||||
|
||||
async fn trigger_queue_state_resync(&self) {
|
||||
let queue_version_ref = {
|
||||
let state = self.state.lock().await;
|
||||
if state.pending.current().is_some() {
|
||||
return;
|
||||
}
|
||||
state.queue.version
|
||||
};
|
||||
|
||||
let command = QueueCommand::new(
|
||||
QueueCommandType::CtrlSrvrAskForQueueState,
|
||||
self.next_action_uuid(),
|
||||
queue_version_ref,
|
||||
Value::Object(Default::default()),
|
||||
);
|
||||
|
||||
if self.send_queue_command(command).await.is_ok() {
|
||||
self.sink
|
||||
.on_event(QconnectAppEvent::QueueResyncTriggered)
|
||||
.await;
|
||||
}
|
||||
}
|
||||
|
||||
fn next_action_uuid(&self) -> String {
|
||||
Uuid::new_v4().to_string()
|
||||
}
|
||||
}
|
||||
|
||||
fn map_server_event(event: &QueueServerEvent, current: &QConnectQueueState) -> QueueEvent {
|
||||
let version = event
|
||||
.queue_version
|
||||
.unwrap_or_else(|| current.version.next_minor());
|
||||
match event.event_type {
|
||||
QueueEventType::SrvrCtrlQueueState => {
|
||||
let mut next = current.clone();
|
||||
next.version = version;
|
||||
next.queue_items = parse_queue_items(&event.payload, "tracks");
|
||||
next.shuffle_mode = parse_bool(&event.payload, "shuffle_mode", current.shuffle_mode);
|
||||
next.autoplay_mode = parse_bool(&event.payload, "autoplay_mode", current.autoplay_mode);
|
||||
next.autoplay_loading =
|
||||
parse_bool(&event.payload, "autoplay_loading", current.autoplay_loading);
|
||||
next.autoplay_items = parse_queue_items(&event.payload, "autoplay_tracks");
|
||||
|
||||
if next.shuffle_mode {
|
||||
let parsed_shuffle = parse_usize_list(&event.payload, "shuffled_track_indexes");
|
||||
next.shuffle_order = if !parsed_shuffle.is_empty() {
|
||||
Some(parsed_shuffle)
|
||||
} else {
|
||||
current.shuffle_order.clone()
|
||||
};
|
||||
} else {
|
||||
next.shuffle_order = None;
|
||||
}
|
||||
|
||||
QueueEvent::QueueStateReplaced {
|
||||
action_uuid: event.action_uuid.clone(),
|
||||
state: next,
|
||||
}
|
||||
}
|
||||
QueueEventType::SrvrCtrlQueueTracksAdded => QueueEvent::TracksAdded {
|
||||
action_uuid: event.action_uuid.clone(),
|
||||
version,
|
||||
tracks: parse_queue_items(&event.payload, "tracks"),
|
||||
shuffle_seed: parse_u64(&event.payload, "shuffle_seed"),
|
||||
autoplay_reset: parse_bool(&event.payload, "autoplay_reset", false),
|
||||
autoplay_loading: parse_bool(&event.payload, "autoplay_loading", false),
|
||||
},
|
||||
QueueEventType::SrvrCtrlQueueTracksLoaded => QueueEvent::TracksLoaded {
|
||||
action_uuid: event.action_uuid.clone(),
|
||||
version,
|
||||
tracks: parse_queue_items(&event.payload, "tracks"),
|
||||
queue_position: parse_u64(&event.payload, "queue_position"),
|
||||
shuffle_mode: parse_optional_bool(&event.payload, "shuffle_mode"),
|
||||
shuffle_seed: parse_u64(&event.payload, "shuffle_seed"),
|
||||
shuffle_pivot_queue_item_id: parse_u64(&event.payload, "shuffle_pivot_queue_item_id"),
|
||||
autoplay_reset: parse_bool(&event.payload, "autoplay_reset", false),
|
||||
autoplay_loading: parse_bool(&event.payload, "autoplay_loading", false),
|
||||
},
|
||||
QueueEventType::SrvrCtrlQueueTracksInserted => QueueEvent::TracksInserted {
|
||||
action_uuid: event.action_uuid.clone(),
|
||||
version,
|
||||
tracks: parse_queue_items(&event.payload, "tracks"),
|
||||
insert_after: parse_u64(&event.payload, "insert_after"),
|
||||
shuffle_seed: parse_u64(&event.payload, "shuffle_seed"),
|
||||
autoplay_reset: parse_bool(&event.payload, "autoplay_reset", false),
|
||||
autoplay_loading: parse_bool(&event.payload, "autoplay_loading", false),
|
||||
},
|
||||
QueueEventType::SrvrCtrlQueueTracksRemoved => QueueEvent::TracksRemoved {
|
||||
action_uuid: event.action_uuid.clone(),
|
||||
version,
|
||||
queue_item_ids: parse_queue_item_ids(&event.payload),
|
||||
autoplay_reset: parse_bool(&event.payload, "autoplay_reset", false),
|
||||
autoplay_loading: parse_bool(&event.payload, "autoplay_loading", false),
|
||||
},
|
||||
QueueEventType::SrvrCtrlQueueTracksReordered => QueueEvent::TracksReordered {
|
||||
action_uuid: event.action_uuid.clone(),
|
||||
version,
|
||||
queue_item_ids: parse_queue_item_ids(&event.payload),
|
||||
insert_after: parse_u64(&event.payload, "insert_after"),
|
||||
autoplay_reset: parse_bool(&event.payload, "autoplay_reset", false),
|
||||
autoplay_loading: parse_bool(&event.payload, "autoplay_loading", false),
|
||||
},
|
||||
QueueEventType::SrvrCtrlQueueCleared => QueueEvent::QueueCleared {
|
||||
action_uuid: event.action_uuid.clone(),
|
||||
version,
|
||||
},
|
||||
QueueEventType::SrvrCtrlShuffleModeSet => QueueEvent::ShuffleModeSet {
|
||||
action_uuid: event.action_uuid.clone(),
|
||||
version,
|
||||
shuffle_mode: parse_bool(&event.payload, "shuffle_mode", false),
|
||||
shuffle_seed: parse_u64(&event.payload, "shuffle_seed"),
|
||||
shuffle_pivot_queue_item_id: parse_u64(&event.payload, "shuffle_pivot_queue_item_id"),
|
||||
autoplay_reset: parse_bool(&event.payload, "autoplay_reset", false),
|
||||
autoplay_loading: parse_bool(&event.payload, "autoplay_loading", false),
|
||||
},
|
||||
QueueEventType::SrvrCtrlAutoplayModeSet => QueueEvent::AutoplayModeSet {
|
||||
action_uuid: event.action_uuid.clone(),
|
||||
version,
|
||||
autoplay_mode: parse_bool(&event.payload, "autoplay_mode", false),
|
||||
autoplay_reset: parse_bool(&event.payload, "autoplay_reset", false),
|
||||
autoplay_loading: parse_bool(&event.payload, "autoplay_loading", false),
|
||||
},
|
||||
QueueEventType::SrvrCtrlAutoplayTracksLoaded => QueueEvent::AutoplayTracksLoaded {
|
||||
action_uuid: event.action_uuid.clone(),
|
||||
version,
|
||||
tracks: parse_queue_items(&event.payload, "tracks"),
|
||||
},
|
||||
QueueEventType::SrvrCtrlAutoplayTracksRemoved => QueueEvent::AutoplayTracksRemoved {
|
||||
action_uuid: event.action_uuid.clone(),
|
||||
version,
|
||||
queue_item_ids: parse_queue_item_ids(&event.payload),
|
||||
},
|
||||
QueueEventType::SrvrCtrlQueueTracksAddedFromAutoplay => QueueEvent::TracksAdded {
|
||||
action_uuid: event.action_uuid.clone(),
|
||||
version,
|
||||
tracks: Vec::new(),
|
||||
shuffle_seed: None,
|
||||
autoplay_reset: false,
|
||||
autoplay_loading: false,
|
||||
},
|
||||
QueueEventType::SrvrCtrlQueueErrorMessage => QueueEvent::QueueError {
|
||||
action_uuid: event.action_uuid.clone(),
|
||||
version: Some(version),
|
||||
code: event
|
||||
.payload
|
||||
.get("error_code")
|
||||
.map(value_to_string)
|
||||
.unwrap_or_else(|| "remote_error".to_string()),
|
||||
message: event
|
||||
.payload
|
||||
.get("error_message")
|
||||
.map(value_to_string)
|
||||
.unwrap_or_else(|| "queue_error_message".to_string()),
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
fn parse_queue_items(payload: &Value, field: &str) -> Vec<QueueItem> {
|
||||
payload
|
||||
.get(field)
|
||||
.and_then(Value::as_array)
|
||||
.map(|items| items.iter().filter_map(parse_queue_item).collect())
|
||||
.unwrap_or_default()
|
||||
}
|
||||
|
||||
fn parse_queue_item(value: &Value) -> Option<QueueItem> {
|
||||
let track_context_uuid = value
|
||||
.get("track_context_uuid")
|
||||
.or_else(|| value.get("context_uuid"))
|
||||
.and_then(Value::as_str)
|
||||
.unwrap_or_default()
|
||||
.to_string();
|
||||
|
||||
let track_id = value
|
||||
.get("track_id")
|
||||
.or_else(|| value.get("trackId"))
|
||||
.and_then(value_as_u64)?;
|
||||
|
||||
let queue_item_id = value
|
||||
.get("queue_item_id")
|
||||
.or_else(|| value.get("queueItemId"))
|
||||
.and_then(value_as_u64)
|
||||
.unwrap_or(track_id);
|
||||
|
||||
Some(QueueItem {
|
||||
track_context_uuid,
|
||||
track_id,
|
||||
queue_item_id,
|
||||
})
|
||||
}
|
||||
|
||||
fn parse_queue_item_ids(payload: &Value) -> Vec<u64> {
|
||||
payload
|
||||
.get("queue_item_ids")
|
||||
.or_else(|| payload.get("queueItemsIds"))
|
||||
.and_then(Value::as_array)
|
||||
.map(|items| items.iter().filter_map(value_as_u64).collect())
|
||||
.unwrap_or_default()
|
||||
}
|
||||
|
||||
fn parse_u64(payload: &Value, field: &str) -> Option<u64> {
|
||||
payload.get(field).and_then(value_as_u64)
|
||||
}
|
||||
|
||||
fn parse_bool(payload: &Value, field: &str, default: bool) -> bool {
|
||||
payload
|
||||
.get(field)
|
||||
.and_then(Value::as_bool)
|
||||
.unwrap_or(default)
|
||||
}
|
||||
|
||||
fn parse_optional_bool(payload: &Value, field: &str) -> Option<bool> {
|
||||
payload.get(field).and_then(Value::as_bool)
|
||||
}
|
||||
|
||||
fn parse_usize_list(payload: &Value, field: &str) -> Vec<usize> {
|
||||
payload
|
||||
.get(field)
|
||||
.and_then(Value::as_array)
|
||||
.map(|entries| {
|
||||
entries
|
||||
.iter()
|
||||
.filter_map(value_as_u64)
|
||||
.filter_map(|value| usize::try_from(value).ok())
|
||||
.collect()
|
||||
})
|
||||
.unwrap_or_default()
|
||||
}
|
||||
|
||||
fn value_as_u64(value: &Value) -> Option<u64> {
|
||||
value
|
||||
.as_u64()
|
||||
.or_else(|| value.as_i64().and_then(|entry| u64::try_from(entry).ok()))
|
||||
}
|
||||
|
||||
fn value_to_string(value: &Value) -> String {
|
||||
value
|
||||
.as_str()
|
||||
.map(ToString::to_string)
|
||||
.unwrap_or_else(|| value.to_string())
|
||||
}
|
||||
|
||||
fn now_ms() -> u64 {
|
||||
SystemTime::now()
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.unwrap_or_default()
|
||||
.as_millis() as u64
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use std::sync::Arc;
|
||||
|
||||
use async_trait::async_trait;
|
||||
use qconnect_core::QueueVersion;
|
||||
use qconnect_transport_ws::InMemoryWsTransport;
|
||||
use serde_json::json;
|
||||
use tokio::sync::Mutex;
|
||||
|
||||
use crate::{QconnectAppEvent, QconnectEventSink};
|
||||
|
||||
use super::QconnectApp;
|
||||
use qconnect_protocol::{QueueCommandType, QueueEventType, QueueServerEvent};
|
||||
use qconnect_transport_ws::WsTransportConfig;
|
||||
|
||||
#[derive(Debug, Default, Clone)]
|
||||
struct TestSink {
|
||||
events: Arc<Mutex<Vec<QconnectAppEvent>>>,
|
||||
}
|
||||
|
||||
impl TestSink {
|
||||
async fn snapshot(&self) -> Vec<QconnectAppEvent> {
|
||||
self.events.lock().await.clone()
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl QconnectEventSink for TestSink {
|
||||
async fn on_event(&self, event: QconnectAppEvent) {
|
||||
self.events.lock().await.push(event);
|
||||
}
|
||||
}
|
||||
|
||||
fn test_config() -> WsTransportConfig {
|
||||
WsTransportConfig {
|
||||
endpoint_url: "wss://example.invalid/ws".to_string(),
|
||||
subscribe_channels: vec![vec![1, 2, 3]],
|
||||
..Default::default()
|
||||
}
|
||||
}
|
||||
|
||||
async fn build_connected_app() -> (
|
||||
QconnectApp<InMemoryWsTransport, TestSink>,
|
||||
TestSink,
|
||||
Arc<InMemoryWsTransport>,
|
||||
tokio::sync::broadcast::Receiver<qconnect_transport_ws::TransportEvent>,
|
||||
) {
|
||||
let transport = Arc::new(InMemoryWsTransport::new());
|
||||
let sink = TestSink::default();
|
||||
let app = QconnectApp::new(Arc::clone(&transport), Arc::new(sink.clone()));
|
||||
let events_rx = app.subscribe_transport_events();
|
||||
app.connect(test_config()).await.expect("connect");
|
||||
(app, sink, transport, events_rx)
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn pending_timeout_triggers_resync() {
|
||||
let (app, sink, transport, _events_rx) = build_connected_app().await;
|
||||
let command = app
|
||||
.build_queue_command(
|
||||
QueueCommandType::CtrlSrvrQueueAddTracks,
|
||||
json!({"track_ids":[101]}),
|
||||
)
|
||||
.await;
|
||||
app.send_queue_command(command)
|
||||
.await
|
||||
.expect("send queue command");
|
||||
|
||||
tokio::task::yield_now().await;
|
||||
tokio::time::advance(std::time::Duration::from_millis(
|
||||
QconnectApp::<InMemoryWsTransport, TestSink>::PENDING_ACTION_TIMEOUT_MS + 10,
|
||||
))
|
||||
.await;
|
||||
tokio::task::yield_now().await;
|
||||
|
||||
let events = sink.snapshot().await;
|
||||
assert!(events
|
||||
.iter()
|
||||
.any(|event| { matches!(event, QconnectAppEvent::PendingActionTimedOut { .. }) }));
|
||||
assert!(events
|
||||
.iter()
|
||||
.any(|event| matches!(event, QconnectAppEvent::QueueResyncTriggered)));
|
||||
|
||||
let sent = transport.sent_messages().await;
|
||||
assert!(
|
||||
sent.len() >= 2,
|
||||
"expected original command plus ask-for-state resync"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn concurrent_remote_event_cancels_pending_and_requests_resync() {
|
||||
let (app, sink, transport, _events_rx) = build_connected_app().await;
|
||||
let command = app
|
||||
.build_queue_command(
|
||||
QueueCommandType::CtrlSrvrQueueAddTracks,
|
||||
json!({"track_ids":[101]}),
|
||||
)
|
||||
.await;
|
||||
let pending_uuid = app
|
||||
.send_queue_command(command)
|
||||
.await
|
||||
.expect("send queue command");
|
||||
|
||||
let remote_event = QueueServerEvent {
|
||||
event_type: QueueEventType::SrvrCtrlQueueTracksAdded,
|
||||
action_uuid: Some("0f892e1a-a2f4-4d18-82c6-31e8daf2ea0f".to_string()),
|
||||
queue_version: Some(QueueVersion::new(1, 1)),
|
||||
payload: json!({"tracks":[]}),
|
||||
};
|
||||
|
||||
app.apply_server_event(remote_event)
|
||||
.await
|
||||
.expect("apply concurrent event");
|
||||
|
||||
let events = sink.snapshot().await;
|
||||
assert!(events.iter().any(|event| {
|
||||
matches!(
|
||||
event,
|
||||
QconnectAppEvent::PendingActionCanceledByConcurrentRemoteEvent { pending_uuid: id, .. } if id == &pending_uuid
|
||||
)
|
||||
}));
|
||||
assert!(events
|
||||
.iter()
|
||||
.any(|event| matches!(event, QconnectAppEvent::QueueResyncTriggered)));
|
||||
|
||||
let sent = transport.sent_messages().await;
|
||||
assert!(
|
||||
sent.len() >= 2,
|
||||
"expected original command plus ask-for-state resync"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn late_queue_error_for_concurrency_canceled_action_is_ignored() {
|
||||
let (app, sink, _, _events_rx) = build_connected_app().await;
|
||||
{
|
||||
let state = app.state_handle();
|
||||
let mut guard = state.lock().await;
|
||||
guard.concurrency_canceled_action_uuid =
|
||||
Some("85fa0dd6-7bd6-4b3c-8f43-b8ee22e65d5e".to_string());
|
||||
}
|
||||
|
||||
let late_error = QueueServerEvent {
|
||||
event_type: QueueEventType::SrvrCtrlQueueErrorMessage,
|
||||
action_uuid: Some("85fa0dd6-7bd6-4b3c-8f43-b8ee22e65d5e".to_string()),
|
||||
queue_version: Some(QueueVersion::new(2, 1)),
|
||||
payload: json!({
|
||||
"error_code": "409",
|
||||
"error_message": "stale_queue_version"
|
||||
}),
|
||||
};
|
||||
|
||||
app.apply_server_event(late_error)
|
||||
.await
|
||||
.expect("apply late queue error");
|
||||
|
||||
let events = sink.snapshot().await;
|
||||
assert!(events.iter().any(|event| {
|
||||
matches!(
|
||||
event,
|
||||
QconnectAppEvent::QueueErrorIgnoredByConcurrency { action_uuid }
|
||||
if action_uuid == "85fa0dd6-7bd6-4b3c-8f43-b8ee22e65d5e"
|
||||
)
|
||||
}));
|
||||
assert!(
|
||||
!events
|
||||
.iter()
|
||||
.any(|event| matches!(event, QconnectAppEvent::QueueUpdated(_))),
|
||||
"ignored late queue error should not mutate queue"
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,14 @@
|
||||
use qconnect_core::PendingActionError;
|
||||
use qconnect_protocol::ProtocolError;
|
||||
use qconnect_transport_ws::WsTransportError;
|
||||
use thiserror::Error;
|
||||
|
||||
#[derive(Debug, Error)]
|
||||
pub enum QconnectAppError {
|
||||
#[error(transparent)]
|
||||
Pending(#[from] PendingActionError),
|
||||
#[error(transparent)]
|
||||
Protocol(#[from] ProtocolError),
|
||||
#[error(transparent)]
|
||||
Transport(#[from] WsTransportError),
|
||||
}
|
||||
@@ -0,0 +1,41 @@
|
||||
use async_trait::async_trait;
|
||||
use qconnect_core::QConnectQueueState;
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub enum QconnectAppEvent {
|
||||
TransportConnected,
|
||||
TransportDisconnected,
|
||||
QueueUpdated(QConnectQueueState),
|
||||
PendingActionStarted {
|
||||
uuid: String,
|
||||
},
|
||||
PendingActionCompleted {
|
||||
uuid: String,
|
||||
},
|
||||
PendingActionTimedOut {
|
||||
uuid: String,
|
||||
timeout_ms: u64,
|
||||
},
|
||||
PendingActionCanceledByConcurrentRemoteEvent {
|
||||
pending_uuid: String,
|
||||
remote_action_uuid: String,
|
||||
},
|
||||
QueueErrorIgnoredByConcurrency {
|
||||
action_uuid: String,
|
||||
},
|
||||
QueueResyncTriggered,
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
pub trait QconnectEventSink: Send + Sync {
|
||||
async fn on_event(&self, event: QconnectAppEvent);
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Default)]
|
||||
pub struct NoOpEventSink;
|
||||
|
||||
#[async_trait]
|
||||
impl QconnectEventSink for NoOpEventSink {
|
||||
async fn on_event(&self, _event: QconnectAppEvent) {}
|
||||
}
|
||||
@@ -0,0 +1,4 @@
|
||||
pub const QBZ_QCONNECT_TRANSPORT: &str = "QBZ_QCONNECT_TRANSPORT";
|
||||
pub const QBZ_QCONNECT_QUEUE_MODEL: &str = "QBZ_QCONNECT_QUEUE_MODEL";
|
||||
pub const QBZ_QCONNECT_STRICT_DOMAIN_ISOLATION: &str = "QBZ_QCONNECT_STRICT_DOMAIN_ISOLATION";
|
||||
pub const QBZ_QCONNECT_PANEL_SWITCH: &str = "QBZ_QCONNECT_PANEL_SWITCH";
|
||||
@@ -0,0 +1,18 @@
|
||||
//! qconnect-app
|
||||
//!
|
||||
//! Application adapter that composes qconnect core + protocol + transport.
|
||||
|
||||
mod app;
|
||||
mod error;
|
||||
mod events;
|
||||
mod feature_flags;
|
||||
mod state;
|
||||
|
||||
pub use app::QconnectApp;
|
||||
pub use error::QconnectAppError;
|
||||
pub use events::{NoOpEventSink, QconnectAppEvent, QconnectEventSink};
|
||||
pub use feature_flags::{
|
||||
QBZ_QCONNECT_PANEL_SWITCH, QBZ_QCONNECT_QUEUE_MODEL, QBZ_QCONNECT_STRICT_DOMAIN_ISOLATION,
|
||||
QBZ_QCONNECT_TRANSPORT,
|
||||
};
|
||||
pub use state::QconnectRuntimeState;
|
||||
@@ -0,0 +1,9 @@
|
||||
use qconnect_core::{PendingActionSlot, QConnectQueueState};
|
||||
|
||||
#[derive(Debug, Default)]
|
||||
pub struct QconnectRuntimeState {
|
||||
pub queue: QConnectQueueState,
|
||||
pub pending: PendingActionSlot,
|
||||
pub transport_connected: bool,
|
||||
pub concurrency_canceled_action_uuid: Option<String>,
|
||||
}
|
||||
Reference in new issue
Block a user