OXYGENHUB
◆ v0.9.21 oxygenhub / crates / hub / src / live.rs
8.2 KiB · Rust · last changed Carry events between the hub's processes on an event bus 16 days ago source blame history raw
5e6d6484 Andy McCall ↶ 1
//! Live collaboration on the conflict resolver (hub spec §6.10): everyone
2
//! looking at the same conflict sees who else is there, the choices each
3
//! makes region by region (the last choice for a region wins), and the
4
//! resolution when one of them commits it. Rooms live in memory: they are
5
//! for people working together now, not a record.
33d98147 Andy McCall 6
//!
7
//! With several web processes (docs/roles.md), choices and landings travel
8
//! on the event bus ([`crate::bus`]) to every process's room for the same
9
//! conflict; who is present, and a newcomer's snapshot, are this process's
10
//! own (a room's viewers on other processes are not counted).
5e6d6484 Andy McCall ↶ 11
12
use std::collections::{BTreeMap, HashMap};
13
use std::sync::{Arc, Mutex};
14
use tokio::sync::broadcast;
15
16
/// One conflict being resolved: repository, bookmark, path.
33d98147 Andy McCall 17
#[derive(Clone, Debug, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
5e6d6484 Andy McCall ↶ 18
pub struct RoomKey {
19
    pub repo_id: i64,
20
    pub bookmark: String,
21
    pub path: String,
22
}
23
24
/// A region's current choice: `side1`…, `all`, `none` or `text`, with the
25
/// text for `text`, and who made it.
33d98147 Andy McCall 26
#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
5e6d6484 Andy McCall ↶ 27
pub struct Choice {
28
    pub choice: String,
29
    pub text: String,
30
    pub by: String,
31
}
32
33
/// What a room tells its viewers.
34
#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize)]
35
#[serde(tag = "kind", rename_all = "lowercase")]
36
pub enum LiveEvent {
37
    /// Everything so far, sent to a viewer who joins.
38
    Snapshot { present: Vec<String>, choices: BTreeMap<String, Choice>, keep: Option<Choice> },
39
    /// Who is looking now.
40
    Presence { present: Vec<String> },
41
    /// A viewer chose for region `region` (`keep` for a candidate page).
42
    Choice { region: String, choice: Choice, client: String },
43
    /// A resolution was committed; `url` is its change.
33d98147 Andy McCall 44
    Landed { url: String, by: String },
45
}
46
47
/// What one viewer does that every viewer of the room hears, in this
48
/// process and (over the bus) in the others.
49
#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
50
#[serde(tag = "kind", rename_all = "lowercase")]
51
pub enum RoomEvent {
52
    /// `client` chose for `region`.
53
    Choice { region: String, choice: Choice, client: String },
54
    /// The conflict was resolved; `url` is the change.
5e6d6484 Andy McCall ↶ 55
    Landed { url: String, by: String },
56
}
57
58
struct Room {
59
    tx: broadcast::Sender<LiveEvent>,
60
    /// Connected viewers by client id: their user names.
61
    present: HashMap<String, String>,
62
    choices: BTreeMap<String, Choice>,
63
}
64
65
impl Room {
66
    fn names(&self) -> Vec<String> {
67
        let mut v: Vec<String> = self.present.values().cloned().collect();
68
        v.sort();
69
        v.dedup();
70
        v
71
    }
72
}
73
74
#[derive(Default)]
75
pub struct Rooms {
76
    rooms: Mutex<HashMap<RoomKey, Room>>,
77
}
78
79
/// A viewer's place in a room; leaving happens when it is dropped (the
80
/// viewer's event stream closed).
81
pub struct Seat {
82
    rooms: Arc<Rooms>,
83
    key: RoomKey,
84
    client: String,
85
}
86
87
impl Drop for Seat {
88
    fn drop(&mut self) {
89
        self.rooms.leave(&self.key, &self.client);
90
    }
91
}
92
93
impl Rooms {
94
    /// Join `key` as `client` (one browser tab) of `user`: the events to
95
    /// follow, starting with a snapshot, and the seat that leaves on drop.
96
    pub fn join(&self, key: RoomKey, client: &str, user: &str) -> (broadcast::Receiver<LiveEvent>, LiveEvent) {
97
        let mut rooms = self.rooms.lock().unwrap();
98
        let room =
99
            rooms.entry(key).or_insert_with(|| Room { tx: broadcast::channel(256).0, present: HashMap::new(), choices: BTreeMap::new() });
100
        room.present.insert(client.to_string(), user.to_string());
101
        let rx = room.tx.subscribe();
102
        let present = room.names();
103
        let _ = room.tx.send(LiveEvent::Presence { present: present.clone() });
104
        let mut choices = room.choices.clone();
105
        let keep = choices.remove("keep");
106
        (rx, LiveEvent::Snapshot { present, choices, keep })
107
    }
108
109
    /// A seat for `client` in `key`, to hold while its stream is open.
110
    pub fn seat(self: &Arc<Self>, key: RoomKey, client: &str) -> Seat {
111
        Seat { rooms: Arc::clone(self), key, client: client.to_string() }
112
    }
113
114
    fn leave(&self, key: &RoomKey, client: &str) {
115
        let mut rooms = self.rooms.lock().unwrap();
116
        let Some(room) = rooms.get_mut(key) else { return };
117
        room.present.remove(client);
118
        if room.present.is_empty() {
119
            // Nobody is looking: forget the room and its choices.
120
            rooms.remove(key);
121
        } else {
122
            let _ = room.tx.send(LiveEvent::Presence { present: room.names() });
123
        }
124
    }
125
33d98147 Andy McCall 126
    /// Record and share a choice; `false` when nobody is in the room here.
5e6d6484 Andy McCall ↶ 127
    pub fn choose(&self, key: &RoomKey, client: &str, region: &str, choice: Choice) -> bool {
33d98147 Andy McCall 128
        self.apply(key, &RoomEvent::Choice { region: region.to_string(), choice, client: client.to_string() })
5e6d6484 Andy McCall ↶ 129
    }
130
131
    /// Tell a room its conflict was resolved.
132
    pub fn landed(&self, key: &RoomKey, url: &str, by: &str) {
33d98147 Andy McCall 133
        self.apply(key, &RoomEvent::Landed { url: url.to_string(), by: by.to_string() });
134
    }
135
136
    /// Apply a viewer's event to the room for `key` in this process, if
137
    /// anyone here is in it: record a choice, and tell the viewers. `false`
138
    /// when nobody here is.
139
    pub fn apply(&self, key: &RoomKey, event: &RoomEvent) -> bool {
140
        let mut rooms = self.rooms.lock().unwrap();
141
        let Some(room) = rooms.get_mut(key) else { return false };
142
        let told = match event {
143
            RoomEvent::Choice { region, choice, client } => {
144
                room.choices.insert(region.clone(), choice.clone());
145
                LiveEvent::Choice { region: region.clone(), choice: choice.clone(), client: client.clone() }
146
            }
147
            RoomEvent::Landed { url, by } => LiveEvent::Landed { url: url.clone(), by: by.clone() },
148
        };
149
        let _ = room.tx.send(told);
150
        true
5e6d6484 Andy McCall ↶ 151
    }
152
153
    /// How many rooms are open (for tests and the admin page).
154
    pub fn open(&self) -> usize {
155
        self.rooms.lock().unwrap().len()
156
    }
157
}
158
159
#[cfg(test)]
160
mod tests {
161
    use super::*;
162
163
    fn key() -> RoomKey {
164
        RoomKey { repo_id: 1, bookmark: "main".into(), path: "a.txt".into() }
165
    }
166
167
    fn pick(choice: &str, by: &str) -> Choice {
168
        Choice { choice: choice.into(), text: String::new(), by: by.into() }
169
    }
170
171
    #[test]
172
    fn viewers_see_each_other_and_each_others_choices() {
173
        let rooms = Arc::new(Rooms::default());
174
        let (mut a_rx, a_snap) = rooms.join(key(), "tab-a", "andy");
175
        assert_eq!(a_snap, LiveEvent::Snapshot { present: vec!["andy".into()], choices: BTreeMap::new(), keep: None });
176
        let seat_a = rooms.seat(key(), "tab-a");
177
        assert!(matches!(a_rx.try_recv(), Ok(LiveEvent::Presence { .. })), "a's own arrival");
178
179
        assert!(rooms.choose(&key(), "tab-a", "1", pick("side2", "andy")));
180
        assert!(matches!(a_rx.try_recv(), Ok(LiveEvent::Choice { region, .. }) if region == "1"));
181
182
        // A second viewer gets the choices so far and is announced.
183
        let (mut b_rx, b_snap) = rooms.join(key(), "tab-b", "bea");
184
        let seat_b = rooms.seat(key(), "tab-b");
185
        let LiveEvent::Snapshot { present, choices, .. } = b_snap else { panic!() };
186
        assert_eq!(present, vec!["andy", "bea"]);
187
        assert_eq!(choices["1"].choice, "side2");
188
        assert_eq!(a_rx.try_recv().unwrap(), LiveEvent::Presence { present: vec!["andy".into(), "bea".into()] });
189
190
        // The last choice for a region wins; keep is apart for candidate pages.
191
        rooms.choose(&key(), "tab-b", "1", pick("all", "bea"));
192
        rooms.choose(&key(), "tab-b", "keep", pick("1", "bea"));
193
        let _ = b_rx.try_recv();
194
        drop(seat_b);
195
        assert_eq!(a_rx.try_recv().unwrap(), LiveEvent::Choice { region: "1".into(), choice: pick("all", "bea"), client: "tab-b".into() });
196
        let _ = a_rx.try_recv();
197
        assert_eq!(a_rx.try_recv().unwrap(), LiveEvent::Presence { present: vec!["andy".into()] }, "bea left");
198
199
        rooms.landed(&key(), "/andy/demo/change/abc", "andy");
200
        assert!(matches!(a_rx.try_recv(), Ok(LiveEvent::Landed { .. })));
201
202
        // The last viewer leaving closes the room.
203
        drop(seat_a);
204
        assert_eq!(rooms.open(), 0);
205
        assert!(!rooms.choose(&key(), "tab-a", "1", pick("none", "andy")));
206
    }
207
}