117 lines
3.4 KiB
Rust
117 lines
3.4 KiB
Rust
use std::collections::HashMap;
|
|
use std::sync::Arc;
|
|
|
|
use async_trait::async_trait;
|
|
use tokio::sync::{Mutex, broadcast};
|
|
|
|
use crate::domain::PresenceUser;
|
|
use crate::ports::{HubEvent, RealtimeNotifier, Subscription};
|
|
|
|
#[derive(Debug)]
|
|
struct ConnectionInfo {
|
|
user_id: i64,
|
|
display_name: String,
|
|
}
|
|
|
|
struct Room {
|
|
sender: broadcast::Sender<HubEvent>,
|
|
connections: HashMap<String, ConnectionInfo>,
|
|
}
|
|
|
|
#[derive(Clone, Default)]
|
|
pub struct InMemoryHub {
|
|
rooms: Arc<Mutex<HashMap<i64, Room>>>,
|
|
}
|
|
|
|
#[async_trait]
|
|
impl RealtimeNotifier for InMemoryHub {
|
|
async fn join(&self, list_id: i64, user_id: i64, display_name: String) -> Subscription {
|
|
let mut rooms = self.rooms.lock().await;
|
|
let room = rooms.entry(list_id).or_insert_with(|| {
|
|
let (sender, _) = broadcast::channel(64);
|
|
Room {
|
|
sender,
|
|
connections: HashMap::new(),
|
|
}
|
|
});
|
|
|
|
let connection_id = hex::encode(crate::security::new_secret());
|
|
let already_present = room
|
|
.connections
|
|
.values()
|
|
.any(|connection| connection.user_id == user_id);
|
|
room.connections.insert(
|
|
connection_id.clone(),
|
|
ConnectionInfo {
|
|
user_id,
|
|
display_name,
|
|
},
|
|
);
|
|
let presence = current_presence(room);
|
|
|
|
if !already_present {
|
|
let _ = room.sender.send(HubEvent::PresenceChanged { list_id });
|
|
}
|
|
let receiver = room.sender.subscribe();
|
|
|
|
Subscription {
|
|
connection_id,
|
|
receiver,
|
|
presence,
|
|
}
|
|
}
|
|
|
|
async fn leave(&self, list_id: i64, connection_id: &str) {
|
|
let mut rooms = self.rooms.lock().await;
|
|
let mut remove_room = false;
|
|
if let Some(room) = rooms.get_mut(&list_id) {
|
|
let removed = room.connections.remove(connection_id);
|
|
if let Some(removed) = removed {
|
|
let still_present = room
|
|
.connections
|
|
.values()
|
|
.any(|connection| connection.user_id == removed.user_id);
|
|
if !still_present {
|
|
let _ = room.sender.send(HubEvent::PresenceChanged { list_id });
|
|
}
|
|
}
|
|
remove_room = room.connections.is_empty();
|
|
}
|
|
if remove_room {
|
|
rooms.remove(&list_id);
|
|
}
|
|
}
|
|
|
|
async fn publish_list_changed(&self, list_id: i64, revision: i64) {
|
|
let rooms = self.rooms.lock().await;
|
|
if let Some(room) = rooms.get(&list_id) {
|
|
let _ = room
|
|
.sender
|
|
.send(HubEvent::ListChanged { list_id, revision });
|
|
}
|
|
}
|
|
|
|
async fn presence(&self, list_id: i64) -> Vec<PresenceUser> {
|
|
let rooms = self.rooms.lock().await;
|
|
rooms
|
|
.get(&list_id)
|
|
.map(current_presence)
|
|
.unwrap_or_default()
|
|
}
|
|
}
|
|
|
|
fn current_presence(room: &Room) -> Vec<PresenceUser> {
|
|
let mut users = HashMap::<i64, PresenceUser>::new();
|
|
for connection in room.connections.values() {
|
|
users
|
|
.entry(connection.user_id)
|
|
.or_insert_with(|| PresenceUser {
|
|
user_id: connection.user_id,
|
|
display_name: connection.display_name.clone(),
|
|
});
|
|
}
|
|
let mut users = users.into_values().collect::<Vec<_>>();
|
|
users.sort_by_key(|user| user.display_name.to_lowercase());
|
|
users
|
|
}
|