base: Replace dashmap usage by std types
This commit is contained in:
committed by
Jonas Platte
parent
1d90dd554c
commit
113bdbd835
Generated
-1
@@ -3123,7 +3123,6 @@ dependencies = [
|
||||
"async-trait",
|
||||
"bitflags 2.4.0",
|
||||
"ctor",
|
||||
"dashmap",
|
||||
"eyeball",
|
||||
"futures-executor",
|
||||
"futures-util",
|
||||
|
||||
@@ -39,7 +39,6 @@ assert_matches = { workspace = true, optional = true }
|
||||
assert_matches2 = { workspace = true, optional = true }
|
||||
async-trait = { workspace = true }
|
||||
bitflags = "2.1.0"
|
||||
dashmap = { workspace = true }
|
||||
eyeball = { workspace = true }
|
||||
futures-util = { workspace = true }
|
||||
http = { workspace = true, optional = true }
|
||||
|
||||
@@ -13,13 +13,12 @@
|
||||
// limitations under the License.
|
||||
|
||||
use std::{
|
||||
collections::{BTreeMap, BTreeSet},
|
||||
collections::{BTreeMap, BTreeSet, HashMap},
|
||||
iter,
|
||||
sync::RwLock,
|
||||
sync::RwLock as StdRwLock,
|
||||
};
|
||||
|
||||
use async_trait::async_trait;
|
||||
use dashmap::DashMap;
|
||||
use matrix_sdk_common::instant::Instant;
|
||||
use ruma::{
|
||||
canonical_json::{redact, RedactedBecause},
|
||||
@@ -48,31 +47,38 @@ use crate::{
|
||||
#[allow(clippy::type_complexity)]
|
||||
#[derive(Debug, Default)]
|
||||
pub struct MemoryStore {
|
||||
user_avatar_url: DashMap<String, String>,
|
||||
sync_token: RwLock<Option<String>>,
|
||||
filters: DashMap<String, String>,
|
||||
account_data: DashMap<GlobalAccountDataEventType, Raw<AnyGlobalAccountDataEvent>>,
|
||||
profiles: DashMap<OwnedRoomId, DashMap<OwnedUserId, MinimalRoomMemberEvent>>,
|
||||
display_names: DashMap<OwnedRoomId, DashMap<String, BTreeSet<OwnedUserId>>>,
|
||||
members: DashMap<OwnedRoomId, DashMap<OwnedUserId, MembershipState>>,
|
||||
room_info: DashMap<OwnedRoomId, RoomInfo>,
|
||||
room_state:
|
||||
DashMap<OwnedRoomId, DashMap<StateEventType, DashMap<String, Raw<AnySyncStateEvent>>>>,
|
||||
room_account_data:
|
||||
DashMap<OwnedRoomId, DashMap<RoomAccountDataEventType, Raw<AnyRoomAccountDataEvent>>>,
|
||||
stripped_room_state:
|
||||
DashMap<OwnedRoomId, DashMap<StateEventType, DashMap<String, Raw<AnyStrippedStateEvent>>>>,
|
||||
stripped_members: DashMap<OwnedRoomId, DashMap<OwnedUserId, MembershipState>>,
|
||||
presence: DashMap<OwnedUserId, Raw<PresenceEvent>>,
|
||||
room_user_receipts: DashMap<
|
||||
OwnedRoomId,
|
||||
DashMap<(String, Option<String>), DashMap<OwnedUserId, (OwnedEventId, Receipt)>>,
|
||||
user_avatar_url: StdRwLock<HashMap<String, String>>,
|
||||
sync_token: StdRwLock<Option<String>>,
|
||||
filters: StdRwLock<HashMap<String, String>>,
|
||||
account_data: StdRwLock<HashMap<GlobalAccountDataEventType, Raw<AnyGlobalAccountDataEvent>>>,
|
||||
profiles: StdRwLock<HashMap<OwnedRoomId, HashMap<OwnedUserId, MinimalRoomMemberEvent>>>,
|
||||
display_names: StdRwLock<HashMap<OwnedRoomId, HashMap<String, BTreeSet<OwnedUserId>>>>,
|
||||
members: StdRwLock<HashMap<OwnedRoomId, HashMap<OwnedUserId, MembershipState>>>,
|
||||
room_info: StdRwLock<HashMap<OwnedRoomId, RoomInfo>>,
|
||||
room_state: StdRwLock<
|
||||
HashMap<OwnedRoomId, HashMap<StateEventType, HashMap<String, Raw<AnySyncStateEvent>>>>,
|
||||
>,
|
||||
room_event_receipts: DashMap<
|
||||
OwnedRoomId,
|
||||
DashMap<(String, Option<String>), DashMap<OwnedEventId, DashMap<OwnedUserId, Receipt>>>,
|
||||
room_account_data: StdRwLock<
|
||||
HashMap<OwnedRoomId, HashMap<RoomAccountDataEventType, Raw<AnyRoomAccountDataEvent>>>,
|
||||
>,
|
||||
custom: DashMap<Vec<u8>, Vec<u8>>,
|
||||
stripped_room_state: StdRwLock<
|
||||
HashMap<OwnedRoomId, HashMap<StateEventType, HashMap<String, Raw<AnyStrippedStateEvent>>>>,
|
||||
>,
|
||||
stripped_members: StdRwLock<HashMap<OwnedRoomId, HashMap<OwnedUserId, MembershipState>>>,
|
||||
presence: StdRwLock<HashMap<OwnedUserId, Raw<PresenceEvent>>>,
|
||||
room_user_receipts: StdRwLock<
|
||||
HashMap<
|
||||
OwnedRoomId,
|
||||
HashMap<(String, Option<String>), HashMap<OwnedUserId, (OwnedEventId, Receipt)>>,
|
||||
>,
|
||||
>,
|
||||
room_event_receipts: StdRwLock<
|
||||
HashMap<
|
||||
OwnedRoomId,
|
||||
HashMap<(String, Option<String>), HashMap<OwnedEventId, HashMap<OwnedUserId, Receipt>>>,
|
||||
>,
|
||||
>,
|
||||
custom: StdRwLock<HashMap<Vec<u8>, Vec<u8>>>,
|
||||
}
|
||||
|
||||
impl MemoryStore {
|
||||
@@ -88,12 +94,18 @@ impl MemoryStore {
|
||||
}
|
||||
StateStoreDataKey::Filter(filter_name) => Ok(self
|
||||
.filters
|
||||
.read()
|
||||
.unwrap()
|
||||
.get(filter_name)
|
||||
.map(|f| StateStoreDataValue::Filter(f.value().clone()))),
|
||||
.cloned()
|
||||
.map(StateStoreDataValue::Filter)),
|
||||
StateStoreDataKey::UserAvatarUrl(user_id) => Ok(self
|
||||
.user_avatar_url
|
||||
.read()
|
||||
.unwrap()
|
||||
.get(user_id.as_str())
|
||||
.map(|u| StateStoreDataValue::UserAvatarUrl(u.value().clone()))),
|
||||
.cloned()
|
||||
.map(StateStoreDataValue::UserAvatarUrl)),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -108,13 +120,13 @@ impl MemoryStore {
|
||||
Some(value.into_sync_token().expect("Session data not a sync token"))
|
||||
}
|
||||
StateStoreDataKey::Filter(filter_name) => {
|
||||
self.filters.insert(
|
||||
self.filters.write().unwrap().insert(
|
||||
filter_name.to_owned(),
|
||||
value.into_filter().expect("Session data not a filter"),
|
||||
);
|
||||
}
|
||||
StateStoreDataKey::UserAvatarUrl(user_id) => {
|
||||
self.filters.insert(
|
||||
self.filters.write().unwrap().insert(
|
||||
user_id.to_string(),
|
||||
value.into_user_avatar_url().expect("Session data not a user avatar url"),
|
||||
);
|
||||
@@ -128,10 +140,10 @@ impl MemoryStore {
|
||||
match key {
|
||||
StateStoreDataKey::SyncToken => *self.sync_token.write().unwrap() = None,
|
||||
StateStoreDataKey::Filter(filter_name) => {
|
||||
self.filters.remove(filter_name);
|
||||
self.filters.write().unwrap().remove(filter_name);
|
||||
}
|
||||
StateStoreDataKey::UserAvatarUrl(user_id) => {
|
||||
self.filters.remove(user_id.as_str());
|
||||
self.filters.write().unwrap().remove(user_id.as_str());
|
||||
}
|
||||
}
|
||||
|
||||
@@ -148,6 +160,8 @@ impl MemoryStore {
|
||||
for (room, users) in &changes.profiles {
|
||||
for (user_id, profile) in users {
|
||||
self.profiles
|
||||
.write()
|
||||
.unwrap()
|
||||
.entry(room.clone())
|
||||
.or_default()
|
||||
.insert(user_id.clone(), profile.clone());
|
||||
@@ -157,164 +171,201 @@ impl MemoryStore {
|
||||
for (room, map) in &changes.ambiguity_maps {
|
||||
for (display_name, display_names) in map {
|
||||
self.display_names
|
||||
.write()
|
||||
.unwrap()
|
||||
.entry(room.clone())
|
||||
.or_default()
|
||||
.insert(display_name.clone(), display_names.clone());
|
||||
}
|
||||
}
|
||||
|
||||
for (event_type, event) in &changes.account_data {
|
||||
self.account_data.insert(event_type.clone(), event.clone());
|
||||
}
|
||||
|
||||
for (room, events) in &changes.room_account_data {
|
||||
for (event_type, event) in events {
|
||||
self.room_account_data
|
||||
.entry(room.clone())
|
||||
.or_default()
|
||||
.insert(event_type.clone(), event.clone());
|
||||
{
|
||||
let mut account_data = self.account_data.write().unwrap();
|
||||
for (event_type, event) in &changes.account_data {
|
||||
account_data.insert(event_type.clone(), event.clone());
|
||||
}
|
||||
}
|
||||
|
||||
for (room, event_types) in &changes.state {
|
||||
for (event_type, events) in event_types {
|
||||
for (state_key, raw_event) in events {
|
||||
self.room_state
|
||||
{
|
||||
let mut room_account_data = self.room_account_data.write().unwrap();
|
||||
for (room, events) in &changes.room_account_data {
|
||||
for (event_type, event) in events {
|
||||
room_account_data
|
||||
.entry(room.clone())
|
||||
.or_default()
|
||||
.entry(event_type.clone())
|
||||
.or_default()
|
||||
.insert(state_key.to_owned(), raw_event.clone());
|
||||
self.stripped_room_state.remove(room);
|
||||
.insert(event_type.clone(), event.clone());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if *event_type == StateEventType::RoomMember {
|
||||
let event = match raw_event.deserialize_as::<SyncRoomMemberEvent>() {
|
||||
Ok(ev) => ev,
|
||||
Err(e) => {
|
||||
let event_id: Option<String> =
|
||||
raw_event.get_field("event_id").ok().flatten();
|
||||
debug!(event_id, "Failed to deserialize member event: {e}");
|
||||
continue;
|
||||
}
|
||||
};
|
||||
{
|
||||
let mut room_state = self.room_state.write().unwrap();
|
||||
let mut stripped_room_state = self.stripped_room_state.write().unwrap();
|
||||
let mut members = self.members.write().unwrap();
|
||||
let mut stripped_members = self.stripped_members.write().unwrap();
|
||||
|
||||
self.stripped_members.remove(room);
|
||||
|
||||
self.members
|
||||
for (room, event_types) in &changes.state {
|
||||
for (event_type, events) in event_types {
|
||||
for (state_key, raw_event) in events {
|
||||
room_state
|
||||
.entry(room.clone())
|
||||
.or_default()
|
||||
.insert(event.state_key().to_owned(), event.membership().clone());
|
||||
.entry(event_type.clone())
|
||||
.or_default()
|
||||
.insert(state_key.to_owned(), raw_event.clone());
|
||||
stripped_room_state.remove(room);
|
||||
|
||||
if *event_type == StateEventType::RoomMember {
|
||||
let event = match raw_event.deserialize_as::<SyncRoomMemberEvent>() {
|
||||
Ok(ev) => ev,
|
||||
Err(e) => {
|
||||
let event_id: Option<String> =
|
||||
raw_event.get_field("event_id").ok().flatten();
|
||||
debug!(event_id, "Failed to deserialize member event: {e}");
|
||||
continue;
|
||||
}
|
||||
};
|
||||
|
||||
stripped_members.remove(room);
|
||||
|
||||
members
|
||||
.entry(room.clone())
|
||||
.or_default()
|
||||
.insert(event.state_key().to_owned(), event.membership().clone());
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
for (room_id, room_info) in &changes.room_infos {
|
||||
self.room_info.insert(room_id.clone(), room_info.clone());
|
||||
{
|
||||
let mut room_info = self.room_info.write().unwrap();
|
||||
for (room_id, info) in &changes.room_infos {
|
||||
room_info.insert(room_id.clone(), info.clone());
|
||||
}
|
||||
}
|
||||
|
||||
for (sender, event) in &changes.presence {
|
||||
self.presence.insert(sender.clone(), event.clone());
|
||||
{
|
||||
let mut presence = self.presence.write().unwrap();
|
||||
for (sender, event) in &changes.presence {
|
||||
presence.insert(sender.clone(), event.clone());
|
||||
}
|
||||
}
|
||||
|
||||
for (room, event_types) in &changes.stripped_state {
|
||||
for (event_type, events) in event_types {
|
||||
for (state_key, raw_event) in events {
|
||||
self.stripped_room_state
|
||||
.entry(room.clone())
|
||||
.or_default()
|
||||
.entry(event_type.clone())
|
||||
.or_default()
|
||||
.insert(state_key.to_owned(), raw_event.clone());
|
||||
{
|
||||
let mut stripped_room_state = self.stripped_room_state.write().unwrap();
|
||||
let mut stripped_members = self.stripped_members.write().unwrap();
|
||||
|
||||
if *event_type == StateEventType::RoomMember {
|
||||
let event = match raw_event.deserialize_as::<StrippedRoomMemberEvent>() {
|
||||
Ok(ev) => ev,
|
||||
Err(e) => {
|
||||
let event_id: Option<String> =
|
||||
raw_event.get_field("event_id").ok().flatten();
|
||||
debug!(
|
||||
event_id,
|
||||
"Failed to deserialize stripped member event: {e}"
|
||||
);
|
||||
continue;
|
||||
}
|
||||
};
|
||||
|
||||
self.stripped_members
|
||||
for (room, event_types) in &changes.stripped_state {
|
||||
for (event_type, events) in event_types {
|
||||
for (state_key, raw_event) in events {
|
||||
stripped_room_state
|
||||
.entry(room.clone())
|
||||
.or_default()
|
||||
.insert(event.state_key, event.content.membership.clone());
|
||||
.entry(event_type.clone())
|
||||
.or_default()
|
||||
.insert(state_key.to_owned(), raw_event.clone());
|
||||
|
||||
if *event_type == StateEventType::RoomMember {
|
||||
let event = match raw_event.deserialize_as::<StrippedRoomMemberEvent>()
|
||||
{
|
||||
Ok(ev) => ev,
|
||||
Err(e) => {
|
||||
let event_id: Option<String> =
|
||||
raw_event.get_field("event_id").ok().flatten();
|
||||
debug!(
|
||||
event_id,
|
||||
"Failed to deserialize stripped member event: {e}"
|
||||
);
|
||||
continue;
|
||||
}
|
||||
};
|
||||
|
||||
stripped_members
|
||||
.entry(room.clone())
|
||||
.or_default()
|
||||
.insert(event.state_key, event.content.membership.clone());
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
for (room, content) in &changes.receipts {
|
||||
for (event_id, receipts) in &content.0 {
|
||||
for (receipt_type, receipts) in receipts {
|
||||
for (user_id, receipt) in receipts {
|
||||
let thread = receipt.thread.as_str().map(ToOwned::to_owned);
|
||||
// Add the receipt to the room user receipts
|
||||
if let Some((old_event, _)) = self
|
||||
.room_user_receipts
|
||||
.entry(room.clone())
|
||||
.or_default()
|
||||
.entry((receipt_type.to_string(), thread.clone()))
|
||||
.or_default()
|
||||
.insert(user_id.clone(), (event_id.clone(), receipt.clone()))
|
||||
{
|
||||
// Remove the old receipt from the room event receipts
|
||||
if let Some(receipt_map) = self.room_event_receipts.get(room) {
|
||||
if let Some(event_map) =
|
||||
receipt_map.get(&(receipt_type.to_string(), thread.clone()))
|
||||
{
|
||||
if let Some(user_map) = event_map.get_mut(&old_event) {
|
||||
user_map.remove(user_id);
|
||||
{
|
||||
let mut room_user_receipts = self.room_user_receipts.write().unwrap();
|
||||
let mut room_event_receipts = self.room_event_receipts.write().unwrap();
|
||||
|
||||
for (room, content) in &changes.receipts {
|
||||
for (event_id, receipts) in &content.0 {
|
||||
for (receipt_type, receipts) in receipts {
|
||||
for (user_id, receipt) in receipts {
|
||||
let thread = receipt.thread.as_str().map(ToOwned::to_owned);
|
||||
// Add the receipt to the room user receipts
|
||||
if let Some((old_event, _)) = room_user_receipts
|
||||
.entry(room.clone())
|
||||
.or_default()
|
||||
.entry((receipt_type.to_string(), thread.clone()))
|
||||
.or_default()
|
||||
.insert(user_id.clone(), (event_id.clone(), receipt.clone()))
|
||||
{
|
||||
// Remove the old receipt from the room event receipts
|
||||
if let Some(receipt_map) = room_event_receipts.get_mut(room) {
|
||||
if let Some(event_map) = receipt_map
|
||||
.get_mut(&(receipt_type.to_string(), thread.clone()))
|
||||
{
|
||||
if let Some(user_map) = event_map.get_mut(&old_event) {
|
||||
user_map.remove(user_id);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Add the receipt to the room event receipts
|
||||
self.room_event_receipts
|
||||
.entry(room.clone())
|
||||
.or_default()
|
||||
.entry((receipt_type.to_string(), thread))
|
||||
.or_default()
|
||||
.entry(event_id.clone())
|
||||
.or_default()
|
||||
.insert(user_id.clone(), receipt.clone());
|
||||
// Add the receipt to the room event receipts
|
||||
room_event_receipts
|
||||
.entry(room.clone())
|
||||
.or_default()
|
||||
.entry((receipt_type.to_string(), thread))
|
||||
.or_default()
|
||||
.entry(event_id.clone())
|
||||
.or_default()
|
||||
.insert(user_id.clone(), receipt.clone());
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let make_room_version = |room_id| {
|
||||
self.room_info
|
||||
.get(room_id)
|
||||
.and_then(|info| info.room_version().cloned())
|
||||
.unwrap_or_else(|| {
|
||||
warn!(?room_id, "Unable to find the room version, assuming version 9");
|
||||
RoomVersionId::V9
|
||||
})
|
||||
};
|
||||
{
|
||||
let room_info = self.room_info.read().unwrap();
|
||||
let mut room_state = self.room_state.write().unwrap();
|
||||
|
||||
for (room_id, redactions) in &changes.redactions {
|
||||
let mut room_version = None;
|
||||
if let Some(room) = self.room_state.get_mut(room_id) {
|
||||
for mut ref_room_mu in room.iter_mut() {
|
||||
for mut ref_evt_mu in ref_room_mu.value_mut().iter_mut() {
|
||||
let raw_evt = ref_evt_mu.value_mut();
|
||||
if let Ok(Some(event_id)) = raw_evt.get_field::<OwnedEventId>("event_id") {
|
||||
if let Some(redaction) = redactions.get(&event_id) {
|
||||
let redacted = redact(
|
||||
raw_evt.deserialize_as::<CanonicalJsonObject>()?,
|
||||
room_version.get_or_insert_with(|| make_room_version(room_id)),
|
||||
Some(RedactedBecause::from_raw_event(redaction)?),
|
||||
)
|
||||
.map_err(StoreError::Redaction)?;
|
||||
*raw_evt = Raw::new(&redacted)?.cast();
|
||||
let make_room_version = |room_id| {
|
||||
room_info.get(room_id).and_then(|info| info.room_version().cloned()).unwrap_or_else(
|
||||
|| {
|
||||
warn!(?room_id, "Unable to find the room version, assuming version 9");
|
||||
RoomVersionId::V9
|
||||
},
|
||||
)
|
||||
};
|
||||
|
||||
for (room_id, redactions) in &changes.redactions {
|
||||
let mut room_version = None;
|
||||
if let Some(room) = room_state.get_mut(room_id) {
|
||||
for ref_room_mu in room.values_mut() {
|
||||
for raw_evt in ref_room_mu.values_mut() {
|
||||
if let Ok(Some(event_id)) =
|
||||
raw_evt.get_field::<OwnedEventId>("event_id")
|
||||
{
|
||||
if let Some(redaction) = redactions.get(&event_id) {
|
||||
let redacted = redact(
|
||||
raw_evt.deserialize_as::<CanonicalJsonObject>()?,
|
||||
room_version
|
||||
.get_or_insert_with(|| make_room_version(room_id)),
|
||||
Some(RedactedBecause::from_raw_event(redaction)?),
|
||||
)
|
||||
.map_err(StoreError::Redaction)?;
|
||||
*raw_evt = Raw::new(&redacted)?.cast();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -337,7 +388,9 @@ impl MemoryStore {
|
||||
return Ok(Vec::new());
|
||||
}
|
||||
|
||||
Ok(user_ids.filter_map(|user_id| self.presence.get(user_id).map(|p| p.clone())).collect())
|
||||
Ok(user_ids
|
||||
.filter_map(|user_id| self.presence.read().unwrap().get(user_id).cloned())
|
||||
.collect())
|
||||
}
|
||||
|
||||
async fn get_state_events(
|
||||
@@ -345,18 +398,26 @@ impl MemoryStore {
|
||||
room_id: &RoomId,
|
||||
event_type: StateEventType,
|
||||
) -> Result<Vec<RawAnySyncOrStrippedState>> {
|
||||
if let Some(v) = self.stripped_room_state.get(room_id).as_ref().and_then(|events| {
|
||||
events.get(&event_type).map(|s| {
|
||||
s.iter().map(|e| RawAnySyncOrStrippedState::Stripped(e.clone())).collect::<Vec<_>>()
|
||||
})
|
||||
}) {
|
||||
Ok(v)
|
||||
} else if let Some(v) = self.room_state.get(room_id).as_ref().and_then(|events| {
|
||||
events.get(&event_type).map(|s| {
|
||||
s.iter().map(|e| RawAnySyncOrStrippedState::Sync(e.clone())).collect::<Vec<_>>()
|
||||
})
|
||||
}) {
|
||||
Ok(v)
|
||||
if let Some(stripped_state_events) = self
|
||||
.stripped_room_state
|
||||
.read()
|
||||
.unwrap()
|
||||
.get(room_id)
|
||||
.and_then(|events| events.get(&event_type))
|
||||
{
|
||||
Ok(stripped_state_events
|
||||
.values()
|
||||
.cloned()
|
||||
.map(RawAnySyncOrStrippedState::Stripped)
|
||||
.collect::<Vec<_>>())
|
||||
} else if let Some(sync_state_events) =
|
||||
self.room_state.read().unwrap().get(room_id).and_then(|events| events.get(&event_type))
|
||||
{
|
||||
Ok(sync_state_events
|
||||
.values()
|
||||
.cloned()
|
||||
.map(RawAnySyncOrStrippedState::Sync)
|
||||
.collect::<Vec<_>>())
|
||||
} else {
|
||||
Ok(Vec::new())
|
||||
}
|
||||
@@ -370,8 +431,9 @@ impl MemoryStore {
|
||||
) -> Result<Vec<RawAnySyncOrStrippedState>> {
|
||||
if let Some(stripped_state_events) = self
|
||||
.stripped_room_state
|
||||
.read()
|
||||
.unwrap()
|
||||
.get(room_id)
|
||||
.as_ref()
|
||||
.and_then(|events| events.get(&event_type))
|
||||
{
|
||||
Ok(state_keys
|
||||
@@ -383,7 +445,7 @@ impl MemoryStore {
|
||||
})
|
||||
.collect())
|
||||
} else if let Some(sync_state_events) =
|
||||
self.room_state.get(room_id).as_ref().and_then(|events| events.get(&event_type))
|
||||
self.room_state.read().unwrap().get(room_id).and_then(|events| events.get(&event_type))
|
||||
{
|
||||
Ok(state_keys
|
||||
.iter()
|
||||
@@ -410,7 +472,8 @@ impl MemoryStore {
|
||||
return Ok(BTreeMap::new());
|
||||
}
|
||||
|
||||
let Some(room_profiles) = self.profiles.get(room_id) else {
|
||||
let profiles = self.profiles.read().unwrap();
|
||||
let Some(room_profiles) = profiles.get(room_id) else {
|
||||
return Ok(BTreeMap::new());
|
||||
};
|
||||
|
||||
@@ -432,26 +495,32 @@ impl MemoryStore {
|
||||
) -> Vec<OwnedUserId> {
|
||||
let map = if stripped { &self.stripped_members } else { &self.members };
|
||||
|
||||
map.get(room_id)
|
||||
.map(|u| {
|
||||
u.iter()
|
||||
.filter_map(|u| memberships.matches(u.value()).then(|| u.key().clone()))
|
||||
map.read()
|
||||
.unwrap()
|
||||
.get(room_id)
|
||||
.map(|members| {
|
||||
members
|
||||
.iter()
|
||||
.filter_map(|(user_id, membership)| {
|
||||
memberships.matches(membership).then_some(user_id)
|
||||
})
|
||||
.cloned()
|
||||
.collect()
|
||||
})
|
||||
.unwrap_or_default()
|
||||
}
|
||||
|
||||
fn get_room_infos(&self) -> Vec<RoomInfo> {
|
||||
self.room_info.iter().map(|r| r.clone()).collect()
|
||||
self.room_info.read().unwrap().values().cloned().collect()
|
||||
}
|
||||
|
||||
fn get_stripped_room_infos(&self) -> Vec<RoomInfo> {
|
||||
self.room_info
|
||||
.iter()
|
||||
.filter_map(|r| match r.state() {
|
||||
RoomState::Invited => Some(r.clone()),
|
||||
_ => None,
|
||||
})
|
||||
.read()
|
||||
.unwrap()
|
||||
.values()
|
||||
.filter(|r| matches!(r.state(), RoomState::Invited))
|
||||
.cloned()
|
||||
.collect()
|
||||
}
|
||||
|
||||
@@ -469,7 +538,8 @@ impl MemoryStore {
|
||||
return Ok(BTreeMap::new());
|
||||
}
|
||||
|
||||
let Some(room_names) = self.display_names.get(room_id) else {
|
||||
let read_guard = &self.display_names.read().unwrap();
|
||||
let Some(room_names) = read_guard.get(room_id) else {
|
||||
return Ok(BTreeMap::new());
|
||||
};
|
||||
|
||||
@@ -480,7 +550,7 @@ impl MemoryStore {
|
||||
&self,
|
||||
event_type: GlobalAccountDataEventType,
|
||||
) -> Result<Option<Raw<AnyGlobalAccountDataEvent>>> {
|
||||
Ok(self.account_data.get(&event_type).map(|e| e.clone()))
|
||||
Ok(self.account_data.read().unwrap().get(&event_type).cloned())
|
||||
}
|
||||
|
||||
async fn get_room_account_data_event(
|
||||
@@ -488,7 +558,13 @@ impl MemoryStore {
|
||||
room_id: &RoomId,
|
||||
event_type: RoomAccountDataEventType,
|
||||
) -> Result<Option<Raw<AnyRoomAccountDataEvent>>> {
|
||||
Ok(self.room_account_data.get(room_id).and_then(|m| m.get(&event_type).map(|e| e.clone())))
|
||||
Ok(self
|
||||
.room_account_data
|
||||
.read()
|
||||
.unwrap()
|
||||
.get(room_id)
|
||||
.and_then(|m| m.get(&event_type))
|
||||
.cloned())
|
||||
}
|
||||
|
||||
async fn get_user_room_receipt_event(
|
||||
@@ -498,9 +574,9 @@ impl MemoryStore {
|
||||
thread: ReceiptThread,
|
||||
user_id: &UserId,
|
||||
) -> Result<Option<(OwnedEventId, Receipt)>> {
|
||||
Ok(self.room_user_receipts.get(room_id).and_then(|m| {
|
||||
Ok(self.room_user_receipts.read().unwrap().get(room_id).and_then(|m| {
|
||||
m.get(&(receipt_type.to_string(), thread.as_str().map(ToOwned::to_owned)))
|
||||
.and_then(|m| m.get(user_id).map(|r| r.clone()))
|
||||
.and_then(|m| m.get(user_id).cloned())
|
||||
}))
|
||||
}
|
||||
|
||||
@@ -513,29 +589,29 @@ impl MemoryStore {
|
||||
) -> Result<Vec<(OwnedUserId, Receipt)>> {
|
||||
Ok(self
|
||||
.room_event_receipts
|
||||
.read()
|
||||
.unwrap()
|
||||
.get(room_id)
|
||||
.and_then(|m| {
|
||||
m.get(&(receipt_type.to_string(), thread.as_str().map(ToOwned::to_owned))).and_then(
|
||||
|m| {
|
||||
m.get(event_id).map(|m| {
|
||||
m.iter().map(|r| (r.key().clone(), r.value().clone())).collect()
|
||||
})
|
||||
},
|
||||
)
|
||||
m.get(&(receipt_type.to_string(), thread.as_str().map(ToOwned::to_owned)))
|
||||
})
|
||||
.and_then(|m| {
|
||||
m.get(event_id)
|
||||
.map(|m| m.iter().map(|(key, value)| (key.clone(), value.clone())).collect())
|
||||
})
|
||||
.unwrap_or_default())
|
||||
}
|
||||
|
||||
async fn get_custom_value(&self, key: &[u8]) -> Result<Option<Vec<u8>>> {
|
||||
Ok(self.custom.get(key).map(|e| e.value().clone()))
|
||||
Ok(self.custom.read().unwrap().get(key).cloned())
|
||||
}
|
||||
|
||||
async fn set_custom_value(&self, key: &[u8], value: Vec<u8>) -> Result<Option<Vec<u8>>> {
|
||||
Ok(self.custom.insert(key.to_vec(), value))
|
||||
Ok(self.custom.write().unwrap().insert(key.to_vec(), value))
|
||||
}
|
||||
|
||||
async fn remove_custom_value(&self, key: &[u8]) -> Result<Option<Vec<u8>>> {
|
||||
Ok(self.custom.remove(key).map(|entry| entry.1))
|
||||
Ok(self.custom.write().unwrap().remove(key))
|
||||
}
|
||||
|
||||
// The in-memory store doesn't cache media
|
||||
@@ -553,16 +629,16 @@ impl MemoryStore {
|
||||
}
|
||||
|
||||
async fn remove_room(&self, room_id: &RoomId) -> Result<()> {
|
||||
self.profiles.remove(room_id);
|
||||
self.display_names.remove(room_id);
|
||||
self.members.remove(room_id);
|
||||
self.room_info.remove(room_id);
|
||||
self.room_state.remove(room_id);
|
||||
self.room_account_data.remove(room_id);
|
||||
self.stripped_room_state.remove(room_id);
|
||||
self.stripped_members.remove(room_id);
|
||||
self.room_user_receipts.remove(room_id);
|
||||
self.room_event_receipts.remove(room_id);
|
||||
self.profiles.write().unwrap().remove(room_id);
|
||||
self.display_names.write().unwrap().remove(room_id);
|
||||
self.members.write().unwrap().remove(room_id);
|
||||
self.room_info.write().unwrap().remove(room_id);
|
||||
self.room_state.write().unwrap().remove(room_id);
|
||||
self.room_account_data.write().unwrap().remove(room_id);
|
||||
self.stripped_room_state.write().unwrap().remove(room_id);
|
||||
self.stripped_members.write().unwrap().remove(room_id);
|
||||
self.room_user_receipts.write().unwrap().remove(room_id);
|
||||
self.room_event_receipts.write().unwrap().remove(room_id);
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -27,7 +27,7 @@ use std::{
|
||||
pin::Pin,
|
||||
result::Result as StdResult,
|
||||
str::Utf8Error,
|
||||
sync::Arc,
|
||||
sync::{Arc, RwLock as StdRwLock},
|
||||
};
|
||||
|
||||
use once_cell::sync::OnceCell;
|
||||
@@ -37,7 +37,6 @@ use once_cell::sync::OnceCell;
|
||||
pub mod integration_tests;
|
||||
mod traits;
|
||||
|
||||
use dashmap::DashMap;
|
||||
#[cfg(feature = "e2e-encryption")]
|
||||
use matrix_sdk_crypto::store::{DynCryptoStore, IntoCryptoStore};
|
||||
pub use matrix_sdk_store_encryption::Error as StoreEncryptionError;
|
||||
@@ -144,7 +143,7 @@ pub(crate) struct Store {
|
||||
session_meta: Arc<OnceCell<SessionMeta>>,
|
||||
/// The current sync token that should be used for the next sync call.
|
||||
pub(super) sync_token: Arc<RwLock<Option<String>>>,
|
||||
rooms: Arc<DashMap<OwnedRoomId, Room>>,
|
||||
rooms: Arc<StdRwLock<BTreeMap<OwnedRoomId, Room>>>,
|
||||
/// A lock to synchronize access to the store, such that data by the sync is
|
||||
/// never overwritten. The sync processing is supposed to use write access,
|
||||
/// such that only it is currently accessing the store overall. Other things
|
||||
@@ -179,7 +178,7 @@ impl Store {
|
||||
pub async fn set_session_meta(&self, session_meta: SessionMeta) -> Result<()> {
|
||||
for info in self.inner.get_room_infos().await? {
|
||||
let room = Room::restore(&session_meta.user_id, self.inner.clone(), info);
|
||||
self.rooms.insert(room.room_id().to_owned(), room);
|
||||
self.rooms.write().unwrap().insert(room.room_id().to_owned(), room);
|
||||
}
|
||||
|
||||
let token =
|
||||
@@ -198,21 +197,23 @@ impl Store {
|
||||
|
||||
/// Get all the rooms this store knows about.
|
||||
pub fn get_rooms(&self) -> Vec<Room> {
|
||||
self.rooms.iter().filter_map(|r| self.get_room(r.key())).collect()
|
||||
self.rooms.read().unwrap().keys().filter_map(|id| self.get_room(id)).collect()
|
||||
}
|
||||
|
||||
/// Get all the rooms this store knows about, filtered by state.
|
||||
pub fn get_rooms_filtered(&self, filter: RoomStateFilter) -> Vec<Room> {
|
||||
self.rooms
|
||||
.read()
|
||||
.unwrap()
|
||||
.iter()
|
||||
.filter(|r| filter.matches(r.state()))
|
||||
.filter_map(|r| self.get_room(r.key()))
|
||||
.filter(|(_, r)| filter.matches(r.state()))
|
||||
.filter_map(|(id, _)| self.get_room(id))
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// Get the room with the given room id.
|
||||
pub fn get_room(&self, room_id: &RoomId) -> Option<Room> {
|
||||
self.rooms.get(room_id).map(|r| r.clone())
|
||||
self.rooms.read().unwrap().get(room_id).cloned()
|
||||
}
|
||||
|
||||
/// Lookup the Room for the given RoomId, or create one, if it didn't exist
|
||||
@@ -222,6 +223,8 @@ impl Store {
|
||||
&self.session_meta.get().expect("Creating room while not being logged in").user_id;
|
||||
|
||||
self.rooms
|
||||
.write()
|
||||
.unwrap()
|
||||
.entry(room_id.to_owned())
|
||||
.or_insert_with(|| Room::new(user_id, self.inner.clone(), room_id, room_type))
|
||||
.clone()
|
||||
|
||||
Reference in New Issue
Block a user