use std::sync::atomic::AtomicU64;
use std::sync::Arc;
use std::time::{Duration, Instant};
use arete_sdk::Frame;
use super::app::App;
const LIVE_WINDOW: Duration = Duration::from_secs(2);
pub struct Tab {
pub title: String,
subscription_id: Option<String>,
view: String,
pub app: App,
last_frame: Option<Instant>,
}
pub struct Tabs {
pub tabs: Vec<Tab>,
pub active: usize,
}
pub struct TabFeed {
pub subscription_id: String,
pub view: String,
pub label: String,
}
impl Tabs {
pub fn new(
feeds: Vec<TabFeed>,
url: &str,
dropped_frames: Arc<AtomicU64>,
snapshots: bool,
) -> Self {
let mut tabs = Vec::new();
if feeds.len() > 1 {
let stack = feeds[0]
.label
.split_once(" \u{b7} ")
.map(|(stack, _)| format!("{stack} \u{b7} "))
.unwrap_or_default();
tabs.push(Tab {
title: "All".to_string(),
subscription_id: None,
view: String::new(),
app: App::new(
format!("{stack}all views"),
url.to_string(),
Arc::clone(&dropped_frames),
)
.with_snapshots(false),
last_frame: None,
});
}
for feed in feeds {
let view_id = feed.view.clone();
tabs.push(Tab {
title: feed.view.clone(),
subscription_id: Some(feed.subscription_id),
view: feed.view,
app: App::new(feed.label, url.to_string(), Arc::clone(&dropped_frames))
.with_snapshots(snapshots)
.with_source_view(view_id),
last_frame: None,
});
}
Self { tabs, active: 0 }
}
pub fn len(&self) -> usize {
self.tabs.len()
}
pub fn current(&mut self) -> &mut App {
&mut self.tabs[self.active].app
}
pub fn apps(&mut self) -> impl Iterator<Item = &mut App> {
self.tabs.iter_mut().map(|tab| &mut tab.app)
}
pub fn next(&mut self) {
self.active = (self.active + 1) % self.tabs.len();
}
pub fn prev(&mut self) {
self.active = (self.active + self.tabs.len() - 1) % self.tabs.len();
}
pub fn select(&mut self, index: usize) {
if index < self.tabs.len() {
self.active = index;
}
}
pub fn is_live(&self, index: usize, now: Instant) -> bool {
self.tabs[index]
.last_frame
.is_some_and(|at| now.duration_since(at) < LIVE_WINDOW)
}
pub fn apply(&mut self, connection: u64, frame: Frame) {
let now = Instant::now();
if self.tabs.len() == 1 {
self.tabs[0].last_frame = Some(now);
self.tabs[0].app.apply_connection_frame(connection, frame);
return;
}
let id = frame.subscription_id().to_string();
let Some(index) = self
.tabs
.iter()
.position(|tab| tab.subscription_id.as_deref() == Some(id.as_str()))
else {
return;
};
let merged = for_all_tab(&self.tabs[index].view, &frame);
let tab = &mut self.tabs[index];
tab.last_frame = Some(now);
tab.app.apply_connection_frame(connection, frame);
if !merged.is_empty() {
let all = &mut self.tabs[0];
all.last_frame = Some(now);
for frame in merged {
all.app.apply_connection_frame(connection, frame);
}
}
}
}
pub fn for_all_tab(view: &str, frame: &Frame) -> Vec<Frame> {
let relabel = |key: &str| format!("{view} \u{b7} {key}");
match frame.clone() {
Frame::Snapshot {
protocol_version,
subscription_id,
mode,
entity,
data,
..
} => data
.into_iter()
.map(|row| Frame::Upsert {
protocol_version,
subscription_id: subscription_id.clone(),
mode,
entity: entity.clone(),
key: relabel(&row.key),
data: row.data,
append: Vec::new(),
seq: None,
offset: None,
})
.collect(),
Frame::Subscribed { .. } | Frame::Unsubscribed { .. } => Vec::new(),
mut frame => {
if let Frame::Upsert { key, .. }
| Frame::Patch { key, .. }
| Frame::Remove { key, .. }
| Frame::Delete { key, .. } = &mut frame
{
*key = relabel(key);
}
vec![frame]
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use arete_sdk::{Mode, SnapshotEntity};
fn feeds() -> Vec<TabFeed> {
["A/list", "B/list"]
.into_iter()
.map(|view| TabFeed {
subscription_id: format!("sub:{view}"),
view: view.to_string(),
label: format!("tip-jar \u{b7} {view}"),
})
.collect()
}
fn tabs() -> Tabs {
Tabs::new(
feeds(),
"ws://localhost/",
Arc::new(AtomicU64::new(0)),
true,
)
}
fn upsert(view: &str, key: &str) -> Frame {
Frame::Upsert {
protocol_version: 2,
subscription_id: format!("sub:{view}"),
mode: Mode::List,
entity: view.to_string(),
key: key.to_string(),
data: serde_json::json!({ "id": key }),
append: Vec::new(),
seq: None,
offset: None,
}
}
fn snapshot(view: &str, keys: &[&str]) -> Frame {
Frame::Snapshot {
protocol_version: 2,
subscription_id: format!("sub:{view}"),
snapshot_id: "s".to_string(),
authoritative: true,
mode: Mode::List,
entity: view.to_string(),
key: None,
data: keys
.iter()
.map(|key| SnapshotEntity {
key: key.to_string(),
data: serde_json::json!({ "id": key }),
})
.collect(),
complete: true,
}
}
#[test]
fn several_views_get_an_all_tab_first() {
let tabs = tabs();
let titles: Vec<&str> = tabs.tabs.iter().map(|tab| tab.title.as_str()).collect();
assert_eq!(titles, ["All", "A/list", "B/list"]);
assert_eq!(tabs.tabs[0].app.view, "tip-jar \u{b7} all views");
assert_eq!(tabs.tabs[1].app.view, "tip-jar \u{b7} A/list");
let single = Tabs::new(
feeds().into_iter().take(1).collect(),
"ws://localhost/",
Arc::new(AtomicU64::new(0)),
true,
);
assert_eq!(single.len(), 1);
}
#[test]
fn frames_reach_their_tab_and_the_all_tab() {
let mut tabs = tabs();
tabs.apply(0, snapshot("A/list", &["1"]));
tabs.apply(0, upsert("B/list", "1"));
tabs.apply(0, upsert("C/list", "9"));
assert_eq!(tabs.tabs[1].app.entity_keys, ["1"]);
assert_eq!(tabs.tabs[2].app.entity_keys, ["1"]);
assert_eq!(
tabs.tabs[0].app.entity_keys,
["A/list \u{b7} 1", "B/list \u{b7} 1"]
);
tabs.apply(0, snapshot("B/list", &["2"]));
assert_eq!(tabs.tabs[2].app.entity_keys, ["2"]);
assert!(tabs.tabs[0]
.app
.entity_keys
.contains(&"A/list \u{b7} 1".to_string()));
}
#[test]
fn tabs_cycle_and_select_by_number() {
let mut tabs = tabs();
tabs.next();
assert_eq!(tabs.active, 1);
tabs.prev();
tabs.prev();
assert_eq!(tabs.active, 2);
tabs.select(0);
assert_eq!(tabs.active, 0);
tabs.select(7);
assert_eq!(tabs.active, 0);
}
#[test]
fn a_tab_is_live_while_frames_arrive() {
let mut tabs = tabs();
let now = Instant::now();
assert!(!tabs.is_live(1, now));
tabs.apply(0, upsert("A/list", "1"));
let now = Instant::now();
assert!(tabs.is_live(0, now) && tabs.is_live(1, now));
assert!(!tabs.is_live(2, now));
assert!(!tabs.is_live(1, now + LIVE_WINDOW));
}
}