Use StateStore to fulfill messages requests

Author: jsparber

matrix_sdk/src/client.rs

@@ -2731,6 +2731,7 @@ mod test {
                     },
                     media::get_content_thumbnail::Method,
                     membership::Invite3pidInit,
+                    message::get_message_events::Direction,
                     session::get_login_types::LoginType,
                     uiaa::{AuthData, UiaaResponse},
                 },
@@ -3895,4 +3896,150 @@ mod test {
 
         assert_eq!(client.whoami().await.unwrap().user_id, user_id);
     }
+
+    #[tokio::test]
+    async fn messages() {
+        let room_id = room_id!("!SVkFJHzfwvuaIEawgC:localhost");
+        let client = logged_in_client().await;
+
+        let _m = mock("GET", Matcher::Regex(r"^/_matrix/client/r0/sync\?.*$".to_string()))
+            .with_status(200)
+            .with_body(test_json::MORE_SYNC.to_string())
+            .match_header("authorization", "Bearer 1234")
+            .create();
+
+        let mocked_messages = mock(
+            "GET",
+            Matcher::Regex(
+                r"^/_matrix/client/r0/rooms/.*/messages.*from=t392-516_47314_0_7_1_1_1_11444_1.*"
+                    .to_string(),
+            ),
+        )
+        .with_status(200)
+        .with_body(test_json::SYNC_ROOM_MESSAGES_BATCH_1.to_string())
+        .match_header("authorization", "Bearer 1234")
+        .create();
+
+        let mocked_context =
+            mock("GET", Matcher::Regex(r"^/_matrix/client/r0/rooms/.*/context/.*".to_string()))
+                .with_status(200)
+                .with_body(test_json::CONTEXT_MESSAGE.to_string())
+                .match_header("authorization", "Bearer 1234")
+                .create();
+
+        let sync_settings = SyncSettings::new().timeout(Duration::from_millis(3000));
+
+        let _ = client.sync_once(sync_settings).await.unwrap();
+
+        let room = client.get_joined_room(&room_id).unwrap();
+
+        // Try to get the timeline starting at an event not known to the store.
+        let events = room
+            .messages(&event_id!("$f3h4d129462ha:example.com"), None, 3, Direction::Backward)
+            .await
+            .unwrap()
+            .unwrap();
+
+        let expected_events = [
+            event_id!("$f3h4d129462ha:example.com"),
+            event_id!("$143273582443PhrSnbefore1:example.org"),
+            event_id!("$143273582443PhrSnbefore2:example.org"),
+        ];
+
+        assert_eq!(events.len(), expected_events.len());
+        assert!(!events
+            .iter()
+            .map(|event| event.event_id())
+            .zip(&expected_events)
+            .any(|(a, b)| &a != b));
+
+        mocked_context.assert();
+
+        let expected_events = [
+            event_id!("$152037280074GZeOm:localhost"),
+            event_id!("$1444812213350496Caaaf:example.com"),
+            event_id!("$1444812213350496Cbbbf:example.com"),
+            event_id!("$1444812213350496Ccccf:example.com"),
+        ];
+
+        let events = room
+            .messages(
+                &event_id!("$152037280074GZeOm:localhost"),
+                None,
+                expected_events.len(),
+                Direction::Backward,
+            )
+            .await
+            .unwrap()
+            .unwrap();
+
+        assert_eq!(events.len(), expected_events.len());
+        assert!(!events
+            .iter()
+            .map(|event| event.event_id())
+            .zip(&expected_events)
+            .any(|(a, b)| &a != b));
+
+        let events = room
+            .messages(
+                &event_id!("$1444812213350496Ccccf:example.com"),
+                None,
+                expected_events.len(),
+                Direction::Forward,
+            )
+            .await
+            .unwrap()
+            .unwrap();
+
+        assert_eq!(events.len(), expected_events.len());
+        assert!(!events
+            .iter()
+            .rev()
+            .map(|event| event.event_id())
+            .zip(&expected_events)
+            .any(|(a, b)| &a != b));
+
+        let end_event = event_id!("$1444812213350496Cbbbf:example.com");
+        let events = room
+            .messages(
+                &event_id!("$152037280074GZeOm:localhost"),
+                Some(&end_event),
+                expected_events.len(),
+                Direction::Backward,
+            )
+            .await
+            .unwrap()
+            .unwrap();
+
+        assert_eq!(events.len(), 3);
+        assert_eq!(events.last().unwrap().event_id(), end_event);
+        assert!(!events
+            .iter()
+            .map(|event| event.event_id())
+            .zip(&expected_events[0..4])
+            .any(|(a, b)| &a != b));
+
+        let end_event = event_id!("$1444812213350496Cbbbf:example.com");
+        let events = room
+            .messages(
+                &event_id!("$1444812213350496Ccccf:example.com"),
+                Some(&end_event),
+                expected_events.len(),
+                Direction::Forward,
+            )
+            .await
+            .unwrap()
+            .unwrap();
+
+        assert_eq!(events.len(), 2);
+        assert_eq!(events.last().unwrap().event_id(), end_event);
+        assert!(!events
+            .iter()
+            .rev()
+            .map(|event| event.event_id())
+            .zip(&expected_events[2..])
+            .any(|(a, b)| &a != b));
+
+        mocked_messages.assert();
+    }
 }

