diff --git a/crates/matrix-sdk-ui/src/timeline/mod.rs b/crates/matrix-sdk-ui/src/timeline/mod.rs index 98333e99e..cec3b3575 100644 --- a/crates/matrix-sdk-ui/src/timeline/mod.rs +++ b/crates/matrix-sdk-ui/src/timeline/mod.rs @@ -484,7 +484,7 @@ impl Timeline { let event_content = AnyMessageLikeEventContent::Reaction(ReactionEventContent::from(annotation.clone())); - let response = room.send(event_content, Some(&txn_id)).await; + let response = room.send(event_content).with_transaction_id(&txn_id).await; match response { Ok(response) => { diff --git a/crates/matrix-sdk-ui/src/timeline/queue.rs b/crates/matrix-sdk-ui/src/timeline/queue.rs index 22e8413ad..c3c70f3cd 100644 --- a/crates/matrix-sdk-ui/src/timeline/queue.rs +++ b/crates/matrix-sdk-ui/src/timeline/queue.rs @@ -200,7 +200,7 @@ impl SendMessageTask { debug!("Spawning message-sending task"); let txn_id = msg.txn_id.clone(); let join_handle = spawn(async move { - let result = room.send(msg.content, Some(&msg.txn_id)).await; + let result = room.send(msg.content).with_transaction_id(&msg.txn_id).await; let (room, send_state) = match result { Ok(response) => (Some(room), EventSendState::Sent { event_id: response.event_id }), Err(error) => (None, EventSendState::SendingFailed { error: Arc::new(error) }), diff --git a/crates/matrix-sdk/src/encryption/identities/users.rs b/crates/matrix-sdk/src/encryption/identities/users.rs index 73749b56a..7c57d47c4 100644 --- a/crates/matrix-sdk/src/encryption/identities/users.rs +++ b/crates/matrix-sdk/src/encryption/identities/users.rs @@ -514,7 +514,7 @@ impl OtherUserIdentity { }; let response = room - .send(RoomMessageEventContent::new(MessageType::VerificationRequest(content)), None) + .send(RoomMessageEventContent::new(MessageType::VerificationRequest(content))) .await?; let verification = diff --git a/crates/matrix-sdk/src/encryption/mod.rs b/crates/matrix-sdk/src/encryption/mod.rs index 0880bad01..b9ce737b9 100644 --- a/crates/matrix-sdk/src/encryption/mod.rs +++ b/crates/matrix-sdk/src/encryption/mod.rs @@ -169,7 +169,7 @@ impl Client { /// let mut reader = std::io::Cursor::new(b"Hello, world!"); /// let encrypted_file = client.prepare_encrypted_file(&mime::TEXT_PLAIN, &mut reader).await?; /// - /// room.send(CustomEventContent { encrypted_file }, None).await?; + /// room.send(CustomEventContent { encrypted_file }).await?; /// # anyhow::Ok(()) }; /// ``` pub fn prepare_encrypted_file<'a, R: Read + ?Sized + 'a>( @@ -339,7 +339,8 @@ impl Client { self.get_room(room_id) .expect("Can't send a message to a room that isn't known to the store") - .send(content, Some(txn_id)) + .send(content) + .with_transaction_id(txn_id) .await } @@ -1237,11 +1238,9 @@ mod tests { let event_id = event_id!("$1:example.org"); let reaction = ReactionEventContent::new(Annotation::new(event_id.into(), "🐈".to_owned())); - room.send(reaction, None).await.expect("Sending the reaction should not fail"); + room.send(reaction).await.expect("Sending the reaction should not fail"); - room.send_raw(json!({}), "m.reaction", None) - .await - .expect("Sending the reaction should not fail"); + room.send_raw(json!({}), "m.reaction").await.expect("Sending the reaction should not fail"); } #[async_test] diff --git a/crates/matrix-sdk/src/matrix_auth/login_builder.rs b/crates/matrix-sdk/src/matrix_auth/login_builder.rs index 95e6aa5dc..4d509cf68 100644 --- a/crates/matrix-sdk/src/matrix_auth/login_builder.rs +++ b/crates/matrix-sdk/src/matrix_auth/login_builder.rs @@ -14,7 +14,9 @@ // limitations under the License. #![cfg_attr(not(target_arch = "wasm32"), deny(clippy::future_not_send))] -use std::future::{Future, IntoFuture}; +#[cfg(feature = "sso-login")] +use std::future::Future; +use std::future::IntoFuture; use matrix_sdk_common::boxed_into_future; use ruma::{ diff --git a/crates/matrix-sdk/src/room/futures.rs b/crates/matrix-sdk/src/room/futures.rs index f40a92af3..65532f2c6 100644 --- a/crates/matrix-sdk/src/room/futures.rs +++ b/crates/matrix-sdk/src/room/futures.rs @@ -23,8 +23,13 @@ use std::io::Cursor; use eyeball::SharedObservable; use matrix_sdk_common::boxed_into_future; use mime::Mime; -use ruma::api::client::message::send_message_event; -use tracing::{Instrument, Span}; +#[cfg(doc)] +use ruma::events::{MessageLikeUnsigned, SyncMessageLikeEvent}; +use ruma::{ + api::client::message::send_message_event, assign, events::MessageLikeEventContent, serde::Raw, + OwnedTransactionId, TransactionId, +}; +use tracing::{debug, Instrument, Span}; use super::Room; use crate::{attachment::AttachmentConfig, Result, TransmissionProgress}; @@ -34,6 +39,167 @@ use crate::{ error::ImageError, }; +/// Future returned by [`Room::send`]. +#[allow(missing_debug_implementations)] +pub struct SendMessageLikeEvent<'a> { + room: &'a Room, + event_type: String, + content: serde_json::Result, + transaction_id: Option, +} + +impl<'a> SendMessageLikeEvent<'a> { + pub(crate) fn new(room: &'a Room, content: impl MessageLikeEventContent) -> Self { + let event_type = content.event_type().to_string(); + let content = serde_json::to_value(&content); + Self { room, event_type, content, transaction_id: None } + } + + /// Set a transaction ID for this event. + /// + /// Since sending message-like events always requires a transaction ID, one + /// is generated if this method is not called. + /// + /// The transaction ID is a locally-unique ID describing a message + /// transaction with the homeserver. + /// + /// * On the sending side, this field is used for re-trying earlier failed + /// transactions. Subsequent messages *must never* re-use an earlier + /// transaction ID. + /// * On the receiving side, the field is used for recognizing our own + /// messages when they arrive down the sync: the server includes the ID in + /// the [`MessageLikeUnsigned`] field `transaction_id` of the + /// corresponding [`SyncMessageLikeEvent`], but only for the *sending* + /// device. Other devices will not see it. This is then used to ignore + /// events sent by our own device and/or to implement local echo. + pub fn with_transaction_id(mut self, txn_id: &TransactionId) -> Self { + self.transaction_id = Some(txn_id.to_owned()); + self + } +} + +impl<'a> IntoFuture for SendMessageLikeEvent<'a> { + type Output = Result; + boxed_into_future!(extra_bounds: 'a); + + fn into_future(self) -> Self::IntoFuture { + let Self { room, event_type, content, transaction_id } = self; + Box::pin(async move { + let content = content?; + assign!(room.send_raw(content, &event_type), { transaction_id }).await + }) + } +} + +/// Future returned by [`Room::send_raw`]. +#[allow(missing_debug_implementations)] +pub struct SendRawMessageLikeEvent<'a> { + room: &'a Room, + event_type: &'a str, + content: serde_json::Value, + tracing_span: Span, + transaction_id: Option, +} + +impl<'a> SendRawMessageLikeEvent<'a> { + pub(crate) fn new(room: &'a Room, event_type: &'a str, content: serde_json::Value) -> Self { + Self { room, event_type, content, tracing_span: Span::current(), transaction_id: None } + } + + /// Set a transaction ID for this event. + /// + /// Since sending message-like events always requires a transaction ID, one + /// is generated if this method is not called. + /// + /// * On the sending side, this field is used for re-trying earlier failed + /// transactions. Subsequent messages *must never* re-use an earlier + /// transaction ID. + /// * On the receiving side, the field is used for recognizing our own + /// messages when they arrive down the sync: the server includes the ID in + /// the [`MessageLikeUnsigned`] field `transaction_id` of the + /// corresponding [`SyncMessageLikeEvent`], but only for the *sending* + /// device. Other devices will not see it. This is then used to ignore + /// events sent by our own device and/or to implement local echo. + pub fn with_transaction_id(mut self, txn_id: &TransactionId) -> Self { + self.transaction_id = Some(txn_id.to_owned()); + self + } +} + +impl<'a> IntoFuture for SendRawMessageLikeEvent<'a> { + type Output = Result; + boxed_into_future!(extra_bounds: 'a); + + fn into_future(self) -> Self::IntoFuture { + let Self { room, event_type, content, tracing_span, transaction_id } = self; + let fut = async move { + room.ensure_room_joined()?; + + let txn_id = transaction_id.unwrap_or_else(TransactionId::new); + tracing::Span::current().record("transaction_id", tracing::field::debug(&txn_id)); + + #[cfg(not(feature = "e2e-encryption"))] + let content = { + debug!("Sending plaintext event to room because we don't have encryption support."); + Raw::new(&content)?.cast() + }; + + #[cfg(feature = "e2e-encryption")] + let (content, event_type) = if room.is_encrypted().await? { + tracing::Span::current().record("encrypted", tracing::field::debug(&txn_id)); + // Reactions are currently famously not encrypted, skip encrypting + // them until they are. + if event_type == "m.reaction" { + debug!("Sending plaintext event because of the event type."); + (Raw::new(&content)?.cast(), event_type) + } else { + debug!( + room_id = room.room_id().as_str(), + "Sending encrypted event because the room is encrypted.", + ); + + if !room.are_members_synced() { + room.sync_members().await?; + } + + // Query keys in case we don't have them for newly synced members. + // + // Note we do it all the time, because we might have sync'd members before + // sending a message (so didn't enter the above branch), but + // could have not query their keys ever. + room.query_keys_for_untracked_users().await?; + + room.preshare_room_key().await?; + + let olm = room.client.olm_machine().await; + let olm = olm.as_ref().expect("Olm machine wasn't started"); + + let encrypted_content = + olm.encrypt_room_event_raw(room.room_id(), content, event_type).await?; + + (encrypted_content.cast(), "m.room.encrypted") + } + } else { + debug!("Sending plaintext event because the room is NOT encrypted.",); + + (Raw::new(&content)?.cast(), event_type) + }; + + let request = send_message_event::v3::Request::new_raw( + room.room_id().to_owned(), + txn_id, + event_type.into(), + content, + ); + + let response = room.client.send(request, None).await?; + Ok(response) + }; + + Box::pin(fut.instrument(tracing_span)) + } +} + /// Future returned by [`Room::send_attachment`]. #[allow(missing_debug_implementations)] pub struct SendAttachment<'a> { diff --git a/crates/matrix-sdk/src/room/mod.rs b/crates/matrix-sdk/src/room/mod.rs index 3c38b660c..f03c66db8 100644 --- a/crates/matrix-sdk/src/room/mod.rs +++ b/crates/matrix-sdk/src/room/mod.rs @@ -69,7 +69,7 @@ use thiserror::Error; use tokio::sync::broadcast; use tracing::{debug, instrument, warn}; -use self::futures::SendAttachment; +use self::futures::{SendAttachment, SendMessageLikeEvent, SendRawMessageLikeEvent}; use crate::{ attachment::AttachmentConfig, error::WrongRoomState, @@ -1291,23 +1291,14 @@ impl Room { /// **Note**: If you just want to send a custom JSON payload to a room, you /// can use the [`send_raw()`][Self::send_raw] method for that. /// + /// If you want to set a transaction ID for the event, use + /// [`.with_transaction_id()`][SendMessageLikeEvent::with_transaction_id] + /// on the returned value before `.await`ing it. + /// /// # Arguments /// /// * `content` - The content of the message event. /// - /// * `txn_id` - A locally-unique ID describing a message transaction with - /// the homeserver. Unless you're doing something special, you can pass in - /// `None` which will create a suitable one for you automatically. - /// * On the sending side, this field is used for re-trying earlier - /// failed transactions. Subsequent messages *must never* re-use an - /// earlier transaction ID. - /// * On the receiving side, the field is used for recognizing our own - /// messages when they arrive down the sync: the server includes the - /// ID in the [`MessageLikeUnsigned`] field [`transaction_id`] of the - /// corresponding [`SyncMessageLikeEvent`], but only for the *sending* - /// device. Other devices will not see it. This is then used to ignore - /// events sent by our own device and/or to implement local echo. - /// /// # Examples /// /// ```no_run @@ -1332,7 +1323,7 @@ impl Room { /// let txn_id = TransactionId::new(); /// /// if let Some(room) = client.get_room(&room_id) { - /// room.send(content, Some(&txn_id)).await?; + /// room.send(content).with_transaction_id(&txn_id).await?; /// } /// /// // Custom events work too: @@ -1352,26 +1343,14 @@ impl Room { /// MilliSecondsSinceUnixEpoch(now.0 + uint!(30_000)) /// }, /// }; - /// let txn_id = TransactionId::new(); /// /// if let Some(room) = client.get_room(&room_id) { - /// room.send(content, Some(&txn_id)).await?; + /// room.send(content).await?; /// } /// # anyhow::Ok(()) }; /// ``` - /// - /// [`SyncMessageLikeEvent`]: ruma::events::SyncMessageLikeEvent - /// [`MessageLikeUnsigned`]: ruma::events::MessageLikeUnsigned - /// [`transaction_id`]: ruma::events::MessageLikeUnsigned#structfield.transaction_id - pub async fn send( - &self, - content: impl MessageLikeEventContent, - txn_id: Option<&TransactionId>, - ) -> Result { - let event_type = content.event_type().to_string(); - let content = serde_json::to_value(&content)?; - - self.send_raw(content, &event_type, txn_id).await + pub fn send(&self, content: impl MessageLikeEventContent) -> SendMessageLikeEvent<'_> { + SendMessageLikeEvent::new(self, content) } /// Run /keys/query requests for all the non-tracked users. @@ -1419,25 +1398,16 @@ impl Room { /// allows sending custom JSON payloads, e.g. constructed using the /// [`serde_json::json!()`] macro. /// + /// If you want to set a transaction ID for the event, use + /// [`.with_transaction_id()`][SendRawMessageLikeEvent::with_transaction_id] + /// on the returned value before `.await`ing it. + /// /// # Arguments /// /// * `content` - The content of the event as a json `Value`. /// /// * `event_type` - The type of the event. /// - /// * `txn_id` - A locally-unique ID describing a message transaction with - /// the homeserver. Unless you're doing something special, you can pass in - /// `None` which will create a suitable one for you automatically. - /// * On the sending side, this field is used for re-trying earlier - /// failed transactions. Subsequent messages *must never* re-use an - /// earlier transaction ID. - /// * On the receiving side, the field is used for recognizing our own - /// messages when they arrive down the sync: the server includes the - /// ID in the [`StateUnsigned`] field [`transaction_id`] of the - /// corresponding [`SyncMessageLikeEvent`], but only for the *sending* - /// device. Other devices will not see it. This is then used to ignore - /// events sent by our own device and/or to implement local echo. - /// /// # Examples /// /// ```no_run @@ -1456,82 +1426,17 @@ impl Room { /// }); /// /// if let Some(room) = client.get_room(&room_id) { - /// room.send_raw(content, "m.room.message", None).await?; + /// room.send_raw(content, "m.room.message").await?; /// } /// # anyhow::Ok(()) }; /// ``` - /// - /// [`SyncMessageLikeEvent`]: ruma::events::SyncMessageLikeEvent - /// [`StateUnsigned`]: ruma::events::StateUnsigned - /// [`transaction_id`]: ruma::events::StateUnsigned#structfield.transaction_id #[instrument(skip_all, fields(event_type, room_id = ?self.room_id(), transaction_id, encrypted))] - pub async fn send_raw( - &self, + pub fn send_raw<'a>( + &'a self, content: serde_json::Value, - event_type: &str, - txn_id: Option<&TransactionId>, - ) -> Result { - self.ensure_room_joined()?; - - let txn_id: OwnedTransactionId = txn_id.map_or_else(TransactionId::new, ToOwned::to_owned); - tracing::Span::current().record("transaction_id", tracing::field::debug(&txn_id)); - - #[cfg(not(feature = "e2e-encryption"))] - let content = { - debug!("Sending plaintext event to room because we don't have encryption support."); - Raw::new(&content)?.cast() - }; - - #[cfg(feature = "e2e-encryption")] - let (content, event_type) = if self.is_encrypted().await? { - tracing::Span::current().record("encrypted", tracing::field::debug(&txn_id)); - // Reactions are currently famously not encrypted, skip encrypting - // them until they are. - if event_type == "m.reaction" { - debug!("Sending plaintext event because of the event type."); - (Raw::new(&content)?.cast(), event_type) - } else { - debug!( - room_id = self.room_id().as_str(), - "Sending encrypted event because the room is encrypted.", - ); - - if !self.are_members_synced() { - self.sync_members().await?; - } - - // Query keys in case we don't have them for newly synced members. - // - // Note we do it all the time, because we might have sync'd members before - // sending a message (so didn't enter the above branch), but - // could have not query their keys ever. - self.query_keys_for_untracked_users().await?; - - self.preshare_room_key().await?; - - let olm = self.client.olm_machine().await; - let olm = olm.as_ref().expect("Olm machine wasn't started"); - - let encrypted_content = - olm.encrypt_room_event_raw(self.room_id(), content, event_type).await?; - - (encrypted_content.cast(), "m.room.encrypted") - } - } else { - debug!("Sending plaintext event because the room is NOT encrypted.",); - - (Raw::new(&content)?.cast(), event_type) - }; - - let request = send_message_event::v3::Request::new_raw( - self.room_id().to_owned(), - txn_id, - event_type.into(), - content, - ); - - let response = self.client.send(request, None).await?; - Ok(response) + event_type: &'a str, + ) -> SendRawMessageLikeEvent<'a> { + SendRawMessageLikeEvent::new(self, event_type, content) } /// Send an attachment to this room. @@ -1665,7 +1570,11 @@ impl Room { ) .await?; - self.send(RoomMessageEventContent::new(content), config.txn_id.as_deref()).await + let mut fut = self.send(RoomMessageEventContent::new(content)); + if let Some(txn_id) = &config.txn_id { + fut = fut.with_transaction_id(txn_id); + } + fut.await } /// Update the power levels of a select set of users of this room. diff --git a/crates/matrix-sdk/src/widget/matrix.rs b/crates/matrix-sdk/src/widget/matrix.rs index 0dafe0116..8faed6a85 100644 --- a/crates/matrix-sdk/src/widget/matrix.rs +++ b/crates/matrix-sdk/src/widget/matrix.rs @@ -116,7 +116,7 @@ impl MatrixDriver { let type_str = event_type.to_string(); Ok(match state_key { Some(key) => self.room.send_state_event_raw(content, &type_str, &key).await?.event_id, - None => self.room.send_raw(content, &type_str, None).await?.event_id, + None => self.room.send_raw(content, &type_str).await?.event_id, }) } diff --git a/crates/matrix-sdk/tests/integration/room/joined.rs b/crates/matrix-sdk/tests/integration/room/joined.rs index 024d6f58b..285a838db 100644 --- a/crates/matrix-sdk/tests/integration/room/joined.rs +++ b/crates/matrix-sdk/tests/integration/room/joined.rs @@ -273,7 +273,7 @@ async fn room_message_send() { let content = RoomMessageEventContent::text_plain("Hello world"); let txn_id = TransactionId::new(); - let response = room.send(content, Some(&txn_id)).await.unwrap(); + let response = room.send(content).with_transaction_id(&txn_id).await.unwrap(); assert_eq!(event_id!("$h29iv0s8:example.com"), response.event_id) } diff --git a/examples/command_bot/src/main.rs b/examples/command_bot/src/main.rs index b15dd3915..f51b35e27 100644 --- a/examples/command_bot/src/main.rs +++ b/examples/command_bot/src/main.rs @@ -22,9 +22,7 @@ async fn on_room_message(event: OriginalSyncRoomMessageEvent, room: Room) { println!("sending"); // send our message to the room we found the "!party" command in - // the last parameter is an optional transaction id which we don't - // care about. - room.send(content, None).await.unwrap(); + room.send(content).await.unwrap(); println!("message sent"); } diff --git a/examples/custom_events/src/main.rs b/examples/custom_events/src/main.rs index 38b578b59..1eca0a9d7 100644 --- a/examples/custom_events/src/main.rs +++ b/examples/custom_events/src/main.rs @@ -59,7 +59,7 @@ async fn on_regular_room_message(event: OriginalSyncRoomMessageEvent, room: Room let content = PingEventContent {}; println!("sending ping"); - room.send(content, None).await.unwrap(); + room.send(content).await.unwrap(); println!("ping sent"); } } @@ -75,7 +75,7 @@ async fn on_ping_event(event: SyncPingEvent, room: Room) { // Send an ack with the event_id of the ping, as our 'protocol' demands let content = AckEventContent { ping_id: event_id }; println!("sending ack"); - room.send(content, None).await.unwrap(); + room.send(content).await.unwrap(); println!("ack sent"); } diff --git a/examples/getting_started/src/main.rs b/examples/getting_started/src/main.rs index abacb8b90..47efa0344 100644 --- a/examples/getting_started/src/main.rs +++ b/examples/getting_started/src/main.rs @@ -161,9 +161,7 @@ async fn on_room_message(event: OriginalSyncRoomMessageEvent, room: Room) { println!("sending"); // send our message to the room we found the "!party" command in - // the last parameter is an optional transaction id which we don't - // care about. - room.send(content, None).await.unwrap(); + room.send(content).await.unwrap(); println!("message sent"); } diff --git a/testing/matrix-sdk-integration-testing/src/tests/e2ee.rs b/testing/matrix-sdk-integration-testing/src/tests/e2ee.rs index 0c036b43d..f0e581b3c 100644 --- a/testing/matrix-sdk-integration-testing/src/tests/e2ee.rs +++ b/testing/matrix-sdk-integration-testing/src/tests/e2ee.rs @@ -324,7 +324,7 @@ async fn test_encryption_missing_member_keys() -> Result<()> { let bob_room = bob.get_room(alice_room.room_id()).unwrap(); let message = "Hello world!"; let bob_message_content = Arc::new(Mutex::new(message)); - bob_room.send(RoomMessageEventContent::text_plain(message), None).await?; + bob_room.send(RoomMessageEventContent::text_plain(message)).await?; warn!("bob is done sending the message"); // Alice was in the room when Bob sent the message, so they'll see it. @@ -383,7 +383,7 @@ async fn test_encryption_missing_member_keys() -> Result<()> { let bob_room = bob.get_room(alice_room.room_id()).unwrap(); let message = "Wassup"; *bob_message_content.lock().unwrap() = message; - bob_room.send(RoomMessageEventContent::text_plain(message), None).await?; + bob_room.send(RoomMessageEventContent::text_plain(message)).await?; warn!("bob is done sending another message"); { @@ -442,7 +442,7 @@ async fn test_failed_members_response() -> Result<()> { let bob_room = bob.get_room(alice_room.room_id()).unwrap(); let message = "Hello world!"; let bob_message_content = Arc::new(Mutex::new(message)); - bob_room.send(RoomMessageEventContent::text_plain(message), None).await?; + bob_room.send(RoomMessageEventContent::text_plain(message)).await?; warn!("bob is done sending the message"); // Alice sees the message. diff --git a/testing/matrix-sdk-integration-testing/src/tests/sliding_sync/notification_client.rs b/testing/matrix-sdk-integration-testing/src/tests/sliding_sync/notification_client.rs index 3131d4ca0..9124a8c46 100644 --- a/testing/matrix-sdk-integration-testing/src/tests/sliding_sync/notification_client.rs +++ b/testing/matrix-sdk-integration-testing/src/tests/sliding_sync/notification_client.rs @@ -141,7 +141,7 @@ async fn test_notification() -> Result<()> { bob.get_room(alice_room.room_id()).unwrap().join().await?; // Now Alice sends a message to Bob. - alice_room.send(RoomMessageEventContent::text_plain("Hello world!"), None).await?; + alice_room.send(RoomMessageEventContent::text_plain("Hello world!")).await?; // In this sync, bob receives the message from Alice. let bob_response = bob.sync_once(SyncSettings::default().token(sync_token)).await?;