ChatStore::set_receipts locks up app for bad connections (#99)

src/base.rs

@@ -666,7 +666,10 @@ impl ChatStore {
             .unwrap_or_else(|| "Untitled Matrix Room".to_string())
     }
 
-    pub async fn set_receipts(&mut self, receipts: Vec<(OwnedRoomId, Receipts)>) {
+    pub async fn set_receipts(
+        &mut self,
+        receipts: Vec<(OwnedRoomId, Receipts)>,
+    ) -> Vec<(OwnedRoomId, OwnedEventId)> {
         let mut updates = vec![];
 
         for (room_id, receipts) in receipts.into_iter() {
@@ -679,11 +682,7 @@ impl ChatStore {
             }
         }
 
-        for (room_id, read_till) in updates.into_iter() {
+        return updates;
-            if let Some(room) = self.worker.client.get_joined_room(&room_id) {
-                let _ = room.read_receipt(read_till.as_ref()).await;
-            }
-        }
     }
 
     pub fn mark_for_load(&mut self, room_id: OwnedRoomId) {

src/worker.rs

@@ -56,6 +56,7 @@ use matrix_sdk::{
         room::RoomType,
         serde::Raw,
         EventEncryptionAlgorithm,
+        OwnedEventId,
         OwnedRoomId,
         OwnedRoomOrAliasId,
         OwnedUserId,
@@ -905,6 +906,7 @@ impl ClientWorker {
         self.rcpt_handle = tokio::spawn({
             let store = store.clone();
             let client = self.client.clone();
+            let mut sent = HashMap::<OwnedRoomId, OwnedEventId>::default();
 
             async move {
                 // Update the displayed read receipts every 5 seconds.
@@ -914,7 +916,22 @@ impl ClientWorker {
                     interval.tick().await;
 
                     let receipts = update_receipts(&client).await;
-                    store.lock().await.application.set_receipts(receipts).await;
+                    let read = store.lock().await.application.set_receipts(receipts).await;
+
+                    for (room_id, read_till) in read.into_iter() {
+                        if let Some(read_sent) = sent.get(&room_id) {
+                            if read_sent == &read_till {
+                                // Skip unchanged receipts.
+                                continue;
+                            }
+                        }
+
+                        if let Some(room) = client.get_joined_room(&room_id) {
+                            if room.read_receipt(&read_till).await.is_ok() {
+                                sent.insert(room_id, read_till);
+                            }
+                        }
+                    }
                 }
             }
         })