Dima/matrix-rust-sdk
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

send queue: rename AbortSendHandle to SendHandle

Browse Source
This commit is contained in:
Benjamin Bouvier
2024-06-26 15:10:26 +02:00
parent d41af396cc
commit 394effbc72
9 changed files with 39 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bindings/matrix-sdk-ffi/src/timeline/mod.rs
+5 -5
View File
@@ -232,9 +232,9 @@ impl Timeline {
pub async fn send(
self: Arc<Self>,
msg: Arc<RoomMessageEventContentWithoutRelation>,
) -> Result<Arc<AbortSendHandle>, ClientError> {
) -> Result<Arc<SendHandle>, ClientError> {
match self.inner.send((*msg).to_owned().with_relation(None).into()).await {
Ok(handle) => Ok(Arc::new(AbortSendHandle { inner: Mutex::new(Some(handle)) })),
Ok(handle) => Ok(Arc::new(SendHandle { inner: Mutex::new(Some(handle)) })),
Err(err) => {
error!("error when sending a message: {err}");
Err(anyhow::anyhow!(err).into())
@@ -647,12 +647,12 @@ impl Timeline {
}
#[derive(uniffi::Object)]
pub struct AbortSendHandle {
inner: Mutex<Option<matrix_sdk::send_queue::AbortSendHandle>>,
pub struct SendHandle {
inner: Mutex<Option<matrix_sdk::send_queue::SendHandle>>,
}
#[uniffi::export(async_runtime = "tokio")]
impl AbortSendHandle {
impl SendHandle {
/// Try to abort the sending of the current event.
///
/// If this returns `true`, then the sending could be aborted, because the
crates/matrix-sdk-ui/src/timeline/builder.rs
+1 -1
View File
@@ -426,7 +426,7 @@ async fn handle_local_echo(echo: LocalEcho, timeline: &TimelineInner) {
.handle_local_event(
echo.transaction_id.clone(),
TimelineEventKind::Message { content, relations: Default::default() },
Some(echo.abort_handle),
Some(echo.send_handle),
)
.await;
crates/matrix-sdk-ui/src/timeline/event_handler.rs
+5 -6
View File
@@ -18,8 +18,7 @@ use as_variant::as_variant;
use eyeball_im::{ObservableVectorTransaction, ObservableVectorTransactionEntry};
use indexmap::{map::Entry, IndexMap};
use matrix_sdk::{
crypto::types::events::UtdCause, deserialized_responses::EncryptionInfo,
send_queue::AbortSendHandle,
crypto::types::events::UtdCause, deserialized_responses::EncryptionInfo, send_queue::SendHandle,
};
use ruma::{
events::{
@@ -72,8 +71,8 @@ pub(super) enum Flow {
/// The transaction id we've used in requests associated to this event.
txn_id: OwnedTransactionId,
/// A handle to abort sending this event.
abort_handle: Option<AbortSendHandle>,
/// A handle to manipulate this event.
send_handle: Option<SendHandle>,
},
/// The event has been received from a remote source (sync, pagination,
@@ -875,10 +874,10 @@ impl<'a, 'o> TimelineEventHandler<'a, 'o> {
let mut reactions = self.pending_reactions().unwrap_or_default();
let kind: EventTimelineItemKind = match &self.ctx.flow {
Flow::Local { txn_id, abort_handle } => LocalEventTimelineItem {
Flow::Local { txn_id, send_handle } => LocalEventTimelineItem {
send_state: EventSendState::NotSentYet,
transaction_id: txn_id.to_owned(),
abort_handle: abort_handle.clone(),
send_handle: send_handle.clone(),
}
.into(),
crates/matrix-sdk-ui/src/timeline/event_item/local.rs
+3 -3
View File
@@ -15,7 +15,7 @@
use std::sync::Arc;
use as_variant::as_variant;
use matrix_sdk::{send_queue::AbortSendHandle, Error};
use matrix_sdk::{send_queue::SendHandle, Error};
use ruma::{EventId, OwnedEventId, OwnedTransactionId};
/// An item for an event that was created locally and not yet echoed back by
@@ -26,8 +26,8 @@ pub(in crate::timeline) struct LocalEventTimelineItem {
pub send_state: EventSendState,
/// The transaction ID.
pub transaction_id: OwnedTransactionId,
/// A handle to abort sending this event, if possible.
pub abort_handle: Option<AbortSendHandle>,
/// A handle to manipulate this event before it is sent, if possible.
pub send_handle: Option<SendHandle>,
}
impl LocalEventTimelineItem {
crates/matrix-sdk-ui/src/timeline/inner/mod.rs
+3 -3
View File
@@ -27,7 +27,7 @@ use matrix_sdk::crypto::OlmMachine;
use matrix_sdk::{
deserialized_responses::SyncTimelineEvent,
event_cache::{paginator::Paginator, RoomEventCache},
send_queue::AbortSendHandle,
send_queue::SendHandle,
Result, Room,
};
#[cfg(test)]
@@ -646,13 +646,13 @@ impl<P: RoomDataProvider> TimelineInner<P> {
&self,
txn_id: OwnedTransactionId,
content: TimelineEventKind,
abort_handle: Option<AbortSendHandle>,
send_handle: Option<SendHandle>,
) {
let sender = self.room_data_provider.own_user_id().to_owned();
let profile = self.room_data_provider.profile_from_user_id(&sender).await;
let mut state = self.state.write().await;
state.handle_local_event(sender, profile, txn_id, abort_handle, content).await;
state.handle_local_event(sender, profile, txn_id, send_handle, content).await;
}
/// Update the send state of a local event represented by a transaction ID.
crates/matrix-sdk-ui/src/timeline/inner/state.rs
+3 -3
View File
@@ -16,7 +16,7 @@ use std::{collections::VecDeque, future::Future, sync::Arc};
use eyeball_im::{ObservableVector, ObservableVectorTransaction, ObservableVectorTransactionEntry};
use indexmap::IndexMap;
use matrix_sdk::{deserialized_responses::SyncTimelineEvent, send_queue::AbortSendHandle};
use matrix_sdk::{deserialized_responses::SyncTimelineEvent, send_queue::SendHandle};
use matrix_sdk_base::deserialized_responses::TimelineEvent;
#[cfg(test)]
use ruma::events::receipt::ReceiptEventContent;
@@ -163,7 +163,7 @@ impl TimelineInnerState {
own_user_id: OwnedUserId,
own_profile: Option<Profile>,
txn_id: OwnedTransactionId,
abort_handle: Option<AbortSendHandle>,
send_handle: Option<SendHandle>,
content: TimelineEventKind,
) {
let ctx = TimelineEventContext {
@@ -176,7 +176,7 @@ impl TimelineInnerState {
read_receipts: Default::default(),
// An event sent by ourself is never matched against push rules.
is_highlighted: false,
flow: Flow::Local { txn_id, abort_handle },
flow: Flow::Local { txn_id, send_handle },
};
let mut txn = self.transaction();
crates/matrix-sdk-ui/src/timeline/mod.rs
+3 -3
View File
@@ -27,7 +27,7 @@ use matrix_sdk::{
event_handler::EventHandlerHandle,
executor::JoinHandle,
room::{Receipts, Room},
send_queue::{AbortSendHandle, RoomSendQueueError},
send_queue::{RoomSendQueueError, SendHandle},
Client, Result,
};
use matrix_sdk_base::RoomState;
@@ -326,7 +326,7 @@ impl Timeline {
pub async fn send(
&self,
content: AnyMessageLikeEventContent,
) -> Result<AbortSendHandle, RoomSendQueueError> {
) -> Result<SendHandle, RoomSendQueueError> {
self.room().send_queue().send(content).await
}
@@ -701,7 +701,7 @@ impl Timeline {
) -> Result<bool, RedactEventError> {
match &event.kind {
EventTimelineItemKind::Local(local) => {
if let Some(handle) = local.abort_handle.clone() {
if let Some(handle) = local.send_handle.clone() {
Ok(handle.abort().await.map_err(RedactEventError::RoomQueueError)?)
} else {
// No abort handle; theoretically unreachable for regular usage of the
crates/matrix-sdk/src/send_queue.rs
+10 -14
View File
@@ -316,7 +316,7 @@ impl RoomSendQueue {
&self,
content: Raw<AnyMessageLikeEventContent>,
event_type: String,
) -> Result<AbortSendHandle, RoomSendQueueError> {
) -> Result<SendHandle, RoomSendQueueError> {
let Some(room) = self.inner.room.get() else {
return Err(RoomSendQueueError::RoomDisappeared);
};
@@ -334,14 +334,11 @@ impl RoomSendQueue {
let _ = self.inner.updates.send(RoomSendQueueUpdate::NewLocalEvent(LocalEcho {
transaction_id: transaction_id.clone(),
serialized_event: content,
abort_handle: AbortSendHandle {
room: self.clone(),
transaction_id: transaction_id.clone(),
},
send_handle: SendHandle { room: self.clone(), transaction_id: transaction_id.clone() },
is_wedged: false,
}));
Ok(AbortSendHandle { transaction_id, room: self.clone() })
Ok(SendHandle { transaction_id, room: self.clone() })
}
/// Queues an event for sending it to this room.
@@ -360,7 +357,7 @@ impl RoomSendQueue {
pub async fn send(
&self,
content: AnyMessageLikeEventContent,
) -> Result<AbortSendHandle, RoomSendQueueError> {
) -> Result<SendHandle, RoomSendQueueError> {
self.send_raw(
Raw::new(&content).map_err(RoomSendQueueStorageError::JsonSerialization)?,
content.event_type().to_string(),
@@ -383,7 +380,7 @@ impl RoomSendQueue {
.map(|queued| LocalEcho {
transaction_id: queued.transaction_id.clone(),
serialized_event: queued.event,
abort_handle: AbortSendHandle {
send_handle: SendHandle {
room: self.clone(),
transaction_id: queued.transaction_id,
},
@@ -722,8 +719,8 @@ pub struct LocalEcho {
/// Content of the event itself (along with its type) that we are about to
/// send.
pub serialized_event: SerializableEventContent,
/// A handle to abort sending the associated event.
pub abort_handle: AbortSendHandle,
/// A handle to manipulate the sending of the associated event.
pub send_handle: SendHandle,
/// Whether trying to send this local echo failed in the past with an
/// unrecoverable error (see [`SendQueueRoomError::is_recoverable`]).
pub is_wedged: bool,
@@ -806,15 +803,14 @@ pub enum RoomSendQueueStorageError {
ClientShuttingDown,
}
/// A way to tentatively abort sending an event that was scheduled to be sent to
/// a room.
/// A handle to manipulate an event that was scheduled to be sent to a room.
#[derive(Clone, Debug)]
pub struct AbortSendHandle {
pub struct SendHandle {
room: RoomSendQueue,
transaction_id: OwnedTransactionId,
}
impl AbortSendHandle {
impl SendHandle {
/// Aborts the sending of the event, if it wasn't sent yet.
///
/// Returns true if the sending could be aborted, false if not (i.e. the
crates/matrix-sdk/tests/integration/send_queue.rs
+6 -7
View File
@@ -54,13 +54,13 @@ fn mock_send_transient_failure() -> Mock {
// A macro to assert on a stream of `RoomSendQueueUpdate`s.
macro_rules! assert_update {
// Check the next stream event is a local echo for a message with the content $body.
// Returns a tuple of (transaction_id, abort_handle).
// Returns a tuple of (transaction_id, send_handle).
($watch:ident => local echo { body = $body:expr }) => {{
assert_let!(
Ok(Ok(RoomSendQueueUpdate::NewLocalEvent(LocalEcho {
serialized_event,
transaction_id: txn,
abort_handle,
send_handle,
// New local echoes should always start as not wedged.
is_wedged: false,
}))) = timeout(Duration::from_secs(1), $watch.recv()).await
@@ -70,7 +70,7 @@ macro_rules! assert_update {
assert_let!(AnyMessageLikeEventContent::RoomMessage(_msg) = content);
assert_eq!(_msg.body(), $body);
(txn, abort_handle)
(txn, send_handle)
}};
// Check the next stream event is a sent event, with optional checks on txn=$txn and
@@ -754,7 +754,7 @@ async fn test_cancellation() {
let local_echo4 = local_echoes.remove(1);
assert_eq!(local_echo4.transaction_id, txn4, "local echoes: {local_echoes:?}");
let handle4 = local_echo4.abort_handle;
let handle4 = local_echo4.send_handle;
assert!(handle4.abort().await.unwrap());
assert_update!(watch => cancelled { txn = txn4 });
@@ -810,8 +810,7 @@ async fn test_abort_after_disable() {
mock_send_transient_failure().expect(3).mount(&server).await;
// One message is queued.
let abort_send_handle =
q.send(RoomMessageEventContent::text_plain("hey there").into()).await.unwrap();
let handle = q.send(RoomMessageEventContent::text_plain("hey there").into()).await.unwrap();
// It is first seen as a local echo,
let (txn, _) = assert_update!(watch => local echo { body = "hey there" });
@@ -828,7 +827,7 @@ async fn test_abort_after_disable() {
assert!(client.send_queue().is_enabled());
// Aborting the sending should work.
assert!(abort_send_handle.abort().await.unwrap());
assert!(handle.abort().await.unwrap());
assert_update!(watch => cancelled { txn = txn });