matrix_sdk/src/room/common.rs

@@ -1,17 +1,20 @@
-use std::{ops::Deref, sync::Arc};
+use std::{cmp::min, convert::TryFrom, ops::Deref, sync::Arc};
 
-use matrix_sdk_base::deserialized_responses::MembersResponse;
+use matrix_sdk_base::deserialized_responses::{MembersResponse, SyncRoomEvent};
 use matrix_sdk_common::locks::Mutex;
 use ruma::{
     api::client::r0::{
+        context::get_context,
         media::{get_content, get_content_thumbnail},
         membership::{get_member_events, join_room_by_id, leave_room},
-        message::get_message_events,
+        message::{get_message_events, get_message_events::Direction},
     },
-    UserId,
+    events::AnyRoomEvent,
+    serde::Raw,
+    EventId, UserId,
 };
 
-use crate::{BaseRoom, Client, Result, RoomMember};
+use crate::{BaseRoom, Client, Result, RoomMember, UInt};
 
 /// A struct containing methods that are common for Joined, Invited and Left
 /// Rooms
@@ -110,43 +113,160 @@ impl Common {
         }
     }
 
-    /// Sends a request to `/_matrix/client/r0/rooms/{room_id}/messages` and
-    /// returns a `get_message_events::Response` that contains a chunk of
-    /// room and state events (`AnyRoomEvent` and `AnyStateEvent`).
+    /// Gets a slice of the timeline of this room
+    ///
+    /// Returns a slice of the timeline between `start` and `end`, no longer
+    /// then `limit`. If the number of events is fewer then `limit` it means
+    /// that in the given direction no more events exist.
+    /// If the timeline doesn't contain an event with the given `start` `None`
+    /// is returned.
     ///
     /// # Arguments
     ///
-    /// * `request` - The easiest way to create this request is using the
-    /// `get_message_events::Request` itself.
+    /// * `start` - An `EventId` that indicates the start of the slice.
+    ///
+    /// * `end` - An `EventId` that indicates the end of the slice.
+    ///
+    /// * `limit` - The maximum number of events that should be returned.
+    ///
+    /// * `direction` - The direction of the search and returned events.
     ///
     /// # Examples
     /// ```no_run
     /// # use std::convert::TryFrom;
     /// use matrix_sdk::Client;
-    /// # use matrix_sdk::identifiers::room_id;
-    /// # use matrix_sdk::api::r0::filter::RoomEventFilter;
-    /// # use matrix_sdk::api::r0::message::get_message_events::Request as MessagesRequest;
+    /// # use matrix_sdk::identifiers::{event_id, room_id};
+    /// # use matrix_sdk::api::r0::message::get_message_events::Direction;
     /// # use url::Url;
     ///
     /// # let homeserver = Url::parse("http://example.com").unwrap();
     /// let room_id = room_id!("!roomid:example.com");
-    /// let request = MessagesRequest::backward(&room_id, "t47429-4392820_219380_26003_2265");
     ///
     /// let mut client = Client::new(homeserver).unwrap();
     /// # let room = client
     /// #    .get_joined_room(&room_id)
     /// #    .unwrap();
     /// # use futures::executor::block_on;
     /// # block_on(async {
