From 8679a23692d249e3069b6e253a7b7e13a9859766 Mon Sep 17 00:00:00 2001 From: Adrien Date: Sat, 23 Dec 2023 14:54:21 +0100 Subject: [PATCH] =?UTF-8?q?=E2=9C=A8=20Add=20rooms=20topics=20management?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/base.rs | 32 +++++++--- src/matrix_interface/client.rs | 99 ++++++++++++++++++++----------- src/matrix_interface/requester.rs | 2 + 3 files changed, 89 insertions(+), 44 deletions(-) diff --git a/src/base.rs b/src/base.rs index 4a1cec0..d1b84e4 100644 --- a/src/base.rs +++ b/src/base.rs @@ -16,7 +16,7 @@ use matrix_sdk::{ }; use tracing::{debug, error, warn}; -use crate::matrix_interface::client::Client; +use crate::matrix_interface::client::{Client, RoomTopicEvent}; use crate::matrix_interface::requester::Requester; use crate::matrix_interface::worker_tasks::LoginStyle; @@ -148,24 +148,38 @@ async fn on_room(room_option: Option, rooms_ref: &UseAtomRef) { } } +async fn on_room_topic( + room_topic_event_option: Option, + rooms_ref: &UseAtomRef, +) { + if let Some(room_topic_event) = room_topic_event_option { + let room_id = room_topic_event.0; + + if let Some(room_ref) = rooms_ref.read().get(&room_id) { + let topic = room_topic_event.1; + + let mut room = room_ref.borrow_mut(); + room.topic = Some(RefCell::new(topic)); + } else { + warn!("No room found with the \"{}\" id", room_id); + } + } +} + pub async fn sync_rooms( mut rx: UnboundedReceiver, app_settings_ref: UseAtomRef, rooms_ref: UseAtomRef, ) { - error!("=== SYNC ROOMS BEG ==="); while let Some(_is_logged) = rx.next().await { - let app_settings_ref = app_settings_ref.read(); - let requester = &app_settings_ref.requester; - - if requester.is_some() { - let rooms_receiver = &requester.as_ref().unwrap().rooms_receiver; - - let mut room_stream = rooms_receiver.stream(); + if let Some(requester) = &app_settings_ref.read().requester { + let mut room_stream = requester.rooms_receiver.stream(); + let mut room_topic_stream = requester.room_topic_receiver.stream(); loop { select! { room = room_stream.next() => on_room(room, &rooms_ref).await, + room_topic_event = room_topic_stream.next() => on_room_topic(room_topic_event, &rooms_ref).await, } } } diff --git a/src/matrix_interface/client.rs b/src/matrix_interface/client.rs index b64779f..bcd205d 100644 --- a/src/matrix_interface/client.rs +++ b/src/matrix_interface/client.rs @@ -1,4 +1,5 @@ // TODO: make a choice: mpsc vs flume. +use std::cell::RefCell; use std::sync::Arc; use std::time::Duration; @@ -13,24 +14,27 @@ use matrix_sdk::{ config::SyncSettings, event_handler::Ctx, room::Room as MatrixRoom, - ruma::events::{ - key::verification::{ - done::{OriginalSyncKeyVerificationDoneEvent, ToDeviceKeyVerificationDoneEvent}, - key::{OriginalSyncKeyVerificationKeyEvent, ToDeviceKeyVerificationKeyEvent}, - request::ToDeviceKeyVerificationRequestEvent, - start::{OriginalSyncKeyVerificationStartEvent, ToDeviceKeyVerificationStartEvent}, + ruma::{ + events::{ + key::verification::{ + done::{OriginalSyncKeyVerificationDoneEvent, ToDeviceKeyVerificationDoneEvent}, + key::{OriginalSyncKeyVerificationKeyEvent, ToDeviceKeyVerificationKeyEvent}, + request::ToDeviceKeyVerificationRequestEvent, + start::{OriginalSyncKeyVerificationStartEvent, ToDeviceKeyVerificationStartEvent}, + }, + presence::PresenceEvent, + reaction::ReactionEventContent, + room::{ + member::{OriginalSyncRoomMemberEvent, RoomMemberEventContent}, + message::RoomMessageEventContent, + name::RoomNameEventContent, + redaction::OriginalSyncRoomRedactionEvent, + topic::RoomTopicEventContent, + }, + typing::SyncTypingEvent, + SyncMessageLikeEvent, SyncStateEvent, }, - presence::PresenceEvent, - reaction::ReactionEventContent, - room::{ - member::{OriginalSyncRoomMemberEvent, RoomMemberEventContent}, - message::RoomMessageEventContent, - name::RoomNameEventContent, - redaction::OriginalSyncRoomRedactionEvent, - topic::RoomTopicEventContent, - }, - typing::SyncTypingEvent, - SyncMessageLikeEvent, SyncStateEvent, + OwnedRoomId, }, Client as MatrixClient, }; @@ -45,14 +49,23 @@ pub enum ClientError { Matrix(#[from] matrix_sdk::Error), } +pub struct RoomTopicEvent(pub OwnedRoomId, pub String); + #[derive(Clone)] struct Senders { rooms_sender: flume::Sender, + room_topic_sender: flume::Sender, } impl Senders { - fn new(rooms_sender: flume::Sender) -> Self { - Self { rooms_sender } + fn new( + rooms_sender: flume::Sender, + room_topic_sender: flume::Sender, + ) -> Self { + Self { + rooms_sender, + room_topic_sender, + } } } @@ -65,13 +78,17 @@ pub struct Client { } impl Client { - pub fn new(client: Arc, rooms_sender: flume::Sender) -> Self { + pub fn new( + client: Arc, + rooms_sender: flume::Sender, + room_topic_sender: flume::Sender, + ) -> Self { Self { initialized: false, client: Some(client), load_handle: None, sync_handle: None, - senders: Senders::new(rooms_sender), + senders: Senders::new(rooms_sender, room_topic_sender), } } @@ -91,21 +108,25 @@ impl Client { dbg!(_ev); } - async fn on_room_topic_event(ev: SyncStateEvent, room: MatrixRoom) { + async fn on_room_topic_event( + ev: SyncStateEvent, + room: MatrixRoom, + senders: Ctx, + ) { debug!("== on_room_topic_event =="); dbg!(&ev); - // if let SyncStateEvent::Original(ev) = ev { - // let room_id = room.room_id().to_owned(); - // let store = reactive_store.read().unwrap().to_owned(); - // if let Some(store_room) = store.rooms.get(&room_id) { - // // store_room.write().unwrap().topic = Some(ev.content.topic); - // // let _ = reactive_store.write(store); - // println!("HOP"); - // } else { - // println!("No room with \"{room_id}\" id known"); - // } - // } + if let SyncStateEvent::Original(ev) = ev { + let room_topic_sender = &senders.room_topic_sender; + let room_id = room.room_id(); + + if let Err(err) = room_topic_sender + .send_async(RoomTopicEvent(room_id.to_owned(), ev.content.topic)) + .await + { + error!("Unable to publish the \"{}\" new topic: {}", room_id, err); + } + } } async fn on_room_member_event(ev: SyncStateEvent, _room: MatrixRoom) { @@ -221,7 +242,9 @@ impl Client { pub async fn spawn(homeserver_url: String) -> Requester { let (tx, rx) = unbounded_channel::(); + let (rooms_sender, rooms_receiver) = unbounded::(); + let (room_topic_sender, room_topic_receiver) = unbounded::(); let matrix_client = Arc::new( MatrixClient::builder() @@ -231,7 +254,7 @@ impl Client { .unwrap(), ); - let mut client = Client::new(matrix_client.clone(), rooms_sender); + let mut client = Client::new(matrix_client.clone(), rooms_sender, room_topic_sender); tokio::spawn({ async move { @@ -243,6 +266,7 @@ impl Client { matrix_client, tx, rooms_receiver, + room_topic_receiver, } } @@ -277,9 +301,14 @@ impl Client { for matrix_rooms in vec![joined_matrix_rooms_ref, invited_matrix_rooms_ref] { for matrix_room in matrix_rooms.iter() { + let topic = match matrix_room.topic() { + None => None, + Some(topic) => Some(RefCell::new(topic)), + }; + let room = Room::new( Arc::new(matrix_room.to_owned()), - None, + topic, matrix_room.is_direct().await.ok(), ); if let Err(err) = rooms_sender.send_async(room).await { diff --git a/src/matrix_interface/requester.rs b/src/matrix_interface/requester.rs index 5c37d45..2930a96 100644 --- a/src/matrix_interface/requester.rs +++ b/src/matrix_interface/requester.rs @@ -3,6 +3,7 @@ use std::sync::Arc; use matrix_sdk::Client as MatrixClient; use tokio::sync::mpsc::UnboundedSender; +use super::client::RoomTopicEvent; use super::worker_tasks::{oneshot, LoginStyle, WorkerTask}; use crate::base::Room; @@ -11,6 +12,7 @@ pub struct Requester { pub matrix_client: Arc, pub tx: UnboundedSender, pub rooms_receiver: flume::Receiver, + pub room_topic_receiver: flume::Receiver, } impl Requester {