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

Merge branch 'master' into history-visiblity-session-share

master
Damir Jelić 2021-03-01 20:47:31 +01:00
parent 2e659afd26 9893ddba74
commit e6f6665fa0
43 changed files with 782 additions and 511 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

37
.github/workflows/docs.yml vendored Normal file
View File

@ -0,0 +1,37 @@
name: docs
on:
push:
branches: [master]
pull_request:
jobs:
docs:
name: docs
runs-on: ubuntu-latest
steps:
- name: Checkout repository
uses: actions/checkout@v2
- name: Install stable toolchain
uses: actions-rs/toolchain@v1
with:
profile: minimal
toolchain: nightly
override: true
- name: Build docs
uses: actions-rs/cargo@v1
env:
RUSTDOCFLAGS: "--enable-index-page -Zunstable-options"
with:
command: doc
args: --no-deps --workspace --features docs
- name: Deploy docs
if: ${{ github.event_name == 'push' && github.ref == 'refs/heads/master' }}
uses: peaceiris/actions-gh-pages@v3
with:
github_token: ${{ secrets.GITHUB_TOKEN }}
publish_dir: ./target/doc/

89
.travis.yml
View File

@ -1,89 +0,0 @@
language: rust
rust: stable
addons:
apt:
packages:
- libssl-dev
jobs:
allow_failures:
- os: osx
name: macOS 10.15
include:
- stage: Format
os: linux
before_script:
- rustup component add rustfmt
script:
- cargo fmt --all -- --check
- stage: Clippy
os: linux
before_script:
- rustup component add clippy
script:
- cargo clippy --all-targets -- -D warnings
- stage: Test
os: linux
- os: windows
script:
- cd matrix_sdk
- cargo test --no-default-features --features "messages, native-tls"
- cd ../matrix_sdk_base
- cargo test --no-default-features --features "messages"
- os: osx
- os: linux
name: native-tls build
script:
- cd matrix_sdk
- cargo build --no-default-features --features "native-tls"
- os: linux
name: rustls-tls build
script:
- cd matrix_sdk
- cargo build --no-default-features --features "rustls-tls"
- os: osx
name: macOS 10.15
osx_image: xcode12
- os: linux
name: Coverage
before_script:
- cargo install cargo-tarpaulin
script:
- cargo tarpaulin --ignore-config --exclude-files "matrix_sdk/examples/*,matrix_sdk_common,matrix_sdk_test" --out Xml
after_success:
- bash <(curl -s https://codecov.io/bash)
- os: linux
name: wasm32-unknown-unknown
before_script:
- |
set -e
cargo install wasm-bindgen-cli
rustup target add wasm32-unknown-unknown
wget https://github.com/emscripten-core/emsdk/archive/master.zip
unzip master.zip
./emsdk-master/emsdk install latest
./emsdk-master/emsdk activate latest
script:
- |
set -e
source emsdk-master/emsdk_env.sh
cd matrix_sdk/examples/wasm_command_bot
cargo build --target wasm32-unknown-unknown
cd -
cd matrix_sdk_base
cargo test --target wasm32-unknown-unknown --no-default-features
script:
- cargo build
- cargo test

1
README.md
View File

@ -2,6 +2,7 @@
[![codecov](https://img.shields.io/codecov/c/github/matrix-org/matrix-rust-sdk/master.svg?style=flat-square)](https://codecov.io/gh/matrix-org/matrix-rust-sdk)
[![License](https://img.shields.io/badge/License-Apache%202.0-yellowgreen.svg?style=flat-square)](https://opensource.org/licenses/Apache-2.0)
[![#matrix-rust-sdk](https://img.shields.io/badge/matrix-%23matrix--rust--sdk-blue?style=flat-square)](https://matrix.to/#/#matrix-rust-sdk:matrix.org)
[![Docs](https://img.shields.io/badge/docs-latest-blue.svg?style=flat-square)](https://docs.rs/matrix-sdk)
# matrix-rust-sdk

47
contrib/mitmproxy/failures.py Normal file
View File

@ -0,0 +1,47 @@
"""
A mitmproxy script that introduces certain request failures in a deterministic
way.
Used mainly for Matrix style requests.
To run execute it with mitmproxy:
>>> mitmproxy -s failures.py`
"""
import time
import json
from mitmproxy import http
from mitmproxy.script import concurrent
REQUEST_COUNT = 0
@concurrent
def request(flow):
global REQUEST_COUNT
REQUEST_COUNT += 1
if REQUEST_COUNT % 2 == 0:
return
elif REQUEST_COUNT % 3 == 0:
flow.response = http.HTTPResponse.make(
500,
b"Gateway error",
)
elif REQUEST_COUNT % 7 == 0:
if "sync" in flow.request.pretty_url:
time.sleep(60)
else:
time.sleep(30)
else:
flow.response = http.HTTPResponse.make(
429,
json.dumps({
"errcode": "M_LIMIT_EXCEEDED",
"error": "Too many requests",
"retry_after_ms": 2000
})
)

37
contrib/mitmproxy/well-known-block.py Normal file
View File

@ -0,0 +1,37 @@
"""
A mitmproxy script that blocks and removes well known Matrix server
information.
There are two ways a Matrix server can trigger the client to reconfigure the
homeserver URL:
1. By responding to a `./well-known/matrix/client` request with a new
homeserver URL.
2. By including a new homeserver URL inside the `/login` response.
To run execute it with mitmproxy:
>>> mitmproxy -s well-known-block.py`
"""
import json
from mitmproxy import http
def request(flow):
if flow.request.path == "/.well-known/matrix/client":
flow.response = http.HTTPResponse.make(
404, # (optional) status code
b"Not found", # (optional) content
{"Content-Type": "text/html"} # (optional) headers
)
def response(flow: http.HTTPFlow):
if flow.request.path == "/_matrix/client/r0/login":
if flow.response.status_code == 200:
body = json.loads(flow.response.content)
body.pop("well_known", None)
flow.response.text = json.dumps(body)

100
design.md
View File

@ -1,100 +0,0 @@
# Matrix Rust SDK
## Design and Layout
#### Async Client
The highest level structure that ties the other pieces of functionality together. The client is responsible for the Request/Response cycle. It can be thought of as a thin layer atop the `BaseClient` passing requests along for the `BaseClient` to handle. A user should be able to write their own `AsyncClient` using the `BaseClient`. It knows how to
- login
- send messages
- encryption ...
- sync client state with the server
- make raw Http requests
#### Base Client/Client State Machine
In addition to Http, the `AsyncClient` passes along methods from the `BaseClient` that deal with `Room`s and `RoomMember`s. This allows the client to keep track of more complicated information that needs to be calculated in some way.
- human-readable room names
- power level?
- ignored list?
- push rulesset?
- more?
#### Crypto State Machine
Given a Matrix response the crypto machine will update its own internal state, along with encryption information. `BaseClient` and the crypto machine together keep track of when to encrypt. It knows when encryption needs to happen based on signals from the `BaseClient`. The crypto state machine is given responses that relate to encryption and can create encrypted request bodies for encryption-related requests. Basically it tells the `BaseClient` to send to-device messages out, and the `BaseClient` is responsible for notifying the crypto state machine when it sent the message so crypto can update state.
#### Client State/Room and RoomMember
The `BaseClient` is responsible for keeping state in sync through the `IncomingResponse`s of `AsyncClient` or querying the `StateStore`. By processing and then delegating incoming `RoomEvent`s, `StateEvent`s, `PresenceEvent`, `IncomingAccountData` and `EphemeralEvent`s to the correct `Room` in the base clients `HashMap<RoomId, Room>` or further to `Room`'s `RoomMember` via the members `HashMap<UserId, RoomMember>`. The `BaseClient` is also responsible for emitting the incoming events to the `EventEmitter` trait.
```rust
/// A Matrix room.
pub struct Room {
/// The unique id of the room.
pub room_id: RoomId,
/// The name of the room, clients use this to represent a room.
pub room_name: RoomName,
/// The mxid of our own user.
pub own_user_id: UserId,
/// The mxid of the room creator.
pub creator: Option<UserId>,
/// The map of room members.
pub members: HashMap<UserId, RoomMember>,
/// A list of users that are currently typing.
pub typing_users: Vec<UserId>,
/// The power level requirements for specific actions in this room
pub power_levels: Option<PowerLevels>,
// TODO when encryption events are handled we store algorithm used and rotation time.
/// A flag indicating if the room is encrypted.
pub encrypted: bool,
/// Number of unread notifications with highlight flag set.
pub unread_highlight: Option<UInt>,
/// Number of unread notifications.
pub unread_notifications: Option<UInt>,
}
```
```rust
pub struct RoomMember {
/// The unique mxid of the user.
pub user_id: UserId,
/// The human readable name of the user.
pub display_name: Option<String>,
/// The matrix url of the users avatar.
pub avatar_url: Option<String>,
/// The time, in ms, since the user interacted with the server.
pub last_active_ago: Option<UInt>,
/// If the user should be considered active.
pub currently_active: Option<bool>,
/// The unique id of the room.
pub room_id: Option<String>,
/// If the member is typing.
pub typing: Option<bool>,
/// The presence of the user, if found.
pub presence: Option<PresenceState>,
/// The presence status message, if found.
pub status_msg: Option<String>,
/// The users power level.
pub power_level: Option<Int>,
/// The normalized power level of this `RoomMember` (0-100).
pub power_level_norm: Option<Int>,
/// The `MembershipState` of this `RoomMember`.
pub membership: MembershipState,
/// The human readable name of this room member.
pub name: String,
/// The events that created the state of this room member.
pub events: Vec<Event>,
/// The `PresenceEvent`s connected to this user.
pub presence_events: Vec<PresenceEvent>,
}
```
#### State Store
The `BaseClient` also has access to a `dyn StateStore` this is an abstraction around a "database" to keep the client state without requesting a full sync from the server on startup. A default implementation that serializes/deserializes JSON to files in a specified directory can be used. The user can also implement `StateStore` to fit any storage solution they choose. The base client handles the storage automatically. There "may be/are TODO" ways for the user to interact directly. The room event handling methods signal if the state was modified; if so, we check if some room state file needs to be overwritten.
- open
- load client/rooms
- store client/room
- update ??
The state store will restore our client state in the `BaseClient` and client authors can just get the latest state that they want to present from the client object. No need to ask the state store for it, this may change if custom setups request this. `StateStore`'s main purpose is to provide load/store functionality and, internally to the crate, update the `BaseClient`.
#### Event Emitter
The consumer of this crate can implement the `EventEmitter` trait for full control over how incoming events are handled by their client. If that isn't enough, it is possible to receive every incoming response with the `AsyncClient::sync_forever` callback.
- list the methods for `EventEmitter`?

5
matrix_sdk/Cargo.toml
View File

@ -50,6 +50,10 @@ default_features = false
version = "0.11.0"
default_features = false
[target.'cfg(not(target_arch = "wasm32"))'.dependencies.backoff]
version = "0.3.0"
features = ["tokio"]
[dependencies.tracing-futures]
version = "0.2.4"
default-features = false
@ -68,7 +72,6 @@ version = "3.0.2"
features = ["wasm-bindgen"]
[dev-dependencies]
async-std = { version = "1.9.0", features = ["unstable"] }
dirs = "3.0.1"
matrix-sdk-test = { version = "0.2.0", path = "../matrix_sdk_test" }
tokio = { version = "1.1.0", default-features = false, features = ["rt-multi-thread", "macros"] }

6
matrix_sdk/examples/autojoin.rs
View File

@ -4,7 +4,7 @@ use tokio::time::{sleep, Duration};
use matrix_sdk::{
self, async_trait,
events::{room::member::MemberEventContent, StrippedStateEvent},
Client, ClientConfig, EventEmitter, RoomState, SyncSettings,
Client, ClientConfig, EventHandler, RoomState, SyncSettings,
};
use url::Url;
@ -19,7 +19,7 @@ impl AutoJoinBot {
}
#[async_trait]
impl EventEmitter for AutoJoinBot {
impl EventHandler for AutoJoinBot {
async fn on_stripped_state_member(
&self,
room: RoomState,
@ -78,7 +78,7 @@ async fn login_and_sync(
println!("logged in as {}", username);
client
.add_event_emitter(Box::new(AutoJoinBot::new(client.clone())))
.set_event_handler(Box::new(AutoJoinBot::new(client.clone())))
.await;
client.sync(SyncSettings::default()).await;

20
matrix_sdk/examples/command_bot.rs
View File

@ -3,10 +3,10 @@ use std::{env, process::exit};
use matrix_sdk::{
self, async_trait,
events::{
room::message::{MessageEventContent, TextMessageEventContent},
room::message::{MessageEventContent, MessageType, TextMessageEventContent},
AnyMessageEventContent, SyncMessageEvent,
},
Client, ClientConfig, EventEmitter, RoomState, SyncSettings,
Client, ClientConfig, EventHandler, RoomState, SyncSettings,
};
use url::Url;
@ -23,7 +23,7 @@ impl CommandBot {
}
#[async_trait]
impl EventEmitter for CommandBot {
impl EventHandler for CommandBot {
async fn on_room_message(
&self,
room: RoomState,
@ -31,13 +31,17 @@ impl EventEmitter for CommandBot {
) {
if let RoomState::Joined(room) = room {
let msg_body = if let SyncMessageEvent {
content: MessageEventContent::Text(TextMessageEventContent { body: msg_body, .. }),
content:
MessageEventContent {
msgtype: MessageType::Text(TextMessageEventContent { body: msg_body, .. }),
..
},
..
} = event
{
msg_body.clone()
msg_body
} else {
String::new()
return;
};
if msg_body.contains("!party") {
@ -88,13 +92,13 @@ async fn login_and_sync(
// add our CommandBot to be notified of incoming messages, we do this after the initial
// sync to avoid responding to messages before the bot was running.
client
.add_event_emitter(Box::new(CommandBot::new(client.clone())))
.set_event_handler(Box::new(CommandBot::new(client.clone())))
.await;
// since we called `sync_once` before we entered our sync loop we must pass
// that sync token to `sync`
let settings = SyncSettings::default().token(client.sync_token().await.unwrap());
// this keeps state from the server streaming in to CommandBot via the EventEmitter trait
// this keeps state from the server streaming in to CommandBot via the EventHandler trait
client.sync(settings).await;
Ok(())

6
matrix_sdk/examples/emoji_verification.rs
View File

@ -10,9 +10,7 @@ use url::Url;
use matrix_sdk::{
self,
events::{
room::message::MessageEventContent, AnySyncMessageEvent, AnySyncRoomEvent, AnyToDeviceEvent,
},
events::{room::message::MessageType, AnySyncMessageEvent, AnySyncRoomEvent, AnyToDeviceEvent},
identifiers::UserId,
Client, LoopCtrl, Sas, SyncSettings,
};
@ -130,7 +128,7 @@ async fn login(
if let AnySyncRoomEvent::Message(event) = event {
match event {
AnySyncMessageEvent::RoomMessage(m) => {
if let MessageEventContent::VerificationRequest(_) = &m.content
if let MessageType::VerificationRequest(_) = &m.content.msgtype
{
let request = client
.get_verification_request(&m.event_id)

2
matrix_sdk/examples/get_profiles.rs
View File

@ -19,7 +19,7 @@ async fn get_profile(client: Client, mxid: &UserId) -> MatrixResult<UserProfile>
let request = profile::get_profile::Request::new(mxid);
// Start the request using matrix_sdk::Client::send
let resp = client.send(request).await?;
let resp = client.send(request, None).await?;
// Use the response and construct a UserProfile struct.
// See https://docs.rs/ruma-client-api/0.9.0/ruma_client_api/r0/profile/get_profile/struct.Response.html

14
matrix_sdk/examples/image_bot.rs
View File

@ -11,10 +11,10 @@ use tokio::sync::Mutex;
use matrix_sdk::{
self, async_trait,
events::{
room::message::{MessageEventContent, TextMessageEventContent},
room::message::{MessageEventContent, MessageType, TextMessageEventContent},
SyncMessageEvent,
},
Client, EventEmitter, RoomState, SyncSettings,
Client, EventHandler, RoomState, SyncSettings,
};
use url::Url;
@ -31,7 +31,7 @@ impl ImageBot {
}
#[async_trait]
impl EventEmitter for ImageBot {
impl EventHandler for ImageBot {
async fn on_room_message(
&self,
room: RoomState,
@ -39,7 +39,11 @@ impl EventEmitter for ImageBot {
) {
if let RoomState::Joined(room) = room {
let msg_body = if let SyncMessageEvent {
content: MessageEventContent::Text(TextMessageEventContent { body: msg_body, .. }),
content:
MessageEventContent {
msgtype: MessageType::Text(TextMessageEventContent { body: msg_body, .. }),
..
},
..
} = event
{
@ -86,7 +90,7 @@ async fn login_and_sync(
client.sync_once(SyncSettings::default()).await.unwrap();
client
.add_event_emitter(Box::new(ImageBot::new(client.clone(), image)))
.set_event_handler(Box::new(ImageBot::new(client.clone(), image)))
.await;
let settings = SyncSettings::default().token(client.sync_token().await.unwrap());

14
matrix_sdk/examples/login.rs
View File

@ -4,16 +4,16 @@ use url::Url;
use matrix_sdk::{
self, async_trait,
events::{
room::message::{MessageEventContent, TextMessageEventContent},
room::message::{MessageEventContent, MessageType, TextMessageEventContent},
SyncMessageEvent,
},
Client, EventEmitter, RoomState, SyncSettings,
Client, EventHandler, RoomState, SyncSettings,
};
struct EventCallback;
#[async_trait]
impl EventEmitter for EventCallback {
impl EventHandler for EventCallback {
async fn on_room_message(
&self,
room: RoomState,
@ -21,7 +21,11 @@ impl EventEmitter for EventCallback {
) {
if let RoomState::Joined(room) = room {
if let SyncMessageEvent {
content: MessageEventContent::Text(TextMessageEventContent { body: msg_body, .. }),
content:
MessageEventContent {
msgtype: MessageType::Text(TextMessageEventContent { body: msg_body, .. }),
..
},
sender,
..
} = event
@ -44,7 +48,7 @@ async fn login(
let homeserver_url = Url::parse(&homeserver_url).expect("Couldn't parse the homeserver URL");
let client = Client::new(homeserver_url).unwrap();
client.add_event_emitter(Box::new(EventCallback)).await;
client.set_event_handler(Box::new(EventCallback)).await;
client
.login(username, password, None, Some("rust-sdk"))

151
matrix_sdk/src/client.rs
View File

@ -41,7 +41,7 @@ use tracing::{error, info, instrument};
use matrix_sdk_base::{
deserialized_responses::{MembersResponse, SyncResponse},
BaseClient, BaseClientConfig, EventEmitter, InvitedRoom, JoinedRoom, LeftRoom, Session, Store,
BaseClient, BaseClientConfig, EventHandler, InvitedRoom, JoinedRoom, LeftRoom, Session, Store,
};
#[cfg(feature = "encryption")]
@ -93,7 +93,7 @@ use matrix_sdk_common::{
room::{
message::{
AudioMessageEventContent, FileMessageEventContent, ImageMessageEventContent,
MessageEventContent, VideoMessageEventContent,
MessageEventContent, MessageType, VideoMessageEventContent,
},
EncryptedFile,
},
@ -118,6 +118,7 @@ use matrix_sdk_common::{
};
use crate::{
error::HttpError,
http_client::{client_with_config, HttpClient, HttpSend},
Error, OutgoingRequest, Result,
};
@ -131,6 +132,12 @@ use crate::{
};
const DEFAULT_SYNC_TIMEOUT: Duration = Duration::from_secs(30);
/// Give the sync a bit more time than the default request timeout does.
const SYNC_REQUEST_TIMEOUT: Duration = Duration::from_secs(15);
/// A conservative upload speed of 1Mbps
const DEFAULT_UPLOAD_SPEED: u64 = 125_000;
/// 5 min minimal upload request timeout, used to clamp the request timeout.
const MIN_UPLOAD_REQUEST_TIMEOUT: Duration = Duration::from_secs(60 * 5);
/// An async/await enabled Matrix client.
///
@ -451,7 +458,7 @@ impl Client {
pub async fn display_name(&self) -> Result<Option<String>> {
let user_id = self.user_id().await.ok_or(Error::AuthenticationRequired)?;
let request = get_display_name::Request::new(&user_id);
let response = self.send(request).await?;
let response = self.send(request, None).await?;
Ok(response.displayname)
}
@ -474,7 +481,7 @@ impl Client {
pub async fn set_display_name(&self, name: Option<&str>) -> Result<()> {
let user_id = self.user_id().await.ok_or(Error::AuthenticationRequired)?;
let request = set_display_name::Request::new(&user_id, name);
self.send(request).await?;
self.send(request, None).await?;
Ok(())
}
@ -499,7 +506,7 @@ impl Client {
pub async fn avatar_url(&self) -> Result<Option<String>> {
let user_id = self.user_id().await.ok_or(Error::AuthenticationRequired)?;
let request = get_avatar_url::Request::new(&user_id);
let response = self.send(request).await?;
let response = self.send(request, None).await?;
Ok(response.avatar_url)
}
@ -512,7 +519,7 @@ impl Client {
pub async fn set_avatar_url(&self, url: Option<&str>) -> Result<()> {
let user_id = self.user_id().await.ok_or(Error::AuthenticationRequired)?;
let request = set_avatar_url::Request::new(&user_id, url);
self.send(request).await?;
self.send(request, None).await?;
Ok(())
}
@ -546,11 +553,11 @@ impl Client {
Ok(())
}
/// Add `EventEmitter` to `Client`.
/// Add `EventHandler` to `Client`.
///
/// The methods of `EventEmitter` are called when the respective `RoomEvents` occur.
pub async fn add_event_emitter(&self, emitter: Box<dyn EventEmitter>) {
self.base_client.add_event_emitter(emitter).await;
/// The methods of `EventHandler` are called when the respective `RoomEvents` occur.
pub async fn set_event_handler(&self, handler: Box<dyn EventHandler>) {
self.base_client.set_event_handler(handler).await;
}
/// Returns the joined rooms this client knows about.
@ -671,7 +678,7 @@ impl Client {
}
);
let response = self.send(request).await?;
let response = self.send(request, None).await?;
self.base_client.receive_login_response(&response).await?;
Ok(response)
@ -733,7 +740,7 @@ impl Client {
info!("Registering to {}", self.homeserver);
let request = registration.into();
self.send(request).await
self.send(request, None).await
}
/// Get or upload a sync filter.
@ -747,7 +754,7 @@ impl Client {
} else {
let user_id = self.user_id().await.ok_or(Error::AuthenticationRequired)?;
let request = FilterUploadRequest::new(&user_id, definition);
let response = self.send(request).await?;
let response = self.send(request, None).await?;
self.base_client
.receive_filter_upload(filter_name, &response)
@ -767,7 +774,7 @@ impl Client {
/// * `room_id` - The `RoomId` of the room to be joined.
pub async fn join_room_by_id(&self, room_id: &RoomId) -> Result<join_room_by_id::Response> {
let request = join_room_by_id::Request::new(room_id);
self.send(request).await
self.send(request, None).await
}
/// Join a room by `RoomId`.
@ -787,7 +794,7 @@ impl Client {
let request = assign!(join_room_by_id_or_alias::Request::new(alias), {
server_name: server_names,
});
self.send(request).await
self.send(request, None).await
}
/// Forget a room by `RoomId`.
@ -799,7 +806,7 @@ impl Client {
/// * `room_id` - The `RoomId` of the room to be forget.
pub async fn forget_room_by_id(&self, room_id: &RoomId) -> Result<forget_room::Response> {
let request = forget_room::Request::new(room_id);
self.send(request).await
self.send(request, None).await
}
/// Ban a user from a room by `RoomId` and `UserId`.
@ -820,7 +827,7 @@ impl Client {
reason: Option<&str>,
) -> Result<ban_user::Response> {
let request = assign!(ban_user::Request::new(room_id, user_id), { reason });
self.send(request).await
self.send(request, None).await
}
/// Kick a user out of the specified room.
@ -841,7 +848,7 @@ impl Client {
reason: Option<&str>,
) -> Result<kick_user::Response> {
let request = assign!(kick_user::Request::new(room_id, user_id), { reason });
self.send(request).await
self.send(request, None).await
}
/// Leave the specified room.
@ -853,7 +860,7 @@ impl Client {
/// * `room_id` - The `RoomId` of the room to leave.
pub async fn leave_room(&self, room_id: &RoomId) -> Result<leave_room::Response> {
let request = leave_room::Request::new(room_id);
self.send(request).await
self.send(request, None).await
}
/// Invite the specified user by `UserId` to the given room.
@ -873,7 +880,7 @@ impl Client {
let recipient = InvitationRecipient::UserId { user_id };
let request = invite_user::Request::new(room_id, recipient);
self.send(request).await
self.send(request, None).await
}
/// Invite the specified user by third party id to the given room.
@ -892,7 +899,7 @@ impl Client {
) -> Result<invite_user::Response> {
let recipient = InvitationRecipient::ThirdPartyId(invite_id);
let request = invite_user::Request::new(room_id, recipient);
self.send(request).await
self.send(request, None).await
}
/// Search the homeserver's directory of public rooms.
@ -938,7 +945,7 @@ impl Client {
since,
server,
});
self.send(request).await
self.send(request, None).await
}
/// Search the homeserver's directory of public rooms with a filter.
@ -976,7 +983,7 @@ impl Client {
room_search: impl Into<get_public_rooms_filtered::Request<'_>>,
) -> Result<get_public_rooms_filtered::Response> {
let request = room_search.into();
self.send(request).await
self.send(request, None).await
}
/// Create a room using the `RoomBuilder` and send the request.
@ -1008,7 +1015,7 @@ impl Client {
room: impl Into<create_room::Request<'_>>,
) -> Result<create_room::Response> {
let request = room.into();
self.send(request).await
self.send(request, None).await
}
/// Sends a request to `/_matrix/client/r0/rooms/{room_id}/messages` and returns
@ -1043,8 +1050,8 @@ impl Client {
&self,
request: impl Into<get_message_events::Request<'_>>,
) -> Result<get_message_events::Response> {
let req = request.into();
self.send(req).await
let request = request.into();
self.send(request, None).await
}
/// Send a request to notify the room of a user typing.
@ -1087,7 +1094,7 @@ impl Client {
let user_id = self.user_id().await.ok_or(Error::AuthenticationRequired)?;
let request = TypingRequest::new(&user_id, room_id, typing.into());
self.send(request).await
self.send(request, None).await
}
/// Send a request to notify the room the user has read specific event.
@ -1106,7 +1113,7 @@ impl Client {
) -> Result<create_receipt::Response> {
let request =
create_receipt::Request::new(room_id, create_receipt::ReceiptType::Read, event_id);
self.send(request).await
self.send(request, None).await
}
/// Send a request to notify the room user has read up to specific event.
@ -1129,7 +1136,7 @@ impl Client {
let request = assign!(set_read_marker::Request::new(room_id, fully_read), {
read_receipt
});
self.send(request).await
self.send(request, None).await
}
/// Share a group session for the given room.
@ -1228,7 +1235,7 @@ impl Client {
/// use matrix_sdk_common::uuid::Uuid;
///
/// let content = AnyMessageEventContent::RoomMessage(
/// MessageEventContent::Text(TextMessageEventContent::plain("Hello world"))
/// MessageEventContent::text_plain("Hello world")
/// );
///
/// let txn_id = Uuid::new_v4();
@ -1260,7 +1267,7 @@ impl Client {
let txn_id = txn_id.unwrap_or_else(Uuid::new_v4).to_string();
let request = send_message_event::Request::new(&room_id, &txn_id, &content);
let response = self.send(request).await?;
let response = self.send(request, None).await?;
Ok(response)
}
@ -1371,26 +1378,26 @@ impl Client {
let content = match content_type.type_() {
mime::IMAGE => {
// TODO create a thumbnail using the image crate?.
MessageEventContent::Image(ImageMessageEventContent {
MessageType::Image(ImageMessageEventContent {
body: body.to_owned(),
info: None,
url: Some(url),
file: encrypted_file,
})
}
mime::AUDIO => MessageEventContent::Audio(AudioMessageEventContent {
mime::AUDIO => MessageType::Audio(AudioMessageEventContent {
body: body.to_owned(),
info: None,
url: Some(url),
file: encrypted_file,
}),
mime::VIDEO => MessageEventContent::Video(VideoMessageEventContent {
mime::VIDEO => MessageType::Video(VideoMessageEventContent {
body: body.to_owned(),
info: None,
url: Some(url),
file: encrypted_file,
}),
_ => MessageEventContent::File(FileMessageEventContent {
_ => MessageType::File(FileMessageEventContent {
filename: None,
body: body.to_owned(),
info: None,
@ -1401,7 +1408,7 @@ impl Client {
self.room_send(
room_id,
AnyMessageEventContent::RoomMessage(content),
AnyMessageEventContent::RoomMessage(MessageEventContent::new(content)),
txn_id,
)
.await
@ -1447,11 +1454,16 @@ impl Client {
let mut data = Vec::new();
reader.read_to_end(&mut data)?;
let timeout = std::cmp::max(
Duration::from_secs(data.len() as u64 / DEFAULT_UPLOAD_SPEED),
MIN_UPLOAD_REQUEST_TIMEOUT,
);
let request = assign!(create_content::Request::new(data), {
content_type: Some(content_type.essence_str()),
});
self.http_client.upload(request).await
Ok(self.http_client.upload(request, Some(timeout)).await?)
}
/// Send an arbitrary request to the server, without updating client state.
@ -1465,6 +1477,9 @@ impl Client {
///
/// * `request` - A filled out and valid request for the endpoint to be hit
///
/// * `timeout` - An optional request timeout setting, this overrides the
/// default request setting if one was set.
///
/// # Example
///
/// ```no_run
@ -1485,18 +1500,22 @@ impl Client {
/// let request = profile::get_profile::Request::new(&user_id);
///
/// // Start the request using Client::send()
/// let response = client.send(request).await.unwrap();
/// let response = client.send(request, None).await.unwrap();
///
/// // Check the corresponding Response struct to find out what types are
/// // returned
/// # })
/// ```
pub async fn send<Request>(&self, request: Request) -> Result<Request::IncomingResponse>
pub async fn send<Request>(
&self,
request: Request,
timeout: Option<Duration>,
) -> Result<Request::IncomingResponse>
where
Request: OutgoingRequest + Debug,
Error: From<FromHttpResponseError<Request::EndpointError>>,
HttpError: From<FromHttpResponseError<Request::EndpointError>>,
{
self.http_client.send(request).await
Ok(self.http_client.send(request, timeout).await?)
}
#[cfg(feature = "encryption")]
@ -1511,7 +1530,7 @@ impl Client {
request.messages.clone(),
);
self.send(request).await
self.send(request, None).await
}
/// Get information of all our own devices.
@ -1540,7 +1559,7 @@ impl Client {
pub async fn devices(&self) -> Result<get_devices::Response> {
let request = get_devices::Request::new();
self.send(request).await
self.send(request, None).await
}
/// Delete the given devices from the server.
@ -1605,13 +1624,13 @@ impl Client {
let mut request = delete_devices::Request::new(devices);
request.auth = auth_data;
self.send(request).await
self.send(request, None).await
}
/// Get the room members for the given room.
pub async fn room_members(&self, room_id: &RoomId) -> Result<MembersResponse> {
let request = get_member_events::Request::new(room_id);
let response = self.send(request).await?;
let response = self.send(request, None).await?;
Ok(self.base_client.receive_members(room_id, &response).await?)
}
@ -1637,7 +1656,12 @@ impl Client {
timeout: sync_settings.timeout,
});
let response = self.send(request).await?;
let timeout = sync_settings
.timeout
.unwrap_or_else(|| Duration::from_secs(0))
+ SYNC_REQUEST_TIMEOUT;
let response = self.send(request, Some(timeout)).await?;
Ok(self.base_client.receive_sync_response(response).await?)
}
@ -1678,11 +1702,10 @@ impl Client {
/// the interesting events through a mpsc channel to another thread e.g. a
/// UI thread.
///
/// ```compile_fail,E0658
/// ```no_run
/// # use matrix_sdk::events::{
/// # room::message::{MessageEvent, MessageEventContent, TextMessageEventContent},
/// # };
/// # use matrix_sdk::Room;
/// # use std::sync::{Arc, RwLock};
/// # use std::time::Duration;
/// # use matrix_sdk::{Client, SyncSettings, LoopCtrl};
@ -1692,7 +1715,7 @@ impl Client {
/// # let homeserver = Url::parse("http://localhost:8080").unwrap();
/// # let mut client = Client::new(homeserver).unwrap();
///
/// use async_std::sync::channel;
/// use tokio::sync::mpsc::channel;
///
/// let (tx, rx) = channel(100);
///
@ -1701,14 +1724,12 @@ impl Client {
/// .timeout(Duration::from_secs(30));
///
/// client
/// .sync_with_callback(sync_settings, async move |response| {
/// .sync_with_callback(sync_settings, |response| async move {
/// let channel = sync_channel;
///
/// for (room_id, room) in response.rooms.join {
/// for event in room.timeline.events {
/// if let Ok(e) = event.deserialize() {
/// channel.send(e).await;
/// }
/// channel.send(event).await.unwrap();
/// }
/// }
///
@ -1778,7 +1799,7 @@ impl Client {
}
OutgoingRequests::SignatureUpload(request) => {
// TODO remove this unwrap.
if let Ok(resp) = self.send(request.clone()).await {
if let Ok(resp) = self.send(request.clone(), None).await {
self.base_client
.mark_request_as_sent(&r.request_id(), &resp)
.await
@ -1838,7 +1859,7 @@ impl Client {
let _lock = self.key_claim_lock.lock().await;
if let Some((request_id, request)) = self.base_client.get_missing_sessions(users).await? {
let response = self.send(request).await?;
let response = self.send(request, None).await?;
self.base_client
.mark_request_as_sent(&request_id, &response)
.await?;
@ -1897,7 +1918,7 @@ impl Client {
request.one_time_keys.as_ref().map_or(0, |k| k.len())
);
let response = self.send(request.clone()).await?;
let response = self.send(request.clone(), None).await?;
self.base_client
.mark_request_as_sent(request_id, &response)
.await?;
@ -1926,7 +1947,7 @@ impl Client {
) -> Result<get_keys::Response> {
let request = assign!(get_keys::Request::new(), { device_keys });
let response = self.send(request).await?;
let response = self.send(request, None).await?;
self.base_client
.mark_request_as_sent(request_id, &response)
.await?;
@ -2079,8 +2100,8 @@ impl Client {
user_signing_key: request.user_signing_key,
});
self.send(request).await?;
self.send(signature_request).await?;
self.send(request, None).await?;
self.send(signature_request, None).await?;
Ok(())
}
@ -2276,7 +2297,7 @@ impl Client {
#[cfg(test)]
mod test {
use crate::ClientConfig;
use crate::{ClientConfig, HttpError};
use super::{
get_public_rooms, get_public_rooms_filtered, register::RegistrationKind, Client,
@ -2471,12 +2492,12 @@ mod test {
.create();
if let Err(err) = client.login("example", "wordpass", None, None).await {
if let crate::Error::RumaResponse(crate::FromHttpResponseError::Http(
crate::ServerError::Known(crate::api::Error {
if let crate::Error::Http(HttpError::FromHttpResponse(
crate::FromHttpResponseError::Http(crate::ServerError::Known(crate::api::Error {
kind,
message,
status_code,
}),
})),
)) = err
{
if let crate::api::error::ErrorKind::Forbidden = kind {
@ -2517,10 +2538,10 @@ mod test {
});
if let Err(err) = client.register(user).await {
if let crate::Error::UiaaError(crate::FromHttpResponseError::Http(
if let crate::Error::Http(HttpError::UiaaError(crate::FromHttpResponseError::Http(
// TODO this should be a UiaaError need to investigate
crate::ServerError::Unknown(e),
)) = err
))) = err
{
assert!(e.to_string().starts_with("EOF while parsing"))
} else {

86
matrix_sdk/src/error.rs
View File

@ -14,13 +14,14 @@
//! Error conditions.
use http::StatusCode;
use matrix_sdk_base::{Error as MatrixError, StoreError};
use matrix_sdk_common::{
api::{
r0::uiaa::{UiaaInfo, UiaaResponse as UiaaError},
Error as RumaClientError,
},
FromHttpResponseError as RumaResponseError, IntoHttpError as RumaIntoHttpError, ServerError,
FromHttpResponseError, IntoHttpError, ServerError,
};
use reqwest::Error as ReqwestError;
use serde_json::Error as JsonError;
@ -33,9 +34,14 @@ use matrix_sdk_base::crypto::store::CryptoStoreError;
/// Result type of the rust-sdk.
pub type Result<T> = std::result::Result<T, Error>;
/// Internal representation of errors.
/// An HTTP error, representing either a connection error or an error while
/// converting the raw HTTP response into a Matrix response.
#[derive(Error, Debug)]
pub enum Error {
pub enum HttpError {
/// An error at the HTTP layer.
#[error(transparent)]
Reqwest(#[from] ReqwestError),
/// Queried endpoint requires authentication but was called on an anonymous client.
#[error("the queried endpoint requires authentication but was called before logging in")]
AuthenticationRequired,
@ -44,9 +50,41 @@ pub enum Error {
#[error("the queried endpoint is not meant for clients")]
NotClientRequest,
/// An error at the HTTP layer.
/// An error converting between ruma_client_api types and Hyper types.
#[error(transparent)]
Reqwest(#[from] ReqwestError),
FromHttpResponse(#[from] FromHttpResponseError<RumaClientError>),
/// An error converting between ruma_client_api types and Hyper types.
#[error(transparent)]
IntoHttp(#[from] IntoHttpError),
/// An error occurred while authenticating.
///
/// When registering or authenticating the Matrix server can send a `UiaaResponse`
/// as the error type, this is a User-Interactive Authentication API response. This
/// represents an error with information about how to authenticate the user.
#[error(transparent)]
UiaaError(#[from] FromHttpResponseError<UiaaError>),
/// The server returned a status code that should be retried.
#[error("Server returned an error {0}")]
Server(StatusCode),
/// The given request can't be cloned and thus can't be retried.
#[error("The request cannot be cloned")]
UnableToCloneRequest,
}
/// Internal representation of errors.
#[derive(Error, Debug)]
pub enum Error {
/// Error doing an HTTP request.
#[error(transparent)]
Http(#[from] HttpError),
/// Queried endpoint requires authentication but was called on an anonymous client.
#[error("the queried endpoint requires authentication but was called before logging in")]
AuthenticationRequired,
/// An error de/serializing type for the `StateStore`
#[error(transparent)]
@ -56,14 +94,6 @@ pub enum Error {
#[error(transparent)]
IO(#[from] IoError),
/// An error converting between ruma_client_api types and Hyper types.
#[error("can't parse the JSON response as a Matrix response")]
RumaResponse(RumaResponseError<RumaClientError>),
/// An error converting between ruma_client_api types and Hyper types.
#[error("can't convert between ruma_client_api and hyper types.")]
IntoHttp(RumaIntoHttpError),
/// An error occurred in the Matrix client library.
#[error(transparent)]
MatrixError(#[from] MatrixError),
@ -76,14 +106,6 @@ pub enum Error {
/// An error occured in the state store.
#[error(transparent)]
StateStore(#[from] StoreError),
/// An error occurred while authenticating.
///
/// When registering or authenticating the Matrix server can send a `UiaaResponse`
/// as the error type, this is a User-Interactive Authentication API response. This
/// represents an error with information about how to authenticate the user.
#[error("User-Interactive Authentication required.")]
UiaaError(RumaResponseError<UiaaError>),
}
impl Error {
@ -99,9 +121,9 @@ impl Error {
/// This method is an convenience method to get to the info the server
/// returned on the first, failed request.
pub fn uiaa_response(&self) -> Option<&UiaaInfo> {
if let Error::UiaaError(RumaResponseError::Http(ServerError::Known(
if let Error::Http(HttpError::UiaaError(FromHttpResponseError::Http(ServerError::Known(
UiaaError::AuthResponse(i),
))) = self
)))) = self
{
Some(i)
} else {
@ -110,20 +132,8 @@ impl Error {
}
}
impl From<RumaResponseError<UiaaError>> for Error {
fn from(error: RumaResponseError<UiaaError>) -> Self {
Self::UiaaError(error)
}
}
impl From<RumaResponseError<RumaClientError>> for Error {
fn from(error: RumaResponseError<RumaClientError>) -> Self {
Self::RumaResponse(error)
}
}
impl From<RumaIntoHttpError> for Error {
fn from(error: RumaIntoHttpError) -> Self {
Self::IntoHttp(error)
impl From<ReqwestError> for Error {
fn from(e: ReqwestError) -> Self {
Error::Http(HttpError::Reqwest(e))
}
}

144
matrix_sdk/src/http_client.rs
View File

@ -14,17 +14,26 @@
use std::{convert::TryFrom, fmt::Debug, sync::Arc};
#[cfg(all(not(test), not(target_arch = "wasm32")))]
use backoff::{future::retry, Error as RetryError, ExponentialBackoff};
#[cfg(all(not(test), not(target_arch = "wasm32")))]
use http::StatusCode;
use http::{HeaderValue, Method as HttpMethod, Response as HttpResponse};
use reqwest::{Client, Response};
use tracing::trace;
use url::Url;
use matrix_sdk_common::{
api::r0::media::create_content, async_trait, locks::RwLock, AsyncTraitDeps, AuthScheme,
FromHttpResponseError,
api::r0::media::create_content, async_trait, instant::Duration, locks::RwLock, AsyncTraitDeps,
AuthScheme, FromHttpResponseError,
};
use crate::{ClientConfig, Error, OutgoingRequest, Result, Session};
use crate::{error::HttpError, ClientConfig, OutgoingRequest, Session};
#[cfg(not(target_arch = "wasm32"))]
const DEFAULT_CONNECTION_TIMEOUT: Duration = Duration::from_secs(5);
#[cfg(not(target_arch = "wasm32"))]
const DEFAULT_REQUEST_TIMEOUT: Duration = Duration::from_secs(10);
/// Abstraction around the http layer. The allows implementors to use different
/// http libraries.
@ -43,7 +52,8 @@ pub trait HttpSend: AsyncTraitDeps {
///
/// ```
/// use std::convert::TryFrom;
/// use matrix_sdk::{HttpSend, Result, async_trait};
/// use matrix_sdk::{HttpSend, async_trait, HttpError};
/// # use std::time::Duration;
///
/// #[derive(Debug)]
/// struct Client(reqwest::Client);
@ -52,7 +62,7 @@ pub trait HttpSend: AsyncTraitDeps {
/// async fn response_to_http_response(
/// &self,
/// mut response: reqwest::Response,
/// ) -> Result<http::Response<Vec<u8>>> {
/// ) -> Result<http::Response<Vec<u8>>, HttpError> {
/// // Convert the reqwest response to a http one.
/// todo!()
/// }
@ -60,7 +70,11 @@ pub trait HttpSend: AsyncTraitDeps {
///
/// #[async_trait]
/// impl HttpSend for Client {
/// async fn send_request(&self, request: http::Request<Vec<u8>>) -> Result<http::Response<Vec<u8>>> {
/// async fn send_request(
/// &self,
/// request: http::Request<Vec<u8>>,
/// timeout: Option<Duration>,
/// ) -> Result<http::Response<Vec<u8>>, HttpError> {
/// Ok(self
/// .response_to_http_response(
/// self.0
@ -74,7 +88,8 @@ pub trait HttpSend: AsyncTraitDeps {
async fn send_request(
&self,
request: http::Request<Vec<u8>>,
) -> Result<http::Response<Vec<u8>>>;
timeout: Option<Duration>,
) -> Result<http::Response<Vec<u8>>, HttpError>;
}
#[derive(Clone, Debug)]
@ -90,7 +105,8 @@ impl HttpClient {
request: Request,
session: Arc<RwLock<Option<Session>>>,
content_type: Option<HeaderValue>,
) -> Result<http::Response<Vec<u8>>> {
timeout: Option<Duration>,
) -> Result<http::Response<Vec<u8>>, HttpError> {
let mut request = {
let read_guard;
let access_token = match Request::METADATA.authentication {
@ -100,11 +116,11 @@ impl HttpClient {
if let Some(session) = read_guard.as_ref() {
Some(session.access_token.as_str())
} else {
return Err(Error::AuthenticationRequired);
return Err(HttpError::AuthenticationRequired);
}
}
AuthScheme::None => None,
_ => return Err(Error::NotClientRequest),
_ => return Err(HttpError::NotClientRequest),
};
request.try_into_http_request(&self.homeserver.to_string(), access_token)?
@ -118,44 +134,51 @@ impl HttpClient {
}
}
self.inner.send_request(request).await
self.inner.send_request(request, timeout).await
}
pub async fn upload(
&self,
request: create_content::Request<'_>,
) -> Result<create_content::Response> {
timeout: Option<Duration>,
) -> Result<create_content::Response, HttpError> {
let response = self
.send_request(request, self.session.clone(), None)
.send_request(request, self.session.clone(), None, timeout)
.await?;
Ok(create_content::Response::try_from(response)?)
}
pub async fn send<Request>(&self, request: Request) -> Result<Request::IncomingResponse>
pub async fn send<Request>(
&self,
request: Request,
timeout: Option<Duration>,
) -> Result<Request::IncomingResponse, HttpError>
where
Request: OutgoingRequest,
Error: From<FromHttpResponseError<Request::EndpointError>>,
Request: OutgoingRequest + Debug,
HttpError: From<FromHttpResponseError<Request::EndpointError>>,
{
let content_type = HeaderValue::from_static("application/json");
let response = self
.send_request(request, self.session.clone(), Some(content_type))
.send_request(request, self.session.clone(), Some(content_type), timeout)
.await?;
trace!("Got response: {:?}", response);
Ok(Request::IncomingResponse::try_from(response)?)
let response = Request::IncomingResponse::try_from(response)?;
Ok(response)
}
}
/// Build a client with the specified configuration.
pub(crate) fn client_with_config(config: &ClientConfig) -> Result<Client> {
pub(crate) fn client_with_config(config: &ClientConfig) -> Result<Client, HttpError> {
let http_client = reqwest::Client::builder();
#[cfg(not(target_arch = "wasm32"))]
let http_client = {
let http_client = match config.timeout {
Some(x) => http_client.timeout(x),
None => http_client,
None => http_client.timeout(DEFAULT_REQUEST_TIMEOUT),
};
let http_client = if config.disable_ssl_verification {
@ -173,12 +196,15 @@ pub(crate) fn client_with_config(config: &ClientConfig) -> Result<Client> {
let user_agent = match &config.user_agent {
Some(a) => a.clone(),
None => HeaderValue::from_str(&format!("matrix-rust-sdk {}", crate::VERSION)).unwrap(),
None => HeaderValue::from_str(&format!("matrix-rust-sdk {}", crate::VERSION))
.expect("Can't construct the version header"),
};
headers.insert(reqwest::header::USER_AGENT, user_agent);
http_client.default_headers(headers)
http_client
.default_headers(headers)
.connect_timeout(DEFAULT_CONNECTION_TIMEOUT)
};
#[cfg(target_arch = "wasm32")]
@ -188,11 +214,15 @@ pub(crate) fn client_with_config(config: &ClientConfig) -> Result<Client> {
Ok(http_client.build()?)
}
async fn response_to_http_response(mut response: Response) -> Result<http::Response<Vec<u8>>> {
async fn response_to_http_response(
mut response: Response,
) -> Result<http::Response<Vec<u8>>, reqwest::Error> {
let status = response.status();
let mut http_builder = HttpResponse::builder().status(status);
let headers = http_builder.headers_mut().unwrap();
let headers = http_builder
.headers_mut()
.expect("Can't get the response builder headers");
for (k, v) in response.headers_mut().drain() {
if let Some(key) = k {
@ -202,7 +232,63 @@ async fn response_to_http_response(mut response: Response) -> Result<http::Respo
let body = response.bytes().await?.as_ref().to_owned();
Ok(http_builder.body(body).unwrap())
Ok(http_builder
.body(body)
.expect("Can't construct a response using the given body"))
}
#[cfg(any(test, target_arch = "wasm32"))]
async fn send_request(
client: &Client,
request: http::Request<Vec<u8>>,
_: Option<Duration>,
) -> Result<http::Response<Vec<u8>>, HttpError> {
let request = reqwest::Request::try_from(request)?;
let response = client.execute(request).await?;
Ok(response_to_http_response(response).await?)
}
#[cfg(all(not(test), not(target_arch = "wasm32")))]
async fn send_request(
client: &Client,
request: http::Request<Vec<u8>>,
timeout: Option<Duration>,
) -> Result<http::Response<Vec<u8>>, HttpError> {
let backoff = ExponentialBackoff::default();
let mut request = reqwest::Request::try_from(request)?;
if let Some(timeout) = timeout {
*request.timeout_mut() = Some(timeout);
}
let request = &request;
let request = || async move {
let request = request.try_clone().ok_or(HttpError::UnableToCloneRequest)?;
let response = client
.execute(request)
.await
.map_err(|e| RetryError::Transient(HttpError::Reqwest(e)))?;
let status_code = response.status();
// TODO TOO_MANY_REQUESTS will have a retry timeout which we should
// use.
if status_code.is_server_error() || response.status() == StatusCode::TOO_MANY_REQUESTS {
return Err(RetryError::Transient(HttpError::Server(status_code)));
}
let response = response_to_http_response(response)
.await
.map_err(|e| RetryError::Permanent(HttpError::Reqwest(e)))?;
Ok(response)
};
let response = retry(backoff, request).await?;
Ok(response)
}
#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
@ -211,10 +297,8 @@ impl HttpSend for Client {
async fn send_request(
&self,
request: http::Request<Vec<u8>>,
) -> Result<http::Response<Vec<u8>>> {
Ok(
response_to_http_response(self.execute(reqwest::Request::try_from(request)?).await?)
.await?,
)
timeout: Option<Duration>,
) -> Result<http::Response<Vec<u8>>, HttpError> {
send_request(&self, request, timeout).await
}
}

8
matrix_sdk/src/lib.rs
View File

@ -66,10 +66,10 @@ compile_error!("only one of 'native-tls' or 'rustls-tls' features can be enabled
#[cfg(feature = "encryption")]
#[cfg_attr(feature = "docs", doc(cfg(encryption)))]
pub use matrix_sdk_base::crypto::LocalTrust;
pub use matrix_sdk_base::crypto::{EncryptionInfo, LocalTrust};
pub use matrix_sdk_base::{
Error as BaseError, EventEmitter, InvitedRoom, JoinedRoom, LeftRoom, RoomInfo, RoomMember,
RoomState, Session, StoreError,
CustomEvent, Error as BaseError, EventHandler, InvitedRoom, JoinedRoom, LeftRoom, RoomInfo,
RoomMember, RoomState, Session, StateChanges, StoreError,
};
pub use matrix_sdk_common::*;
@ -90,7 +90,7 @@ pub use client::{Client, ClientConfig, LoopCtrl, SyncSettings};
#[cfg(feature = "encryption")]
#[cfg_attr(feature = "docs", doc(cfg(encryption)))]
pub use device::Device;
pub use error::{Error, Result};
pub use error::{Error, HttpError, Result};
pub use http_client::HttpSend;
#[cfg(feature = "encryption")]
#[cfg_attr(feature = "docs", doc(cfg(encryption)))]

2
matrix_sdk/src/sas.rs
View File

@ -54,7 +54,7 @@ impl Sas {
}
if let Some(s) = signature {
self.client.send(s).await?;
self.client.send(s, None).await?;
}
Ok(())

6
matrix_sdk_base/examples/state_inspector.rs
View File

@ -1,4 +1,4 @@
use std::{convert::TryFrom, fmt::Debug, io, sync::Arc};
use std::{convert::TryFrom, fmt::Debug, sync::Arc};
use futures::executor::block_on;
use serde::Serialize;
@ -388,7 +388,7 @@ impl Inspector {
}
}
fn main() -> io::Result<()> {
fn main() {
let argparse = Argparse::new("state-inspector")
.global_setting(ArgParseSettings::DisableVersion)
.global_setting(ArgParseSettings::VersionlessSubcommands)
@ -430,6 +430,4 @@ fn main() -> io::Result<()> {
} else {
block_on(inspector.run(matches));
}
Ok(())
}

60
matrix_sdk_base/src/client.rs
View File

@ -62,11 +62,11 @@ use zeroize::Zeroizing;
use crate::{
error::Result,
event_emitter::Emitter,
event_handler::Handler,
rooms::{RoomInfo, RoomType, StrippedRoomInfo},
session::Session,
store::{ambiguity_map::AmbiguityCache, Result as StoreResult, StateChanges, Store},
EventEmitter, RoomState,
EventHandler, RoomState,
};
pub type Token = String;
@ -153,7 +153,7 @@ fn hoist_room_event_prev_content(
Ok(ev)
}
/// Signals to the `BaseClient` which `RoomState` to send to `EventEmitter`.
/// Signals to the `BaseClient` which `RoomState` to send to `EventHandler`.
#[derive(Debug)]
pub enum RoomStateType {
/// Represents a joined room, the `joined_rooms` HashMap will be used.
@ -183,9 +183,9 @@ pub struct BaseClient {
cryptostore: Arc<Mutex<Option<Box<dyn CryptoStore>>>>,
store_path: Arc<Option<PathBuf>>,
store_passphrase: Arc<Option<Zeroizing<String>>>,
/// Any implementor of EventEmitter will act as the callbacks for various
/// Any implementor of EventHandler will act as the callbacks for various
/// events.
event_emitter: Arc<RwLock<Option<Emitter>>>,
event_handler: Arc<RwLock<Option<Handler>>>,
}
#[cfg(not(tarpaulin_include))]
@ -331,7 +331,7 @@ impl BaseClient {
cryptostore: Mutex::new(crypto_store).into(),
store_path: config.store_path.into(),
store_passphrase: config.passphrase.into(),
event_emitter: RwLock::new(None).into(),
event_handler: RwLock::new(None).into(),
})
}
@ -430,15 +430,15 @@ impl BaseClient {
self.sync_token.read().await.clone()
}
/// Add `EventEmitter` to `Client`.
/// Add `EventHandler` to `Client`.
///
/// The methods of `EventEmitter` are called when the respective `RoomEvents` occur.
pub async fn add_event_emitter(&self, emitter: Box<dyn EventEmitter>) {
let emitter = Emitter {
inner: emitter,
/// The methods of `EventHandler` are called when the respective `RoomEvents` occur.
pub async fn set_event_handler(&self, handler: Box<dyn EventHandler>) {
let handler = Handler {
inner: handler,
store: self.store.clone(),
};
*self.event_emitter.write().await = Some(emitter);
*self.event_handler.write().await = Some(handler);
}
async fn handle_timeline(
@ -498,20 +498,17 @@ impl BaseClient {
},
#[cfg(feature = "encryption")]
AnySyncRoomEvent::Message(message) => {
if let AnySyncMessageEvent::RoomEncrypted(encrypted) = message {
if let Some(olm) = self.olm_machine().await {
if let Ok(decrypted) =
olm.decrypt_room_event(encrypted, room_id).await
{
match decrypted.deserialize() {
Ok(decrypted) => e = decrypted,
Err(e) => {
warn!(
"Error deserializing a decrypted event {:?} ",
e
)
}
AnySyncRoomEvent::Message(AnySyncMessageEvent::RoomEncrypted(
encrypted,
)) => {
if let Some(olm) = self.olm_machine().await {
if let Ok(decrypted) =
olm.decrypt_room_event(encrypted, room_id).await
{
match decrypted.deserialize() {
Ok(decrypted) => e = decrypted,
Err(e) => {
warn!("Error deserializing a decrypted event {:?} ", e)
}
}
}
@ -731,7 +728,12 @@ impl BaseClient {
// decryptes to-device events, but leaves room events alone.
// This makes sure that we have the deryption keys for the room
// events at hand.
o.receive_sync_response(&response).await?
o.receive_sync_changes(
&response.to_device,
&response.device_lists,
&response.device_one_time_keys_count,
)
.await?
} else {
response
.to_device
@ -943,8 +945,8 @@ impl BaseClient {
},
};
if let Some(emitter) = self.event_emitter.read().await.as_ref() {
emitter.emit_sync(&response).await;
if let Some(handler) = self.event_handler.read().await.as_ref() {
handler.handle_sync(&response).await;
}
Ok(response)

90
matrix_sdk_base/src/event_emitter/mod.rs → matrix_sdk_base/src/event_handler/mod.rs
View File

@ -52,41 +52,41 @@ use crate::{
};
use matrix_sdk_common::async_trait;
pub(crate) struct Emitter {
pub(crate) inner: Box<dyn EventEmitter>,
pub(crate) struct Handler {
pub(crate) inner: Box<dyn EventHandler>,
pub(crate) store: Store,
}
impl Deref for Emitter {
type Target = dyn EventEmitter;
impl Deref for Handler {
type Target = dyn EventHandler;
fn deref(&self) -> &Self::Target {
&*self.inner
}
}
impl Emitter {
impl Handler {
fn get_room(&self, room_id: &RoomId) -> Option<RoomState> {
self.store.get_room(room_id)
}
pub(crate) async fn emit_sync(&self, response: &SyncResponse) {
pub(crate) async fn handle_sync(&self, response: &SyncResponse) {
for (room_id, room_info) in &response.rooms.join {
if let Some(room) = self.get_room(room_id) {
for event in &room_info.ephemeral.events {
self.emit_ephemeral_event(room.clone(), event).await;
self.handle_ephemeral_event(room.clone(), event).await;
}
for event in &room_info.account_data.events {
self.emit_account_data_event(room.clone(), event).await;
self.handle_account_data_event(room.clone(), event).await;
}
for event in &room_info.state.events {
self.emit_state_event(room.clone(), event).await;
self.handle_state_event(room.clone(), event).await;
}
for event in &room_info.timeline.events {
self.emit_timeline_event(room.clone(), event).await;
self.handle_timeline_event(room.clone(), event).await;
}
}
}
@ -94,15 +94,15 @@ impl Emitter {
for (room_id, room_info) in &response.rooms.leave {
if let Some(room) = self.get_room(room_id) {
for event in &room_info.account_data.events {
self.emit_account_data_event(room.clone(), event).await;
self.handle_account_data_event(room.clone(), event).await;
}
for event in &room_info.state.events {
self.emit_state_event(room.clone(), event).await;
self.handle_state_event(room.clone(), event).await;
}
for event in &room_info.timeline.events {
self.emit_timeline_event(room.clone(), event).await;
self.handle_timeline_event(room.clone(), event).await;
}
}
}
@ -110,7 +110,7 @@ impl Emitter {
for (room_id, room_info) in &response.rooms.invite {
if let Some(room) = self.get_room(room_id) {
for event in &room_info.invite_state.events {
self.emit_stripped_state_event(room.clone(), event).await;
self.handle_stripped_state_event(room.clone(), event).await;
}
}
}
@ -120,7 +120,7 @@ impl Emitter {
}
}
async fn emit_timeline_event(&self, room: RoomState, event: &AnySyncRoomEvent) {
async fn handle_timeline_event(&self, room: RoomState, event: &AnySyncRoomEvent) {
match event {
AnySyncRoomEvent::State(event) => match event {
AnySyncStateEvent::RoomMember(e) => self.on_room_member(room, e).await,
@ -160,7 +160,7 @@ impl Emitter {
}
}
async fn emit_state_event(&self, room: RoomState, event: &AnySyncStateEvent) {
async fn handle_state_event(&self, room: RoomState, event: &AnySyncStateEvent) {
match event {
AnySyncStateEvent::RoomMember(member) => self.on_state_member(room, &member).await,
AnySyncStateEvent::RoomName(name) => self.on_state_name(room, &name).await,
@ -185,9 +185,9 @@ impl Emitter {
}
}
pub(crate) async fn emit_stripped_state_event(
pub(crate) async fn handle_stripped_state_event(
&self,
// TODO these events are only emitted in invited rooms.
// TODO these events are only handleted in invited rooms.
room: RoomState,
event: &AnyStrippedStateEvent,
) {
@ -216,7 +216,7 @@ impl Emitter {
}
}
pub(crate) async fn emit_account_data_event(&self, room: RoomState, event: &AnyBasicEvent) {
pub(crate) async fn handle_account_data_event(&self, room: RoomState, event: &AnyBasicEvent) {
match event {
AnyBasicEvent::Presence(presence) => self.on_non_room_presence(room, &presence).await,
AnyBasicEvent::IgnoredUserList(ignored) => {
@ -227,7 +227,7 @@ impl Emitter {
}
}
pub(crate) async fn emit_ephemeral_event(
pub(crate) async fn handle_ephemeral_event(
&self,
room: RoomState,
event: &AnySyncEphemeralRoomEvent,
@ -262,7 +262,7 @@ pub enum CustomEvent<'c> {
StrippedState(&'c StrippedStateEvent<CustomEventContent>),
}
/// This trait allows any type implementing `EventEmitter` to specify event callbacks for each event.
/// This trait allows any type implementing `EventHandler` to specify event callbacks for each event.
/// The `Client` calls each method when the corresponding event is received.
///
/// # Examples
@ -273,21 +273,25 @@ pub enum CustomEvent<'c> {
/// # use matrix_sdk_base::{
/// # self,
/// # events::{
/// # room::message::{MessageEventContent, TextMessageEventContent},
/// # room::message::{MessageEventContent, MessageType, TextMessageEventContent},
/// # SyncMessageEvent
/// # },
/// # EventEmitter, RoomState
/// # EventHandler, RoomState
/// # };
/// # use matrix_sdk_common::{async_trait, locks::RwLock};
///
/// struct EventCallback;
///
/// #[async_trait]
/// impl EventEmitter for EventCallback {
/// impl EventHandler for EventCallback {
/// async fn on_room_message(&self, room: RoomState, event: &SyncMessageEvent<MessageEventContent>) {
/// if let RoomState::Joined(room) = room {
/// if let SyncMessageEvent {
/// content: MessageEventContent::Text(TextMessageEventContent { body: msg_body, .. }),
/// content:
/// MessageEventContent {
/// msgtype: MessageType::Text(TextMessageEventContent { body: msg_body, .. }),
/// ..
/// },
/// sender,
/// ..
/// } = event
@ -304,7 +308,7 @@ pub enum CustomEvent<'c> {
/// ```
#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
pub trait EventEmitter: Send + Sync {
pub trait EventHandler: Send + Sync {
// ROOM EVENTS from `IncomingTimeline`
/// Fires when `Client` receives a `RoomEvent::RoomMember` event.
async fn on_room_member(&self, _: RoomState, _: &SyncStateEvent<MemberEventContent>) {}
@ -496,11 +500,11 @@ mod test {
pub use wasm_bindgen_test::*;
#[derive(Clone)]
pub struct EvEmitterTest(Arc<Mutex<Vec<String>>>);
pub struct EvHandlerTest(Arc<Mutex<Vec<String>>>);
#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
impl EventEmitter for EvEmitterTest {
impl EventHandler for EvHandlerTest {
async fn on_room_member(&self, _: RoomState, _: &SyncStateEvent<MemberEventContent>) {
self.0.lock().await.push("member".to_string())
}
@ -740,13 +744,13 @@ mod test {
}
#[async_test]
async fn event_emitter_joined() {
async fn event_handler_joined() {
let vec = Arc::new(Mutex::new(Vec::new()));
let test_vec = Arc::clone(&vec);
let emitter = Box::new(EvEmitterTest(vec));
let handler = Box::new(EvHandlerTest(vec));
let client = get_client().await;
client.add_event_emitter(emitter).await;
client.set_event_handler(handler).await;
let response = sync_response(SyncResponseFile::Default);
client.receive_sync_response(response).await.unwrap();
@ -772,13 +776,13 @@ mod test {
}
#[async_test]
async fn event_emitter_invite() {
async fn event_handler_invite() {
let vec = Arc::new(Mutex::new(Vec::new()));
let test_vec = Arc::clone(&vec);
let emitter = Box::new(EvEmitterTest(vec));
let handler = Box::new(EvHandlerTest(vec));
let client = get_client().await;
client.add_event_emitter(emitter).await;
client.set_event_handler(handler).await;
let response = sync_response(SyncResponseFile::Invite);
client.receive_sync_response(response).await.unwrap();
@ -795,13 +799,13 @@ mod test {
}
#[async_test]
async fn event_emitter_leave() {
async fn event_handler_leave() {
let vec = Arc::new(Mutex::new(Vec::new()));
let test_vec = Arc::clone(&vec);
let emitter = Box::new(EvEmitterTest(vec));
let handler = Box::new(EvHandlerTest(vec));
let client = get_client().await;
client.add_event_emitter(emitter).await;
client.set_event_handler(handler).await;
let response = sync_response(SyncResponseFile::Leave);
client.receive_sync_response(response).await.unwrap();
@ -825,13 +829,13 @@ mod test {
}
#[async_test]
async fn event_emitter_more_events() {
async fn event_handler_more_events() {
let vec = Arc::new(Mutex::new(Vec::new()));
let test_vec = Arc::clone(&vec);
let emitter = Box::new(EvEmitterTest(vec));
let handler = Box::new(EvHandlerTest(vec));
let client = get_client().await;
client.add_event_emitter(emitter).await;
client.set_event_handler(handler).await;
let response = sync_response(SyncResponseFile::All);
client.receive_sync_response(response).await.unwrap();
@ -850,13 +854,13 @@ mod test {
}
#[async_test]
async fn event_emitter_voip() {
async fn event_handler_voip() {
let vec = Arc::new(Mutex::new(Vec::new()));
let test_vec = Arc::clone(&vec);
let emitter = Box::new(EvEmitterTest(vec));
let handler = Box::new(EvHandlerTest(vec));
let client = get_client().await;
client.add_event_emitter(emitter).await;
client.set_event_handler(handler).await;
let response = sync_response(SyncResponseFile::Voip);
client.receive_sync_response(response).await.unwrap();

6
matrix_sdk_base/src/lib.rs
View File

@ -46,17 +46,17 @@ pub use matrix_sdk_common::*;
mod client;
mod error;
mod event_emitter;
mod event_handler;
mod rooms;
mod session;
mod store;
pub use event_emitter::EventEmitter;
pub use event_handler::{CustomEvent, EventHandler};
pub use rooms::{
InvitedRoom, JoinedRoom, LeftRoom, Room, RoomInfo, RoomMember, RoomState, StrippedRoom,
StrippedRoomInfo,
};
pub use store::{StateStore, Store, StoreError};
pub use store::{StateChanges, StateStore, Store, StoreError};
pub use client::{BaseClient, BaseClientConfig, RoomStateType};

8
matrix_sdk_base/src/rooms/members.rs
View File

@ -31,7 +31,7 @@ pub struct RoomMember {
pub(crate) event: Arc<MemberEvent>,
pub(crate) profile: Arc<Option<MemberEventContent>>,
pub(crate) presence: Arc<Option<PresenceEvent>>,
pub(crate) power_levles: Arc<Option<SyncStateEvent<PowerLevelsEventContent>>>,
pub(crate) power_levels: Arc<Option<SyncStateEvent<PowerLevelsEventContent>>>,
pub(crate) max_power_level: i64,
pub(crate) is_room_creator: bool,
pub(crate) display_name_ambiguous: bool,
@ -43,7 +43,7 @@ impl RoomMember {
&self.event.state_key
}
/// Get the display name of the member if ther is one.
/// Get the display name of the member if there is one.
pub fn display_name(&self) -> Option<&str> {
if let Some(p) = self.profile.as_ref() {
p.displayname.as_deref()
@ -64,7 +64,7 @@ impl RoomMember {
}
}
/// Get the avatar url of the member, if ther is one.
/// Get the avatar url of the member, if there is one.
pub fn avatar_url(&self) -> Option<&str> {
match self.profile.as_ref() {
Some(p) => p.avatar_url.as_deref(),
@ -86,7 +86,7 @@ impl RoomMember {
/// Get the power level of this member.
pub fn power_level(&self) -> i64 {
self.power_levles
self.power_levels
.as_ref()
.as_ref()
.map(|e| {

2
matrix_sdk_base/src/rooms/normal.rs
View File

@ -420,7 +420,7 @@ impl Room {
event: member_event.into(),
profile: profile.into(),
presence: presence.into(),
power_levles: power.into(),
power_levels: power.into(),
max_power_level,
is_room_creator,
display_name_ambiguous: ambiguous,

4
matrix_sdk_base/src/store/ambiguity_map.rs
View File

@ -195,7 +195,7 @@ impl AmbiguityCache {
let old_display_name = if let Some(event) = old_event {
if matches!(event.content.membership, Join | Invite) {
let dispaly_name = if let Some(d) = changes
let display_name = if let Some(d) = changes
.profiles
.get(room_id)
.and_then(|p| p.get(&member_event.state_key))
@ -213,7 +213,7 @@ impl AmbiguityCache {
event.content.displayname.clone()
};
Some(dispaly_name.unwrap_or_else(|| event.state_key.localpart().to_string()))
Some(display_name.unwrap_or_else(|| event.state_key.localpart().to_string()))
} else {
None
}

25
matrix_sdk_base/src/store/mod.rs
View File

@ -355,26 +355,41 @@ impl Deref for Store {
}
}
/// Store state changes and pass them to the StateStore.
#[derive(Debug, Default)]
pub struct StateChanges {
/// The sync token that relates to this update.
pub sync_token: Option<String>,
/// A user session, containing an access token and information about the associated user account.
pub session: Option<Session>,
/// A mapping of event type string to `AnyBasicEvent`.
pub account_data: BTreeMap<String, AnyBasicEvent>,
/// A mapping of `UserId` to `PresenceEvent`.
pub presence: BTreeMap<UserId, PresenceEvent>,
/// A mapping of `RoomId` to a map of users and their `MemberEvent`.
pub members: BTreeMap<RoomId, BTreeMap<UserId, MemberEvent>>,
/// A mapping of `RoomId` to a map of users and their `MemberEventContent`.
pub profiles: BTreeMap<RoomId, BTreeMap<UserId, MemberEventContent>>,
pub ambiguity_maps: BTreeMap<RoomId, BTreeMap<String, BTreeSet<UserId>>>,
pub(crate) ambiguity_maps: BTreeMap<RoomId, BTreeMap<String, BTreeSet<UserId>>>,
/// A mapping of `RoomId` to a map of event type string to a state key and `AnySyncStateEvent`.
pub state: BTreeMap<RoomId, BTreeMap<String, BTreeMap<String, AnySyncStateEvent>>>,
/// A mapping of `RoomId` to a map of event type string to `AnyBasicEvent`.
pub room_account_data: BTreeMap<RoomId, BTreeMap<String, AnyBasicEvent>>,
/// A map of `RoomId` to `RoomInfo`.
pub room_infos: BTreeMap<RoomId, RoomInfo>,
/// A mapping of `RoomId` to a map of event type to a map of state key to `AnyStrippedStateEvent`.
pub stripped_state: BTreeMap<RoomId, BTreeMap<String, BTreeMap<String, AnyStrippedStateEvent>>>,
/// A mapping of `RoomId` to a map of users and their `StrippedMemberEvent`.
pub stripped_members: BTreeMap<RoomId, BTreeMap<UserId, StrippedMemberEvent>>,
/// A map of `RoomId` to `StrippedRoomInfo`.
pub invited_room_info: BTreeMap<RoomId, StrippedRoomInfo>,
}
impl StateChanges {
/// Create a new `StateChanges` struct with the given sync_token.
pub fn new(sync_token: String) -> Self {
Self {
sync_token: Some(sync_token),
@ -382,25 +397,30 @@ impl StateChanges {
}
}
/// Update the `StateChanges` struct with the given `PresenceEvent`.
pub fn add_presence_event(&mut self, event: PresenceEvent) {
self.presence.insert(event.sender.clone(), event);
}
/// Update the `StateChanges` struct with the given `RoomInfo`.
pub fn add_room(&mut self, room: RoomInfo) {
self.room_infos
.insert(room.room_id.as_ref().to_owned(), room);
}
/// Update the `StateChanges` struct with the given `StrippedRoomInfo`.
pub fn add_stripped_room(&mut self, room: StrippedRoomInfo) {
self.invited_room_info
.insert(room.room_id.as_ref().to_owned(), room);
}
/// Update the `StateChanges` struct with the given `AnyBasicEvent`.
pub fn add_account_data(&mut self, event: AnyBasicEvent) {
self.account_data
.insert(event.content().event_type().to_owned(), event);
}
/// Update the `StateChanges` struct with the given room with a new `AnyBasicEvent`.
pub fn add_room_account_data(&mut self, room_id: &RoomId, event: AnyBasicEvent) {
self.room_account_data
.entry(room_id.to_owned())
@ -408,6 +428,7 @@ impl StateChanges {
.insert(event.content().event_type().to_owned(), event);
}
/// Update the `StateChanges` struct with the given room with a new `AnyStrippedStateEvent`.
pub fn add_stripped_state_event(&mut self, room_id: &RoomId, event: AnyStrippedStateEvent) {
self.stripped_state
.entry(room_id.to_owned())
@ -417,6 +438,7 @@ impl StateChanges {
.insert(event.state_key().to_string(), event);
}
/// Update the `StateChanges` struct with the given room with a new `StrippedMemberEvent`.
pub fn add_stripped_member(&mut self, room_id: &RoomId, event: StrippedMemberEvent) {
let user_id = event.state_key.clone();
@ -426,6 +448,7 @@ impl StateChanges {
.insert(user_id, event);
}
/// Update the `StateChanges` struct with the given room with a new `AnySyncStateEvent`.
pub fn add_state_event(&mut self, room_id: &RoomId, event: AnySyncStateEvent) {
self.state
.entry(room_id.to_owned())

4
matrix_sdk_common/Cargo.toml
View File

@ -22,8 +22,8 @@ async-trait = "0.1.42"
[dependencies.ruma]
version = "0.0.2"
git = "https://github.com/ruma/ruma"
rev = "8c109d3c0a7ec66b352dc82677d30db7cb0723eb"
features = ["client-api", "unstable-pre-spec"]
rev = "c27e66741a8cb0cf5dba45ae3a977f4d6bba715d"
features = ["client-api", "compat", "unstable-pre-spec"]
[target.'cfg(not(target_arch = "wasm32"))'.dependencies]
uuid = { version = "0.8.2", default-features = false, features = ["v4", "serde"] }

3
matrix_sdk_crypto/Cargo.toml
View File

@ -53,6 +53,9 @@ matrix-sdk-test = { version = "0.2.0", path = "../matrix_sdk_test" }
indoc = "1.0.3"
criterion = { version = "0.3.4", features = ["async", "async_futures", "html_reports"] }
[target.'cfg(target_os = "linux")'.dev-dependencies]
pprof = { version = "0.4.2", features = ["flamegraph"] }
[[bench]]
name = "crypto_bench"
harness = false

76
matrix_sdk_crypto/benches/crypto_bench.rs
View File

@ -1,12 +1,13 @@
#[cfg(target_os = "linux")]
mod perf;
use std::convert::TryFrom;
use criterion::{
async_executor::FuturesExecutor, criterion_group, criterion_main, BenchmarkId, Criterion,
Throughput,
};
use criterion::{async_executor::FuturesExecutor, *};
use futures::executor::block_on;
use matrix_sdk_common::{
api::r0::keys::get_keys,
api::r0::keys::{claim_keys, get_keys},
identifiers::{user_id, DeviceIdBox, UserId},
uuid::Uuid,
};
@ -29,7 +30,14 @@ fn keys_query_response() -> get_keys::Response {
get_keys::Response::try_from(data).expect("Can't parse the keys upload response")
}
pub fn receive_keys_query(c: &mut Criterion) {
fn keys_claim_response() -> claim_keys::Response {
let data = include_bytes!("./keys_claim.json");
let data: Value = serde_json::from_slice(data).unwrap();
let data = response_from_file(&data);
claim_keys::Response::try_from(data).expect("Can't parse the keys upload response")
}
pub fn keys_query(c: &mut Criterion) {
let machine = OlmMachine::new(&alice_id(), &alice_device_id());
let response = keys_query_response();
let uuid = Uuid::new_v4();
@ -42,11 +50,14 @@ pub fn receive_keys_query(c: &mut Criterion) {
+ response.self_signing_keys.len()
+ response.user_signing_keys.len();
let mut group = c.benchmark_group("key query throughput");
let mut group = c.benchmark_group("Keys querying");
group.throughput(Throughput::Elements(count as u64));
group.bench_with_input(
BenchmarkId::new("key_query", "150 devices key query response parsing"),
BenchmarkId::new(
"Keys querying",
"150 device keys parsing and signature checking",
),
&response,
|b, response| {
b.to_async(FuturesExecutor)
@ -56,5 +67,52 @@ pub fn receive_keys_query(c: &mut Criterion) {
group.finish()
}
criterion_group!(benches, receive_keys_query);
pub fn keys_claiming(c: &mut Criterion) {
let keys_query_response = keys_query_response();
let uuid = Uuid::new_v4();
let response = keys_claim_response();
let count = response
.one_time_keys
.values()
.fold(0, |acc, d| acc + d.len());
let mut group = c.benchmark_group("Keys claiming throughput");
group.throughput(Throughput::Elements(count as u64));
let name = format!("{} one-time keys claiming and session creation", count);
group.bench_with_input(
BenchmarkId::new("One-time keys claiming", &name),
&response,
|b, response| {
b.iter_batched(
|| {
let machine = OlmMachine::new(&alice_id(), &alice_device_id());
block_on(machine.mark_request_as_sent(&uuid, &keys_query_response)).unwrap();
machine
},
move |machine| block_on(machine.mark_request_as_sent(&uuid, response)).unwrap(),
BatchSize::SmallInput,
)
},
);
group.finish()
}
fn criterion() -> Criterion {
#[cfg(target_os = "linux")]
let criterion = Criterion::default().with_profiler(perf::FlamegraphProfiler::new(100));
#[cfg(not(target_os = "linux"))]
let criterion = Criterion::default();
criterion
}
criterion_group! {
name = benches;
config = criterion();
targets = keys_query, keys_claiming
}
criterion_main!(benches);

1
matrix_sdk_crypto/benches/keys_claim.json Normal file
View File

File diff suppressed because one or more lines are too long

78
matrix_sdk_crypto/benches/perf.rs Normal file
View File

@ -0,0 +1,78 @@
//! This is a simple Criterion Profiler implementation using pprof.
//!
//! It's mostly a direct copy from here: https://www.jibbow.com/posts/criterion-flamegraphs/
use std::{fs::File, os::raw::c_int, path::Path};
use criterion::profiler::Profiler;
use pprof::ProfilerGuard;
/// Small custom profiler that can be used with Criterion to create a flamegraph for benchmarks.
/// Also see [the Criterion documentation on this][custom-profiler].
///
/// ## Example on how to enable the custom profiler:
///
/// ```
/// mod perf;
/// use perf::FlamegraphProfiler;
///
/// fn fibonacci_profiled(criterion: &mut Criterion) {
/// // Use the criterion struct as normal here.
/// }
///
/// fn custom() -> Criterion {
/// Criterion::default().with_profiler(FlamegraphProfiler::new())
/// }
///
/// criterion_group! {
/// name = benches;
/// config = custom();
/// targets = fibonacci_profiled
/// }
/// ```
///
/// The neat thing about this is that it will sample _only_ the benchmark, and not other stuff like
/// the setup process.
///
/// Further, it will only kick in if `--profile-time <time>` is passed to the benchmark binary.
/// A flamegraph will be created for each individual benchmark in its report directory under
/// `profile/flamegraph.svg`.
///
/// [custom-profiler]: https://bheisler.github.io/criterion.rs/book/user_guide/profiling.html#implementing-in-process-profiling-hooks
pub struct FlamegraphProfiler<'a> {
frequency: c_int,
active_profiler: Option<ProfilerGuard<'a>>,
}
impl<'a> FlamegraphProfiler<'a> {
pub fn new(frequency: c_int) -> Self {
FlamegraphProfiler {
frequency,
active_profiler: None,
}
}
}
impl<'a> Profiler for FlamegraphProfiler<'a> {
fn start_profiling(&mut self, _benchmark_id: &str, _benchmark_dir: &Path) {
self.active_profiler = Some(ProfilerGuard::new(self.frequency).unwrap());
}
fn stop_profiling(&mut self, _benchmark_id: &str, benchmark_dir: &Path) {
std::fs::create_dir_all(benchmark_dir)
.expect("Can't create a directory to store the benchmarking report");
let flamegraph_path = benchmark_dir.join("flamegraph.svg");
let flamegraph_file = File::create(&flamegraph_path)
.expect("File system error while creating flamegraph.svg");
if let Some(profiler) = self.active_profiler.take() {
profiler
.report()
.build()
.expect("Can't build profiling report")
.flamegraph(flamegraph_file)
.expect("Error writing flamegraph");
}
}
}

6
matrix_sdk_crypto/src/file_encryption/attachments.rs
View File

@ -257,12 +257,18 @@ impl<'a, R: Read + 'a> AttachmentEncryptor<'a, R> {
}
}
/// Struct holding all the information that is needed to decrypt an encrypted
/// file.
#[derive(Debug, Serialize, Deserialize)]
pub struct EncryptionInfo {
#[serde(rename = "v")]
/// The version of the encryption scheme.
pub version: String,
/// The web key that was used to encrypt the file.
pub web_key: JsonWebKey,
/// The initialization vector that was used to encrypt the file.
pub iv: String,
/// The hashes that can be used to check the validity of the file.
pub hashes: BTreeMap<String, String>,
}

2
matrix_sdk_crypto/src/file_encryption/mod.rs
View File

@ -1,5 +1,5 @@
mod attachments;
mod key_export;
pub use attachments::{AttachmentDecryptor, AttachmentEncryptor, DecryptorError};
pub use attachments::{AttachmentDecryptor, AttachmentEncryptor, DecryptorError, EncryptionInfo};
pub use key_export::{decrypt_key_export, encrypt_key_export};

22
matrix_sdk_crypto/src/identities/manager.rs
View File

@ -43,6 +43,8 @@ pub(crate) struct IdentityManager {
}
impl IdentityManager {
const MAX_KEY_QUERY_USERS: usize = 250;
pub fn new(user_id: Arc<UserId>, device_id: Arc<DeviceIdBox>, store: Store) -> Self {
IdentityManager {
user_id,
@ -298,19 +300,19 @@ impl IdentityManager {
///
/// [`OlmMachine`]: struct.OlmMachine.html
/// [`receive_keys_query_response`]: #method.receive_keys_query_response
pub async fn users_for_key_query(&self) -> Option<KeysQueryRequest> {
let mut users = self.store.users_for_key_query();
pub async fn users_for_key_query(&self) -> Vec<KeysQueryRequest> {
let users = self.store.users_for_key_query();
if users.is_empty() {
None
Vec::new()
} else {
let mut device_keys: BTreeMap<UserId, Vec<Box<DeviceId>>> = BTreeMap::new();
let users: Vec<UserId> = users.into_iter().collect();
for user in users.drain() {
device_keys.insert(user, Vec::new());
}
Some(KeysQueryRequest::new(device_keys))
users
.chunks(Self::MAX_KEY_QUERY_USERS)
.map(|u| u.iter().map(|u| (u.clone(), Vec::new())).collect())
.map(KeysQueryRequest::new)
.collect()
}
}
@ -566,7 +568,7 @@ pub(crate) mod test {
#[async_test]
async fn test_manager_creation() {
let manager = manager();
assert!(manager.users_for_key_query().await.is_none())
assert!(manager.users_for_key_query().await.is_empty())
}
#[async_test]

9
matrix_sdk_crypto/src/key_request.rs
View File

@ -664,6 +664,15 @@ impl KeyRequestMachine {
Some(session)
};
if let Some(s) = &session {
info!(
"Received a forwarded room key from {} for room {} with session id {}",
event.sender,
s.room_id(),
s.session_id()
);
}
Ok((
Some(AnyToDeviceEvent::ForwardedRoomKey(event.clone())),
session,

2
matrix_sdk_crypto/src/lib.rs
View File

@ -42,7 +42,7 @@ mod verification;
pub use error::{MegolmError, OlmError};
pub use file_encryption::{
decrypt_key_export, encrypt_key_export, AttachmentDecryptor, AttachmentEncryptor,
DecryptorError,
DecryptorError, EncryptionInfo,
};
pub use identities::{
Device, LocalTrust, OwnUserIdentity, ReadOnlyDevice, UserDevices, UserIdentities, UserIdentity,

84
matrix_sdk_crypto/src/machine.rs
View File

@ -27,7 +27,7 @@ use matrix_sdk_common::{
upload_keys,
upload_signatures::Request as UploadSignaturesRequest,
},
sync::sync_events::Response as SyncResponse,
sync::sync_events::{DeviceLists, ToDevice as RumaToDevice},
},
assign,
deserialized_responses::ToDevice,
@ -304,16 +304,17 @@ impl OlmMachine {
requests.push(r);
}
if let Some(r) =
self.identity_manager
.users_for_key_query()
.await
.map(|r| OutgoingRequest {
request_id: Uuid::new_v4(),
request: Arc::new(r.into()),
})
for request in self
.identity_manager
.users_for_key_query()
.await
.into_iter()
.map(|r| OutgoingRequest {
request_id: Uuid::new_v4(),
request: Arc::new(r.into()),
})
{
requests.push(r);
requests.push(request);
}
requests.append(&mut self.outgoing_to_device_requests());
@ -599,6 +600,13 @@ impl OlmMachine {
session_key,
None,
)?;
info!(
"Received a new room key from {} for room {} with session id {}",
event.sender,
event.content.room_id,
session.session_id()
);
let event = AnyToDeviceEvent::RoomKey(event.clone());
Ok((Some(event), Some(session)))
}
@ -764,19 +772,31 @@ impl OlmMachine {
self.account.update_uploaded_key_count(key_count).await;
}
/// Handle a sync response and update the internal state of the Olm machine.
/// Handle a to-device and one-time key counts from a sync response.
///
/// This will decrypt to-device events but will not touch events in the room
/// timeline.
/// This will decrypt and handle to-device events returning the decrypted
/// versions of them.
///
/// To decrypt an event from the room timeline call [`decrypt_room_event`].
///
/// # Arguments
///
/// * `response` - The sync latest sync response.
/// * `to_device_events` - The to-device events of the current sync
/// response.
///
/// * `changed_devices` - The list of devices that changed in this sync
/// resopnse.
///
/// * `one_time_keys_count` - The current one-time keys counts that the sync
/// response returned.
///
/// [`decrypt_room_event`]: #method.decrypt_room_event
pub async fn receive_sync_response(&self, response: &SyncResponse) -> OlmResult<ToDevice> {
pub async fn receive_sync_changes(
&self,
to_device_events: &RumaToDevice,
changed_devices: &DeviceLists,
one_time_keys_counts: &BTreeMap<DeviceKeyAlgorithm, UInt>,
) -> OlmResult<ToDevice> {
// Remove verification objects that have expired or are done.
self.verification_machine.garbage_collect();
@ -787,10 +807,9 @@ impl OlmMachine {
..Default::default()
};
self.update_one_time_key_count(&response.device_one_time_keys_count)
.await;
self.update_one_time_key_count(one_time_keys_counts).await;
for user_id in &response.device_lists.changed {
for user_id in &changed_devices.changed {
if let Err(e) = self.identity_manager.mark_user_as_changed(&user_id).await {
error!("Error marking a tracked user as changed {:?}", e);
}
@ -798,13 +817,17 @@ impl OlmMachine {
let mut events = Vec::new();
for event_result in &response.to_device.events {
let mut event = if let Ok(e) = event_result.deserialize() {
e
} else {
// Skip invalid events.
warn!("Received an invalid to-device event {:?}", event_result);
continue;
for event_result in &to_device_events.events {
let mut event = match event_result.deserialize() {
Ok(e) => e,
Err(e) => {
// Skip invalid events.
warn!(
"Received an invalid to-device event {:?} {:?}",
e, event_result
);
continue;
}
};
info!("Received a to-device event {:?}", event);
@ -921,7 +944,10 @@ impl OlmMachine {
// TODO check if this is from a verified device.
let (decrypted_event, _) = session.decrypt(event).await?;
trace!("Successfully decrypted Megolm event {:?}", decrypted_event);
trace!(
"Successfully decrypted a Megolm event {:?}",
decrypted_event
);
// TODO set the encryption info on the event (is it verified, was it
// decrypted, sender key...)
@ -1176,7 +1202,7 @@ pub(crate) mod test {
events::{
room::{
encrypted::EncryptedEventContent,
message::{MessageEventContent, TextMessageEventContent},
message::{MessageEventContent, MessageType},
},
AnyMessageEventContent, AnySyncMessageEvent, AnySyncRoomEvent, AnyToDeviceEvent,
EventType, SyncMessageEvent, ToDeviceEvent, Unsigned,
@ -1721,7 +1747,7 @@ pub(crate) mod test {
let plaintext = "It is a secret to everybody";
let content = MessageEventContent::Text(TextMessageEventContent::plain(plaintext));
let content = MessageEventContent::text_plain(plaintext);
let encrypted_content = alice
.encrypt(
@ -1753,7 +1779,7 @@ pub(crate) mod test {
..
})) => {
assert_eq!(&sender, alice.user_id());
if let MessageEventContent::Text(c) = &content {
if let MessageType::Text(c) = &content.msgtype {
assert_eq!(&c.body, plaintext);
} else {
panic!("Decrypted event has a missmatched content");

10
matrix_sdk_crypto/src/olm/account.rs
View File

@ -191,10 +191,10 @@ impl Account {
}
pub async fn update_uploaded_key_count(&self, key_count: &BTreeMap<DeviceKeyAlgorithm, UInt>) {
let one_time_key_count = key_count.get(&DeviceKeyAlgorithm::SignedCurve25519);
let count: u64 = one_time_key_count.map_or(0, |c| (*c).into());
self.inner.update_uploaded_key_count(count);
if let Some(count) = key_count.get(&DeviceKeyAlgorithm::SignedCurve25519) {
let count: u64 = (*count).into();
self.inner.update_uploaded_key_count(count);
}
}
pub async fn receive_keys_upload_response(
@ -349,7 +349,7 @@ impl Account {
(SessionType::New(session), plaintext)
};
trace!("Successfully decrypted a Olm message: {}", plaintext);
trace!("Successfully decrypted an Olm message: {}", plaintext);
let (event, signing_key) = match self.parse_decrypted_to_device_event(sender, &plaintext) {
Ok(r) => r,

7
matrix_sdk_crypto/src/olm/group_sessions/mod.rs
View File

@ -129,10 +129,7 @@ mod test {
};
use matrix_sdk_common::{
events::{
room::message::{MessageEventContent, TextMessageEventContent},
AnyMessageEventContent,
},
events::{room::message::MessageEventContent, AnyMessageEventContent},
identifiers::{room_id, user_id},
};
@ -156,7 +153,7 @@ mod test {
assert!(!session.expired());
let _ = session
.encrypt(AnyMessageEventContent::RoomMessage(
MessageEventContent::Text(TextMessageEventContent::plain("Test message")),
MessageEventContent::text_plain("Test message"),
))
.await;
assert!(session.expired());

10
matrix_sdk_crypto/src/store/sled.rs
View File

@ -426,11 +426,12 @@ impl CryptoStore for SledStore {
}
async fn save_account(&self, account: ReadOnlyAccount) -> Result<()> {
let pickle = account.pickle(self.get_pickle_mode()).await;
self.account
.insert("account".encode(), serde_json::to_vec(&pickle)?)?;
let changes = Changes {
account: Some(account),
..Default::default()
};
Ok(())
self.save_changes(changes).await
}
async fn save_changes(&self, changes: Changes) -> Result<()> {
@ -569,6 +570,7 @@ impl CryptoStore for SledStore {
async fn save_value(&self, key: String, value: String) -> Result<()> {
self.values.insert(key.as_str().encode(), value.as_str())?;
self.inner.flush_async().await?;
Ok(())
}

4
matrix_sdk_crypto/src/verification/machine.rs
View File

@ -20,7 +20,7 @@ use tracing::{info, trace, warn};
use matrix_sdk_common::{
events::{
room::message::MessageEventContent, AnyMessageEvent, AnySyncMessageEvent, AnySyncRoomEvent,
room::message::MessageType, AnyMessageEvent, AnySyncMessageEvent, AnySyncRoomEvent,
AnyToDeviceEvent,
},
identifiers::{DeviceId, EventId, RoomId, UserId},
@ -233,7 +233,7 @@ impl VerificationMachine {
match m {
AnySyncMessageEvent::RoomMessage(m) => {
if let MessageEventContent::VerificationRequest(r) = &m.content {
if let MessageType::VerificationRequest(r) = &m.content.msgtype {
if self.account.user_id() == &r.to {
info!(
"Received a new verification request from {} {}",

1
matrix_sdk_crypto/src/verification/requests.rs
View File

@ -134,6 +134,7 @@ impl VerificationRequest {
self.inner.lock().unwrap().accept()
}
#[allow(clippy::unnecessary_wraps)]
pub(crate) fn receive_ready(
&self,
sender: &UserId,

4
matrix_sdk_crypto/src/verification/sas/sas_state.rs
View File

@ -1345,7 +1345,7 @@ mod test {
*method = AcceptMethod::Custom(CustomContent {
method: "m.sas.custom".to_string(),
fields: vec![].into_iter().collect(),
data: Default::default(),
});
alice
@ -1394,7 +1394,7 @@ mod test {
*method = StartMethod::Custom(CustomStartContent {
method: "m.sas.custom".to_string(),
fields: vec![].into_iter().collect(),
data: Default::default(),
});
SasState::<Started>::from_start_event(bob.clone(), alice_device, None, start_content)