| 1 | use std::collections::HashMap; |
| 2 | |
| 3 | use futures_util::{ |
| 4 | FutureExt, |
| 5 | future::{Shared, WeakShared}, |
| 6 | }; |
| 7 | use matrix_sdk::Client; |
| 8 | use ruma::OwnedRoomId; |
| 9 | |
| 10 | use super::executor::EngineBoxFuture; |
| 11 | use crate::Error; |
| 12 | |
| 13 | type Flight = Shared<EngineBoxFuture<'static, Result<(), Error>>>; |
| 14 | |
| 15 | /// one warm-up per room at a time, later asks join the one in flight |
| 16 | #[derive(Default)] |
| 17 | pub(super) struct Preparations( |
| 18 | HashMap<OwnedRoomId, WeakShared<EngineBoxFuture<'static, Result<(), Error>>>>, |
| 19 | ); |
| 20 | |
| 21 | impl Preparations { |
| 22 | pub(super) fn join(&mut self, client: &Client, room_id: OwnedRoomId) -> Flight { |
| 23 | self.0.retain(|_, flight| flight.upgrade().is_some()); |
| 24 | if let Some(flight) = self.0.get(&room_id).and_then(WeakShared::upgrade) { |
| 25 | return flight; |
| 26 | } |
| 27 | let work: EngineBoxFuture<'static, _> = Box::pin(prepare(client.clone(), room_id.clone())); |
| 28 | let flight = work.shared(); |
| 29 | if let Some(weak) = flight.downgrade() { |
| 30 | self.0.insert(room_id, weak); |
| 31 | } |
| 32 | flight |
| 33 | } |
| 34 | } |
| 35 | |
| 36 | /// the sdk's own pre-send work, run early. a send landing mid-share waits on the |
| 37 | /// sdk's per-room share lock instead of starting another (forked sdk, see design-decisions) |
| 38 | async fn prepare(client: Client, room_id: OwnedRoomId) -> Result<(), Error> { |
| 39 | let room = client.get_room(&room_id).ok_or_else(Error::room_unknown)?; |
| 40 | room.prepare_encryption() |
| 41 | .await |
| 42 | .map_err(|source| Error::from_sdk("preparing room encryption failed", source)) |
| 43 | } |