-    /// assert!(room.messages(request).await.is_ok());
+    /// assert!(room.messages(&event_id!("$xxxxxx:example.org"), None, 10, Direction::Backward).await.is_ok());
     /// # });
     /// ```
     pub async fn messages(
         &self,
-        request: impl Into<get_message_events::Request<'_>>,
-    ) -> Result<get_message_events::Response> {
-        let request = request.into();
-        self.client.send(request, None).await
+        start: &EventId,
+        end: Option<&EventId>,
+        limit: usize,
+        direction: Direction,
+    ) -> Result<Option<Vec<SyncRoomEvent>>> {
+        let room_id = self.inner.room_id();
+        let events = if let Some(mut stored) = self
+            .client
+            .store()
+            .get_timeline(room_id, Some(start), end, Some(limit), direction.clone())
+            .await?
+        {
+            // We found a gab or the end of the stored timeline.
+            if let Some(token) = stored.token {
+                let mut request = get_message_events::Request::new(
+                    self.inner.room_id(),
+                    &token,
+                    direction.clone(),
+                );
+                request.limit =
+                    UInt::try_from((limit - stored.events.len()) as u64).unwrap_or(UInt::MAX);
+
+                let response = self.client.send(request, None).await?;
+
+                // FIXME: we may recevied an invalied server response that ruma considers valid
+                // See https://github.com/ruma/ruma/issues/644
+                if response.end.is_none() && response.start.is_none() {
+                    return Ok(Some(stored.events));
+                }
+
+                let response_events = self
+                    .client
+                    .base_client
+                    .receive_messages(room_id, &direction, &response)
+                    .await?;
+
+                let mut response_events = if let Some(end) = end {
+                    if let Some(position) =
+                        response_events.iter().position(|event| &event.event_id() == end)
+                    {
+                        response_events.into_iter().take(position + 1).collect()
+                    } else {
+                        response_events
+                    }
+                } else {
+                    response_events
+                };
+
+                match direction {
+                    Direction::Forward => {
+                        response_events.append(&mut stored.events);
+                        stored.events = response_events;
+                    }
+                    Direction::Backward => stored.events.append(&mut response_events),
+                }
+            }
+            stored.events
+        } else {
+            // Fallback to context API because we don't know the start event
+            let mut request = get_context::Request::new(room_id, start);
+
+            // We need to take limit twice because the context api returns events before
+            // and after the given event
+            request.limit = UInt::try_from((limit * 2) as u64).unwrap_or(UInt::MAX);
+
+            let mut context = self.client.send(request, None).await?;
+
+            let event = if let Some(event) = context.event {
+                event
+            } else {
+                return Ok(None);
+            };
+
+            let mut response = get_message_events::Response::new();
+            response.start = context.start;
+            response.end = context.end;
+            let before_length = context.events_before.len();
+            let after_length = context.events_after.len();
+            let mut events: Vec<Raw<AnyRoomEvent>> =
+                context.events_after.into_iter().rev().collect();
+            events.push(event);
+            events.append(&mut context.events_before);
+            response.chunk = events;
+            response.state = context.state;
+            let response_events = self
+                .client
+                .base_client
+                .receive_messages(room_id, &Direction::Backward, &response)
+                .await?;
+
+            let response_events: Vec<SyncRoomEvent> = match direction {
+                Direction::Forward => {
+                    let lower_bound = if before_length > limit { before_length - limit } else { 0 };
+                    response_events[lower_bound..=before_length].to_vec()
+                }
+                Direction::Backward => response_events
+                    [after_length..min(response_events.len(), after_length + limit)]
+                    .to_vec(),
+            };
+
+            if let Some(end) = end {
+                if let Some(position) =
+                    response_events.iter().position(|event| &event.event_id() == end)
+                {
+                    response_events.into_iter().take(position + 1).collect()
+                } else {
+                    response_events
+                }
+            } else {
+                response_events
+            }
+        };
+
+        Ok(Some(events))
     }
 
     pub(crate) async fn request_members(&self) -> Result<Option<MembersResponse>> {