From 4265240bf5d124e76bfa29fd9be165e5780de26a Mon Sep 17 00:00:00 2001 From: Semen Evdokimov Date: Fri, 4 Sep 2020 14:46:49 +0300 Subject: [PATCH 1/9] Upgrade deps [run ci] --- Cargo.lock | 88 +++++++++++++++++++------------------- Cargo.toml | 11 ++--- jason/Cargo.toml | 2 +- src/conf/turn.rs | 21 +++++++++ src/turn/coturn_metrics.rs | 16 +------ src/turn/service.rs | 13 +----- 6 files changed, 75 insertions(+), 76 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index fab217a48..5f016028f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -316,7 +316,7 @@ version = "0.2.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a71bf475cbe07281d0b3696abb48212db118e7e23219f13596ce865235ff5766" dependencies = [ - "proc-macro2 1.0.19", + "proc-macro2 1.0.20", "quote 1.0.7", "syn 1.0.39", ] @@ -327,7 +327,7 @@ version = "0.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b95aceadaf327f18f0df5962fedc1bde2f870566a0b9f65c89508a3b1f79334c" dependencies = [ - "proc-macro2 1.0.19", + "proc-macro2 1.0.20", "quote 1.0.7", "syn 1.0.39", ] @@ -391,9 +391,9 @@ checksum = "cff77d8686867eceff3105329d4698d96c2391c176d5d03adc90c7389162b5b8" [[package]] name = "async-channel" -version = "1.4.1" +version = "1.4.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "59386c3aa61f4e14c4ddda1a6744c119b4bf278ec9f866d3c20bc5728ee0eb97" +checksum = "21279cfaa4f47df10b1816007e738ca3747ef2ee53ffc51cdbf57a8bb266fee3" dependencies = [ "concurrent-queue", "event-listener", @@ -485,7 +485,7 @@ version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "25f9db3b38af870bf7e5cc649167533b493928e50744e2c30ae350230b414670" dependencies = [ - "proc-macro2 1.0.19", + "proc-macro2 1.0.20", "quote 1.0.7", "syn 1.0.39", ] @@ -502,7 +502,7 @@ version = "0.1.40" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "687c230d85c0a52504709705fc8a53e4a692b83a2184f03dae73e38e1e93a783" dependencies = [ - "proc-macro2 1.0.19", + "proc-macro2 1.0.20", "quote 1.0.7", "syn 1.0.39", ] @@ -877,7 +877,7 @@ checksum = "f0c960ae2da4de88a91b2d920c2a7233b400bc33cb28453a2987822d8392519b" dependencies = [ "fnv", "ident_case", - "proc-macro2 1.0.19", + "proc-macro2 1.0.20", "quote 1.0.7", "strsim 0.9.3", "syn 1.0.39", @@ -910,9 +910,9 @@ dependencies = [ [[package]] name = "deadpool-redis" -version = "0.6.0" +version = "0.6.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "62110f0b55b658cd5772ce60dbc24281ba6b8d3d0172115feb178beebf506363" +checksum = "24db9f0328c6db1d76edc14b250aa25e13999a14508abcfd9f2200cb26f2eed5" dependencies = [ "async-trait", "config", @@ -931,7 +931,7 @@ checksum = "a2658621297f2cf68762a6f7dc0bb7e1ff2cfd6583daef8ee0fed6f7ec468ec0" dependencies = [ "darling", "derive_builder_core", - "proc-macro2 1.0.19", + "proc-macro2 1.0.20", "quote 1.0.7", "syn 1.0.39", ] @@ -943,7 +943,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2791ea3e372c8495c0bc2033991d76b512cd799d07491fbd6890124db9458bef" dependencies = [ "darling", - "proc-macro2 1.0.19", + "proc-macro2 1.0.20", "quote 1.0.7", "syn 1.0.39", ] @@ -954,7 +954,7 @@ version = "0.99.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "298998b1cf6b5b2c8a7b023dfd45821825ce3ba8a8af55c921a0e734e4653f76" dependencies = [ - "proc-macro2 1.0.19", + "proc-macro2 1.0.20", "quote 1.0.7", "syn 1.0.39", ] @@ -1026,7 +1026,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7c5f0096a91d210159eceb2ff5e1c4da18388a170e1e3ce948aac9c8fdbbf595" dependencies = [ "heck", - "proc-macro2 1.0.19", + "proc-macro2 1.0.20", "quote 1.0.7", "syn 1.0.39", ] @@ -1053,7 +1053,7 @@ version = "0.1.8" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "aa4da3c766cd7a0db8242e326e9e4e081edd567072893ed320008189715366a4" dependencies = [ - "proc-macro2 1.0.19", + "proc-macro2 1.0.20", "quote 1.0.7", "syn 1.0.39", "synstructure", @@ -1222,7 +1222,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d0b5a30a4328ab5473878237c447333c093297bded83a4983d10f4deea240d39" dependencies = [ "proc-macro-hack", - "proc-macro2 1.0.19", + "proc-macro2 1.0.20", "quote 1.0.7", "syn 1.0.39", ] @@ -1768,7 +1768,7 @@ dependencies = [ "Inflector", "async-trait", "medea-jason", - "proc-macro2 1.0.19", + "proc-macro2 1.0.20", "quote 1.0.7", "syn 1.0.39", "synstructure", @@ -1862,9 +1862,9 @@ dependencies = [ [[package]] name = "mockall" -version = "0.7.2" +version = "0.8.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "01458f8a19b10cb28195290942e3149161c75acf67ebc8fbf714ab67a2b943bc" +checksum = "c73da7b57d2fc759ea8d17af2441d6d4e31926a9040a76dd67a09d52f59de657" dependencies = [ "cfg-if", "downcast", @@ -1877,12 +1877,12 @@ dependencies = [ [[package]] name = "mockall_derive" -version = "0.7.2" +version = "0.8.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a673cb441f78cd9af4f5919c28576a3cc325fb6b54e42f7047dacce3c718c17b" +checksum = "fb6f291ea23c9f81892649dd70a472d2bcbf12a27db63af2821f44576f80b524" dependencies = [ "cfg-if", - "proc-macro2 1.0.19", + "proc-macro2 1.0.20", "quote 1.0.7", "syn 1.0.39", ] @@ -2075,7 +2075,7 @@ version = "0.4.23" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2c0e815c3ee9a031fdf5af21c10aa17c573c9c6a566328d99e3936c34e36461f" dependencies = [ - "proc-macro2 1.0.19", + "proc-macro2 1.0.20", "quote 1.0.7", "syn 1.0.39", ] @@ -2172,9 +2172,9 @@ dependencies = [ [[package]] name = "proc-macro2" -version = "1.0.19" +version = "1.0.20" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "04f5f085b5d71e2188cb8271e5da0161ad52c3f227a661a3c135fdf28e258b12" +checksum = "175c513d55719db99da20232b06cda8bab6b83ec2d04e3283edf0213c37c1a29" dependencies = [ "unicode-xid 0.2.1", ] @@ -2215,7 +2215,7 @@ checksum = "537aa19b95acde10a12fec4301466386f757403de4cd4e5b4fa78fb5ecb18f72" dependencies = [ "anyhow", "itertools", - "proc-macro2 1.0.19", + "proc-macro2 1.0.20", "quote 1.0.7", "syn 1.0.39", ] @@ -2257,7 +2257,7 @@ version = "1.0.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "aa563d17ecb180e500da1cfd2b028310ac758de548efdd203e18f283af693f37" dependencies = [ - "proc-macro2 1.0.19", + "proc-macro2 1.0.20", ] [[package]] @@ -2360,9 +2360,9 @@ dependencies = [ [[package]] name = "redis" -version = "0.16.0" +version = "0.17.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7b94c6247d45d78d24481a5b7aca146f414ec0f5e39e175f294d1876b943eeeb" +checksum = "95357caf2640abc54651b93c98a8df4fe1ccbf44b8e601ccdf43d5c1451f29ac" dependencies = [ "async-std", "async-trait", @@ -2528,7 +2528,7 @@ version = "1.0.115" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "609feed1d0a73cc36a0182a840a9b37b4a82f0b1150369f0536a9e3f2a31dc48" dependencies = [ - "proc-macro2 1.0.19", + "proc-macro2 1.0.20", "quote 1.0.7", "syn 1.0.39", ] @@ -2581,7 +2581,7 @@ version = "1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4070d2c9b9d258465ad1d82aabb985b84cd9a3afa94da25ece5a9938ba5f1606" dependencies = [ - "proc-macro2 1.0.19", + "proc-macro2 1.0.20", "quote 1.0.7", "syn 1.0.39", ] @@ -2600,9 +2600,9 @@ dependencies = [ [[package]] name = "serial_test" -version = "0.4.0" +version = "0.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fef5f7c7434b2f2c598adc6f9494648a1e41274a75c0ba4056f680ae0c117fd6" +checksum = "1b15f74add9a9d4a3eb2bf739c9a427d266d3895b53d992c3a7c234fec2ff1f1" dependencies = [ "lazy_static", "parking_lot 0.10.2", @@ -2611,11 +2611,11 @@ dependencies = [ [[package]] name = "serial_test_derive" -version = "0.4.0" +version = "0.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d08338d8024b227c62bd68a12c7c9883f5c66780abaef15c550dc56f46ee6515" +checksum = "65f59259be9fc1bf677d06cc1456e97756004a1a5a577480f71430bd7c17ba33" dependencies = [ - "proc-macro2 1.0.19", + "proc-macro2 1.0.20", "quote 1.0.7", "syn 1.0.39", ] @@ -2735,7 +2735,7 @@ version = "0.6.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "133659a15339456eeeb07572eb02a91c91e9815e9cbc89566944d2c8d3efdbf6" dependencies = [ - "proc-macro2 1.0.19", + "proc-macro2 1.0.20", "quote 1.0.7", "syn 1.0.39", ] @@ -2787,7 +2787,7 @@ version = "1.0.39" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "891d8d6567fe7c7f8835a3a98af4208f3846fba258c1bc3c31d6e506239f11f9" dependencies = [ - "proc-macro2 1.0.19", + "proc-macro2 1.0.20", "quote 1.0.7", "unicode-xid 0.2.1", ] @@ -2798,7 +2798,7 @@ version = "0.12.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b834f2d66f734cb897113e34aaff2f1ab4719ca946f9a7358dba8f8064148701" dependencies = [ - "proc-macro2 1.0.19", + "proc-macro2 1.0.20", "quote 1.0.7", "syn 1.0.39", "unicode-xid 0.2.1", @@ -2906,7 +2906,7 @@ version = "0.2.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f0c3acc6aa564495a0f2e1d59fab677cd7f81a19994cfc7f3ad0e64301560389" dependencies = [ - "proc-macro2 1.0.19", + "proc-macro2 1.0.20", "quote 1.0.7", "syn 1.0.39", ] @@ -2984,7 +2984,7 @@ version = "0.3.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "19970cf58f3acc820962be74c4021b8bbc8e8a1c4e3a02095d0aa60cde5f3633" dependencies = [ - "proc-macro2 1.0.19", + "proc-macro2 1.0.20", "prost-build", "quote 1.0.7", "syn 1.0.39", @@ -3195,7 +3195,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "80e0ccfc3378da0cce270c946b676a376943f5cd16aeba64568e7939806f4ada" dependencies = [ - "proc-macro2 1.0.19", + "proc-macro2 1.0.20", "quote 1.0.7", "syn 1.0.39", ] @@ -3390,7 +3390,7 @@ dependencies = [ "bumpalo", "lazy_static", "log", - "proc-macro2 1.0.19", + "proc-macro2 1.0.20", "quote 1.0.7", "syn 1.0.39", "wasm-bindgen-shared", @@ -3424,7 +3424,7 @@ version = "0.2.67" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "841a6d1c35c6f596ccea1f82504a192a60378f64b3bb0261904ad8f2f5657556" dependencies = [ - "proc-macro2 1.0.19", + "proc-macro2 1.0.20", "quote 1.0.7", "syn 1.0.39", "wasm-bindgen-backend", @@ -3457,7 +3457,7 @@ version = "0.3.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "51611ce8e84cba89379d91fc5074bacc5530f69da1c09a2853d906129d12b3b8" dependencies = [ - "proc-macro2 1.0.19", + "proc-macro2 1.0.20", "quote 1.0.7", ] diff --git a/Cargo.toml b/Cargo.toml index ed83c1e25..fe3fa9815 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -37,7 +37,7 @@ bytes = "0.5" chrono = "0.4" config = "0.10" deadpool = "0.5" -deadpool-redis = "=0.6.0" +deadpool-redis = "0.6.1" derive_more = "0.99" dotenv = "0.15" failure = "0.1" @@ -69,18 +69,19 @@ url = "2.1" features = ["release_max_level_trace", "max_level_trace"] [dependencies.redis] package = "redis" - version = "0.16" + version = "0.17" default-features = false features = ["tokio-comp"] [dev-dependencies] +# TODO: upgrade this dependency when awc will be updated actix-codec = "0.2" actix-rt = "1.0" awc = "1.0" derive_builder = "0.9" function_name = "0.2" lazy_static = "1.4" -mockall = "0.7" -serial_test = "0.4" -serial_test_derive = "0.4" +mockall = "0.8" +serial_test = "0.5" +serial_test_derive = "0.5" tempfile = "3.1" diff --git a/jason/Cargo.toml b/jason/Cargo.toml index b81f4e5f4..6f47780b3 100644 --- a/jason/Cargo.toml +++ b/jason/Cargo.toml @@ -42,7 +42,7 @@ log = "0.4" medea-client-api-proto = { path = "../proto/client-api", features = ["jason"] } medea-macro = { path = "../crates/medea-macro" } medea-reactive = { path = "../crates/medea-reactive" } -mockall = { version = "0.7", optional = true } +mockall = { version = "0.8", optional = true } predicates-tree = { version = "1.0", optional = true } serde = { version = "1.0", features = ["derive"] } serde_json = "1.0" diff --git a/src/conf/turn.rs b/src/conf/turn.rs index 808874426..716118807 100644 --- a/src/conf/turn.rs +++ b/src/conf/turn.rs @@ -2,6 +2,7 @@ use std::{borrow::Cow, time::Duration}; +use redis::ConnectionInfo; use serde::{Deserialize, Serialize}; use smart_default::SmartDefault; @@ -76,6 +77,12 @@ pub struct Redis { #[default = 6379] pub port: u16, + /// Username to authenticate on Redis database server with. + /// + /// Defaults to empty value. + #[default = ""] + pub username: Cow<'static, str>, + /// Password to authenticate on Redis database server with. /// /// Defaults to `turn`. @@ -95,6 +102,20 @@ pub struct Redis { pub connect_timeout: Duration, } +impl From<&Redis> for ConnectionInfo { + fn from(cf: &Redis) -> Self { + Self { + username: Some(cf.username.to_string()).filter(|u| !u.is_empty()), + addr: Box::new(redis::ConnectionAddr::Tcp( + cf.host.to_string(), + cf.port, + )), + db: cf.db_number, + passwd: Some(cf.pass.to_string()).filter(|p| !p.is_empty()), + } + } +} + /// Settings of [Coturn]'s admin interface. /// /// [Coturn]: https://github.com/coturn/coturn diff --git a/src/turn/coturn_metrics.rs b/src/turn/coturn_metrics.rs index e0b177146..5d2566abb 100644 --- a/src/turn/coturn_metrics.rs +++ b/src/turn/coturn_metrics.rs @@ -11,7 +11,7 @@ use actix::{ fut::Either, Actor, ActorFuture, AsyncContext, StreamHandler, WrapFuture, }; use futures::{channel::mpsc, StreamExt as _}; -use redis::{ConnectionAddr, ConnectionInfo, RedisError}; +use redis::{ConnectionInfo, RedisError}; use crate::{ log::prelude::*, @@ -55,19 +55,7 @@ impl CoturnMetricsService { cf: &crate::conf::turn::Turn, peer_traffic_watcher: Arc, ) -> Result { - let connection_info = ConnectionInfo { - addr: Box::new(ConnectionAddr::Tcp( - cf.db.redis.host.to_string(), - cf.db.redis.port, - )), - db: cf.db.redis.db_number, - passwd: if cf.db.redis.pass.is_empty() { - None - } else { - Some(cf.db.redis.pass.to_string()) - }, - }; - let client = redis::Client::open(connection_info)?; + let client = redis::Client::open(ConnectionInfo::from(&cf.db.redis))?; Ok(Self { client, diff --git a/src/turn/service.rs b/src/turn/service.rs index 326d47b3a..6a0aee332 100644 --- a/src/turn/service.rs +++ b/src/turn/service.rs @@ -162,18 +162,7 @@ pub fn new_turn_auth_service<'a>( ) -> Result, TurnServiceErr> { let turn_db = TurnDatabase::new( cf.db.redis.connect_timeout, - ConnectionInfo { - addr: Box::new(redis::ConnectionAddr::Tcp( - cf.db.redis.host.to_string(), - cf.db.redis.port, - )), - db: cf.db.redis.db_number, - passwd: if cf.db.redis.pass.is_empty() { - None - } else { - Some(cf.db.redis.pass.to_string()) - }, - }, + ConnectionInfo::from(&cf.db.redis), )?; let coturn_cli = CoturnTelnetClient::new( From caa47b73b8aaee213ea8eb472a7b21c90a986fc7 Mon Sep 17 00:00:00 2001 From: Semen Evdokimov Date: Fri, 4 Sep 2020 14:55:24 +0300 Subject: [PATCH 2/9] Upd CHANGELOG and config.toml --- CHANGELOG.md | 4 +++- Cargo.toml | 2 +- config.toml | 6 ++++++ 3 files changed, 10 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 096366538..27385b185 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -47,10 +47,11 @@ All user visible changes to this project will be documented in this file. This p - `[server.control.grpc]` section to configure Control API gRPC server ([#33]); - `[turn.cli]` and `[turn.cli.pool]` sections to configure access to [Coturn] admin interface ([#84]); - `server.client.http.public_url` option to configure public URL of Client API HTTP server ([#33]); - - `rpc.ping_interval` option to configure `Ping`s sending interval ([#75]). + - `rpc.ping_interval` option to configure `Ping`s sending interval ([#75]); - `[media]` section to configure timeouts involved for determining media flow liveness ([#98]): - `max_lag`; - `init_timeout`. + - `turn.db.redis.username` option to configure username to authenticate on Redis database server with ([#135]). - Testing: - E2E tests for signalling ([#28]). @@ -75,6 +76,7 @@ All user visible changes to this project will be documented in this file. This p [#98]: /../../pull/98 [#105]: /../../pull/105 [#132]: /../../pull/132 +[#135]: /../../pull/135 diff --git a/Cargo.toml b/Cargo.toml index fe3fa9815..144b03c44 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -37,7 +37,7 @@ bytes = "0.5" chrono = "0.4" config = "0.10" deadpool = "0.5" -deadpool-redis = "0.6.1" +deadpool-redis = "0.6" derive_more = "0.99" dotenv = "0.15" failure = "0.1" diff --git a/config.toml b/config.toml index 678f115c4..3a51d0005 100644 --- a/config.toml +++ b/config.toml @@ -117,6 +117,12 @@ # Default: # port = 6379 +# Username to authenticate on Redis database server with. +# +# Env var: MEDEA_TURN__DB__REDIS__USERNAME +# Default: +# username = "" + # Password to authenticate on Coturn's Redis database server with. # # Env var: MEDEA_TURN__DB__REDIS__PASS From 023b06bef7003cf1e3f2b39071dafe464a4d0172 Mon Sep 17 00:00:00 2001 From: Semen Evdokimov Date: Fri, 4 Sep 2020 18:15:53 +0300 Subject: [PATCH 3/9] Refactor some traits to async traits [run ci] --- src/api/control/callback/clients/grpc.rs | 15 +++++++-------- src/api/control/callback/clients/mod.rs | 7 +++---- 2 files changed, 10 insertions(+), 12 deletions(-) diff --git a/src/api/control/callback/clients/grpc.rs b/src/api/control/callback/clients/grpc.rs index c57c58b0f..5014a52ca 100644 --- a/src/api/control/callback/clients/grpc.rs +++ b/src/api/control/callback/clients/grpc.rs @@ -6,7 +6,7 @@ use std::fmt; use medea_control_api_proto::grpc::callback::{ callback_client::CallbackClient as ProtoCallbackClient }; -use futures::future::{FutureExt, LocalBoxFuture}; +use async_trait::async_trait; use tonic::transport::Channel; use crate::api::control::callback::{ @@ -40,17 +40,16 @@ impl GrpcCallbackClient { } } +#[async_trait(?Send)] impl CallbackClient for GrpcCallbackClient { - fn send( + async fn send( &self, request: CallbackRequest, - ) -> LocalBoxFuture<'static, Result<(), CallbackClientError>> { + ) -> Result<(), CallbackClientError> { let mut client = self.client.clone(); - async move { - client.on_event(tonic::Request::new(request.into())).await?; - Ok(()) - } - .boxed_local() + client.on_event(tonic::Request::new(request.into())).await?; + + Ok(()) } } diff --git a/src/api/control/callback/clients/mod.rs b/src/api/control/callback/clients/mod.rs index c69483ed8..5a43af6df 100644 --- a/src/api/control/callback/clients/mod.rs +++ b/src/api/control/callback/clients/mod.rs @@ -4,6 +4,7 @@ pub mod grpc; use std::{fmt, sync::Arc}; +use async_trait::async_trait; use derive_more::From; use futures::future::{FutureExt, LocalBoxFuture}; @@ -24,13 +25,11 @@ pub enum CallbackClientError { TonicTransport(tonic::transport::Error), } +#[async_trait(?Send)] #[cfg_attr(test, mockall::automock)] pub trait CallbackClient: fmt::Debug + Send + Sync { /// Sends provided [`CallbackRequest`]. - fn send( - &self, - request: CallbackRequest, - ) -> LocalBoxFuture<'static, Result<()>>; + async fn send(&self, request: CallbackRequest) -> Result<()>; } #[cfg(test)] From c223c97a432197fa4fbb7aeb2d95b8fbc2268095 Mon Sep 17 00:00:00 2001 From: Semen Evdokimov Date: Mon, 7 Sep 2020 13:46:44 +0300 Subject: [PATCH 4/9] Use async_trait for RpcClient --- jason/src/api/mod.rs | 37 +++++++++++++++-------------- jason/src/rpc/mod.rs | 56 ++++++++++++++++++-------------------------- 2 files changed, 43 insertions(+), 50 deletions(-) diff --git a/jason/src/api/mod.rs b/jason/src/api/mod.rs index 220c9399e..215724542 100644 --- a/jason/src/api/mod.rs +++ b/jason/src/api/mod.rs @@ -52,34 +52,37 @@ impl Jason { /// Returns [`RoomHandle`] for [`Room`]. pub fn init_room(&self) -> RoomHandle { - let rpc = WebSocketRpcClient::new(Box::new(|token| { + let rpc = Rc::new(WebSocketRpcClient::new(Box::new(|token| { Box::pin(async move { let ws = WebSocketRpcTransport::new(&token) .await .map_err(|e| tracerr::new!(e))?; Ok(Rc::new(ws) as Rc) }) - })); + }))); let peer_repository = Box::new(peer::Repository::new(Rc::clone( &self.0.borrow().media_manager, ))); - let inner = self.0.clone(); - spawn_local(rpc.on_normal_close().map(move |res| { - // TODO: Don't close all rooms when multiple rpc connections - // will be supported. - let reason = res.unwrap_or_else(|_| { - ClientDisconnect::RpcClientUnexpectedlyDropped.into() - }); - inner - .borrow_mut() - .rooms - .drain(..) - .for_each(|room| room.close(reason)); - inner.borrow_mut().media_manager = Rc::default(); - })); + spawn_local({ + let rpc = Rc::clone(&rpc); + let inner = self.0.clone(); + async move { + let reason = rpc.on_normal_close().await.unwrap_or_else(|_| { + ClientDisconnect::RpcClientUnexpectedlyDropped.into() + }); + // TODO: Don't close all rooms when multiple rpc connections + // will be supported. + inner + .borrow_mut() + .rooms + .drain(..) + .for_each(|room| room.close(reason)); + inner.borrow_mut().media_manager = Rc::default(); + } + }); - let room = Room::new(Rc::new(rpc), peer_repository); + let room = Room::new(rpc, peer_repository); let handle = room.new_handle(); self.0.borrow_mut().rooms.push(room); handle diff --git a/jason/src/rpc/mod.rs b/jason/src/rpc/mod.rs index df840c3cc..86f1c7602 100644 --- a/jason/src/rpc/mod.rs +++ b/jason/src/rpc/mod.rs @@ -7,6 +7,7 @@ pub mod websocket; use std::{cell::RefCell, rc::Rc, time::Duration}; +use async_trait::async_trait; use derive_more::{Display, From}; use futures::{ channel::{mpsc, oneshot}, @@ -200,6 +201,7 @@ pub enum RpcClientError { } /// Client to talk with server via Client API RPC. +#[async_trait(?Send)] #[cfg_attr(feature = "mockable", mockall::automock)] pub trait RpcClient { /// Tries to upgrade [`State`] of this [`RpcClient`] to [`State::Open`]. @@ -216,10 +218,10 @@ pub trait RpcClient { /// /// If [`RpcClient`] already in [`State::Open`] then this function will be /// instantly resolved. - fn connect( + async fn connect( &self, token: String, - ) -> LocalBoxFuture<'static, Result<(), Traced>>; + ) -> Result<(), Traced>; /// Returns [`Stream`] of all [`Event`]s received by this [`RpcClient`]. /// @@ -239,9 +241,7 @@ pub trait RpcClient { /// abnormal close [`RpcClient::on_connection_loss`] will be thrown. /// /// [`Future`]: std::future::Future - fn on_normal_close( - &self, - ) -> LocalBoxFuture<'static, Result>; + async fn on_normal_close(&self) -> Result; /// Sets reason, that will be passed to underlying transport when this /// client will be dropped. @@ -585,37 +585,29 @@ impl WebSocketRpcClient { } } +#[async_trait(?Send)] impl RpcClient for WebSocketRpcClient { - fn connect( + async fn connect( &self, token: String, - ) -> LocalBoxFuture<'static, Result<(), Traced>> { - let weak_inner = Rc::downgrade(&self.0); - Box::pin(async move { - if let Some(this) = weak_inner.upgrade().map(Self) { - let current_token = this.0.borrow().token.clone(); - if let Some(current_token) = current_token { - if current_token == token { - let state = this.0.borrow().state.borrow().clone(); - match state { - ClientState::Open(_) => Ok(()), - ClientState::Connecting => { - this.connecting_result().await - } - ClientState::Closed(_) => { - this.establish_connection(token).await - } - } - } else { - this.establish_connection(token).await + ) -> Result<(), Traced> { + let current_token = self.0.borrow().token.clone(); + if let Some(current_token) = current_token { + if current_token == token { + let state = self.0.borrow().state.borrow().clone(); + match state { + ClientState::Open(_) => Ok(()), + ClientState::Connecting => self.connecting_result().await, + ClientState::Closed(_) => { + self.establish_connection(token).await } - } else { - this.establish_connection(token).await } } else { - Err(tracerr::new!(RpcClientError::NoSocket)) + self.establish_connection(token).await } - }) + } else { + self.establish_connection(token).await + } } // TODO: proper sub registry @@ -647,12 +639,10 @@ impl RpcClient for WebSocketRpcClient { } } - fn on_normal_close( - &self, - ) -> LocalBoxFuture<'static, Result> { + async fn on_normal_close(&self) -> Result { let (tx, rx) = oneshot::channel(); self.0.borrow_mut().on_close_subscribers.push(tx); - Box::pin(rx) + rx.await } fn on_connection_loss(&self) -> LocalBoxStream<'static, ()> { From a7e4288f5560a5cc82f7ec87cd5b6c81c29bf682 Mon Sep 17 00:00:00 2001 From: Semen Evdokimov Date: Mon, 7 Sep 2020 13:55:15 +0300 Subject: [PATCH 5/9] Get rid of second Rc for WebSocketRpcClient [run ci] --- jason/src/api/mod.rs | 1 - jason/src/api/room.rs | 3 +-- jason/src/rpc/mod.rs | 31 ++++++++++++++++--------------- 3 files changed, 17 insertions(+), 18 deletions(-) diff --git a/jason/src/api/mod.rs b/jason/src/api/mod.rs index 215724542..9e3e20dee 100644 --- a/jason/src/api/mod.rs +++ b/jason/src/api/mod.rs @@ -5,7 +5,6 @@ mod room; use std::{cell::RefCell, rc::Rc}; -use futures::FutureExt as _; use wasm_bindgen::prelude::*; use wasm_bindgen_futures::spawn_local; diff --git a/jason/src/api/room.rs b/jason/src/api/room.rs index 541ecd26d..255576c34 100644 --- a/jason/src/api/room.rs +++ b/jason/src/api/room.rs @@ -220,8 +220,7 @@ impl RoomHandle { ))); } - inner - .rpc + Rc::clone(&inner.rpc) .connect(token) .await .map_err(tracerr::map_from_and_wrap!( => RoomError))?; diff --git a/jason/src/rpc/mod.rs b/jason/src/rpc/mod.rs index 86f1c7602..8ecafe8f8 100644 --- a/jason/src/rpc/mod.rs +++ b/jason/src/rpc/mod.rs @@ -219,7 +219,7 @@ pub trait RpcClient { /// If [`RpcClient`] already in [`State::Open`] then this function will be /// instantly resolved. async fn connect( - &self, + self: Rc, token: String, ) -> Result<(), Traced>; @@ -320,8 +320,8 @@ type RpcTransportFactory = Box< >; impl Inner { - fn new(rpc_transport_factory: RpcTransportFactory) -> Rc> { - Rc::new(RefCell::new(Self { + fn new(rpc_transport_factory: RpcTransportFactory) -> RefCell { + RefCell::new(Self { sock: None, on_close_subscribers: Vec::new(), subs: Vec::new(), @@ -333,7 +333,7 @@ impl Inner { state: ObservableCell::new(ClientState::Closed( ClosedStateReason::NeverConnected, )), - })) + }) } } @@ -345,7 +345,7 @@ impl Inner { /// Client API RPC client to talk with server via [WebSocket]. /// /// [WebSocket]: https://developer.mozilla.org/ru/docs/WebSockets -pub struct WebSocketRpcClient(Rc>); +pub struct WebSocketRpcClient(RefCell); impl WebSocketRpcClient { /// Creates new [`WebSocketRpcClient`] with provided [`RpcTransportFactory`] @@ -440,7 +440,7 @@ impl WebSocketRpcClient { /// Starts [`Heartbeat`] with provided [`RpcSettings`] for provided /// [`RpcTransport`]. async fn start_heartbeat( - &self, + self: Rc, transport: Rc, rpc_settings: RpcSettings, ) -> Result<(), Traced> { @@ -455,10 +455,10 @@ impl WebSocketRpcClient { Heartbeat::start(transport, ping_interval, idle_timeout); let mut on_idle = heartbeat.on_idle(); - let weak_this = Rc::downgrade(&self.0); + let weak_this = Rc::downgrade(&self); spawn_local(async move { while on_idle.next().await.is_some() { - if let Some(this) = weak_this.upgrade().map(Self) { + if let Some(this) = weak_this.upgrade() { this.handle_connection_loss(ClosedStateReason::Idle); } } @@ -470,7 +470,7 @@ impl WebSocketRpcClient { /// Tries to establish [`RpcClient`] connection. async fn establish_connection( - &self, + self: Rc, token: String, ) -> Result<(), Traced> { self.0.borrow_mut().token = Some(token.clone()); @@ -492,7 +492,8 @@ impl WebSocketRpcClient { // wait for ServerMsg::RpcSettings if let Some(msg) = transport.on_message().next().await { if let ServerMsg::RpcSettings(rpc_settings) = msg { - self.start_heartbeat(Rc::clone(&transport), rpc_settings) + Rc::clone(&self) + .start_heartbeat(Rc::clone(&transport), rpc_settings) .await?; } else { let close_reason = @@ -514,10 +515,10 @@ impl WebSocketRpcClient { // subscribe to transport close let mut transport_state_changes = transport.on_state_change(); - let weak_inner = Rc::downgrade(&self.0); + let weak_inner = Rc::downgrade(&self); spawn_local(async move { while let Some(state) = transport_state_changes.next().await { - if let Some(this) = weak_inner.upgrade().map(Self) { + if let Some(this) = weak_inner.upgrade() { if let TransportState::Closed(msg) = state { this.handle_close_message(msg); } @@ -526,11 +527,11 @@ impl WebSocketRpcClient { }); // subscribe to transport message received - let this_clone = Rc::downgrade(&self.0); + let this_clone = Rc::downgrade(&self); let mut on_socket_message = transport.on_message(); spawn_local(async move { while let Some(msg) = on_socket_message.next().await { - if let Some(this) = this_clone.upgrade().map(Self) { + if let Some(this) = this_clone.upgrade() { this.on_transport_message(msg) } } @@ -588,7 +589,7 @@ impl WebSocketRpcClient { #[async_trait(?Send)] impl RpcClient for WebSocketRpcClient { async fn connect( - &self, + self: Rc, token: String, ) -> Result<(), Traced> { let current_token = self.0.borrow().token.clone(); From 6685eba4b55374adc88bdc2782c6f2a1c9696a59 Mon Sep 17 00:00:00 2001 From: Semen Evdokimov Date: Mon, 7 Sep 2020 17:25:52 +0300 Subject: [PATCH 6/9] Fix tests [run ci] --- jason/tests/rpc/mod.rs | 62 +++++++++++++++++++++--------------------- 1 file changed, 31 insertions(+), 31 deletions(-) diff --git a/jason/tests/rpc/mod.rs b/jason/tests/rpc/mod.rs index 4768e3e90..60d435289 100644 --- a/jason/tests/rpc/mod.rs +++ b/jason/tests/rpc/mod.rs @@ -29,10 +29,10 @@ use medea_jason::rpc::websocket::TransportState; wasm_bindgen_test_configure!(run_in_browser); /// Creates [`WebSocketRpcClient`] with the provided [`MockRpcTransport`]. -fn new_client(transport: Rc) -> WebSocketRpcClient { - WebSocketRpcClient::new(Box::new(move |_| { +fn new_client(transport: Rc) -> Rc { + Rc::new(WebSocketRpcClient::new(Box::new(move |_| { Box::pin(future::ok(transport.clone() as Rc)) - })) + }))) } /// Returns result for [`RpcTransport::on_message`] with [`LocalBoxStream`], @@ -63,7 +63,7 @@ async fn message_received_from_transport_is_transmitted_to_sub() { peer_ids: Vec::new(), }; - let ws = WebSocketRpcClient::new(Box::new(|_| { + let ws = Rc::new(WebSocketRpcClient::new(Box::new(|_| { let mut transport = MockRpcTransport::new(); transport.expect_on_state_change().return_once(|| { stream::once(async { TransportState::Open }).boxed() @@ -82,10 +82,10 @@ async fn message_received_from_transport_is_transmitted_to_sub() { transport.expect_set_close_reason().return_const(()); Box::pin(future::ok(Rc::new(transport) as Rc)) - })); + }))); let mut stream = ws.subscribe(); - ws.connect(String::new()).await.unwrap(); + ws.clone().connect(String::new()).await.unwrap(); assert_eq!(stream.next().await.unwrap(), SRV_EVENT); } @@ -148,7 +148,7 @@ async fn transport_is_dropped_when_client_is_dropped() { let rpc_transport = Rc::new(transport); let ws = new_client(rpc_transport.clone()); - ws.connect(String::new()).await.unwrap(); + ws.clone().connect(String::new()).await.unwrap(); ws.set_close_reason(ClientDisconnect::RoomClosed); drop(ws); delay_for(100).await; @@ -186,7 +186,7 @@ async fn send_goes_to_transport() { transport.expect_set_close_reason().return_const(()); let ws = new_client(Rc::new(transport)); - ws.connect(String::new()).await.unwrap(); + ws.clone().connect(String::new()).await.unwrap(); let (test_tx, test_rx) = oneshot::channel(); let test_peer_id = PeerId(9999); let test_sdp_offer = "Hello world!".to_string(); @@ -231,7 +231,7 @@ mod on_close { /// Returns [`WebSocketRpcClient`] which will be resolved /// [`WebSocketRpcClient::on_close`] [`Future`] with provided /// [`CloseMsg`]. - async fn get_client(close_msg: CloseMsg) -> WebSocketRpcClient { + async fn get_client(close_msg: CloseMsg) -> Rc { let mut transport = MockRpcTransport::new(); transport.expect_on_state_change().return_once(move || { stream::iter(vec![ @@ -250,7 +250,7 @@ mod on_close { transport.expect_set_close_reason().return_const(()); let ws = new_client(Rc::new(transport)); - ws.connect(String::new()).await.unwrap(); + ws.clone().connect(String::new()).await.unwrap(); ws } @@ -326,7 +326,7 @@ mod transport_close_reason_on_drop { /// resolved with [`RpcTransport`]'s close reason /// ([`ClientDisconnect`]). async fn get_client( - ) -> (WebSocketRpcClient, oneshot::Receiver) { + ) -> (Rc, oneshot::Receiver) { let mut transport = MockRpcTransport::new(); transport.expect_on_state_change().return_once(|| { stream::once(async { TransportState::Open }).boxed() @@ -346,7 +346,7 @@ mod transport_close_reason_on_drop { }); let ws = new_client(Rc::new(transport)); - ws.connect(String::new()).await.unwrap(); + ws.clone().connect(String::new()).await.unwrap(); (ws, test_rx) } @@ -430,7 +430,7 @@ mod connect { #[wasm_bindgen_test] async fn closed() { let (test_tx, mut test_rx) = mpsc::unbounded(); - let ws = WebSocketRpcClient::new(Box::new(move |_| { + let ws = Rc::new(WebSocketRpcClient::new(Box::new(move |_| { test_tx.unbounded_send(()).unwrap(); let mut transport = MockRpcTransport::new(); transport.expect_on_message().times(3).returning(|| { @@ -446,8 +446,8 @@ mod connect { }); let transport = Rc::new(transport); Box::pin(future::ok(transport as Rc)) - })); - ws.connect(String::new()).await.unwrap(); + }))); + ws.clone().connect(String::new()).await.unwrap(); timeout(500, test_rx.next()).await.unwrap().unwrap(); } @@ -468,7 +468,7 @@ mod connect { #[wasm_bindgen_test] async fn connecting() { let mut connecting_count: i32 = 0; - let ws = WebSocketRpcClient::new(Box::new(move |_| { + let ws = Rc::new(WebSocketRpcClient::new(Box::new(move |_| { Box::pin(async move { let mut transport = MockRpcTransport::new(); transport.expect_on_message().times(3).returning(|| { @@ -491,8 +491,8 @@ mod connect { Ok(Rc::clone(&transport) as Rc) } }) - })); - let first_connect_fut = ws.connect(String::new()); + }))); + let first_connect_fut = ws.clone().connect(String::new()); spawn_local(async move { first_connect_fut.await.unwrap(); }); @@ -516,7 +516,7 @@ mod connect { #[wasm_bindgen_test] async fn open() { let mut connection_count = 0; - let ws = WebSocketRpcClient::new(Box::new(move |_| { + let ws = Rc::new(WebSocketRpcClient::new(Box::new(move |_| { Box::pin(async move { connection_count += 1; if connection_count > 1 { @@ -537,8 +537,8 @@ mod connect { let transport = Rc::new(transport); Ok(transport as Rc) }) - })); - ws.connect(String::new()).await.unwrap(); + }))); + ws.clone().connect(String::new()).await.unwrap(); timeout(50, ws.connect(String::new())) .await @@ -558,8 +558,8 @@ mod on_connection_loss { idle_timeout_ms: Option, ping_interval_ms: Option, transport_changes: Option, - ) -> WebSocketRpcClient { - let ws = WebSocketRpcClient::new(Box::new(move |_| { + ) -> Rc { + let ws = Rc::new(WebSocketRpcClient::new(Box::new(move |_| { Box::pin(async move { let mut transport = MockRpcTransport::new(); transport.expect_on_message().times(3).returning(move || { @@ -586,8 +586,8 @@ mod on_connection_loss { let transport = Rc::new(transport); Ok(transport as Rc) }) - })); - ws.connect(String::new()).await.unwrap(); + }))); + ws.clone().connect(String::new()).await.unwrap(); ws } @@ -730,7 +730,7 @@ mod on_reconnected { /// first [`RpcClient`] connection. #[wasm_bindgen_test] async fn doesnt_fires_on_first_connection() { - let ws = WebSocketRpcClient::new(Box::new(move |_| { + let ws = Rc::new(WebSocketRpcClient::new(Box::new(move |_| { Box::pin(async move { let mut transport = MockRpcTransport::new(); transport.expect_on_message().times(3).returning(|| { @@ -747,10 +747,10 @@ mod on_reconnected { Ok(Rc::new(transport) as Rc) }) - })); + }))); let mut on_reconnected_stream = ws.on_reconnected(); - ws.connect(String::new()).await.unwrap(); + ws.clone().connect(String::new()).await.unwrap(); timeout(10, on_reconnected_stream.next()).await.unwrap_err(); } @@ -769,7 +769,7 @@ mod on_reconnected { let on_close_mock_clone = on_state_change_mock.clone(); let on_message_mock_clone = on_message_mock.clone(); - let ws = WebSocketRpcClient::new(Box::new(move |_| { + let ws = Rc::new(WebSocketRpcClient::new(Box::new(move |_| { let messages_mock = on_message_mock_clone.clone(); let on_close_mock = on_close_mock_clone.clone(); Box::pin(async move { @@ -786,10 +786,10 @@ mod on_reconnected { let transport = Rc::new(transport); Ok(transport as Rc) }) - })); + }))); let mut on_reconnected_stream = ws.on_reconnected(); - ws.connect(String::new()).await.unwrap(); + ws.clone().connect(String::new()).await.unwrap(); on_state_change_mock .set(TransportState::Closed(CloseMsg::Abnormal(1006))); From 7f249ecd06d9ca8ea80513ad594a661ce2707b0b Mon Sep 17 00:00:00 2001 From: alexlapa Date: Tue, 8 Sep 2020 10:29:39 +0300 Subject: [PATCH 7/9] corrections [run ci] --- Cargo.lock | 96 ++++++++++++------------- Cargo.toml | 3 +- jason/src/api/mod.rs | 2 +- src/api/control/callback/service.rs | 2 +- src/conf/control.rs | 2 +- src/conf/log.rs | 2 +- src/conf/media.rs | 2 +- src/conf/mod.rs | 2 +- src/conf/rpc.rs | 2 +- src/conf/server.rs | 4 +- src/conf/shutdown.rs | 2 +- src/conf/turn.rs | 2 +- src/signalling/peers/traffic_watcher.rs | 27 +++---- 13 files changed, 73 insertions(+), 75 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 5f016028f..6d5c2a440 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -135,7 +135,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a60f9ba7c4e6df97f3aacb14bb5c0cd7d98a49dcbaed0d7f292912ad9a6a3ed2" dependencies = [ "quote 1.0.7", - "syn 1.0.39", + "syn 1.0.40", ] [[package]] @@ -318,7 +318,7 @@ checksum = "a71bf475cbe07281d0b3696abb48212db118e7e23219f13596ce865235ff5766" dependencies = [ "proc-macro2 1.0.20", "quote 1.0.7", - "syn 1.0.39", + "syn 1.0.40", ] [[package]] @@ -329,7 +329,7 @@ checksum = "b95aceadaf327f18f0df5962fedc1bde2f870566a0b9f65c89508a3b1f79334c" dependencies = [ "proc-macro2 1.0.20", "quote 1.0.7", - "syn 1.0.39", + "syn 1.0.40", ] [[package]] @@ -435,9 +435,9 @@ dependencies = [ [[package]] name = "async-mutex" -version = "1.2.0" +version = "1.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "065de1ccf10280d0d75c2f3a71a970ee1007c85c51aa3e7deee1df100f1dfadb" +checksum = "66941c2577c4fa351e4ce5fdde8f86c69b88d623f3b955be1bc7362a23434632" dependencies = [ "event-listener", ] @@ -487,7 +487,7 @@ checksum = "25f9db3b38af870bf7e5cc649167533b493928e50744e2c30ae350230b414670" dependencies = [ "proc-macro2 1.0.20", "quote 1.0.7", - "syn 1.0.39", + "syn 1.0.40", ] [[package]] @@ -504,7 +504,7 @@ checksum = "687c230d85c0a52504709705fc8a53e4a692b83a2184f03dae73e38e1e93a783" dependencies = [ "proc-macro2 1.0.20", "quote 1.0.7", - "syn 1.0.39", + "syn 1.0.40", ] [[package]] @@ -803,12 +803,12 @@ dependencies = [ [[package]] name = "crossbeam-channel" -version = "0.4.3" +version = "0.4.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "09ee0cc8804d5393478d743b035099520087a5186f3b93fa58cec08fa62407b6" +checksum = "b153fe7cbef478c567df0f972e02e6d736db11affe43dfc9c56a9374d1adfb87" dependencies = [ - "cfg-if", "crossbeam-utils", + "maybe-uninit", ] [[package]] @@ -880,7 +880,7 @@ dependencies = [ "proc-macro2 1.0.20", "quote 1.0.7", "strsim 0.9.3", - "syn 1.0.39", + "syn 1.0.40", ] [[package]] @@ -891,7 +891,7 @@ checksum = "d9b5a2f4ac4969822c62224815d069952656cadc7084fdca9751e6d959189b72" dependencies = [ "darling_core", "quote 1.0.7", - "syn 1.0.39", + "syn 1.0.40", ] [[package]] @@ -933,7 +933,7 @@ dependencies = [ "derive_builder_core", "proc-macro2 1.0.20", "quote 1.0.7", - "syn 1.0.39", + "syn 1.0.40", ] [[package]] @@ -945,7 +945,7 @@ dependencies = [ "darling", "proc-macro2 1.0.20", "quote 1.0.7", - "syn 1.0.39", + "syn 1.0.40", ] [[package]] @@ -956,7 +956,7 @@ checksum = "298998b1cf6b5b2c8a7b023dfd45821825ce3ba8a8af55c921a0e734e4653f76" dependencies = [ "proc-macro2 1.0.20", "quote 1.0.7", - "syn 1.0.39", + "syn 1.0.40", ] [[package]] @@ -1028,7 +1028,7 @@ dependencies = [ "heck", "proc-macro2 1.0.20", "quote 1.0.7", - "syn 1.0.39", + "syn 1.0.40", ] [[package]] @@ -1055,7 +1055,7 @@ checksum = "aa4da3c766cd7a0db8242e326e9e4e081edd567072893ed320008189715366a4" dependencies = [ "proc-macro2 1.0.20", "quote 1.0.7", - "syn 1.0.39", + "syn 1.0.40", "synstructure", ] @@ -1224,7 +1224,7 @@ dependencies = [ "proc-macro-hack", "proc-macro2 1.0.20", "quote 1.0.7", - "syn 1.0.39", + "syn 1.0.40", ] [[package]] @@ -1316,12 +1316,9 @@ dependencies = [ [[package]] name = "hashbrown" -version = "0.8.2" +version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e91b62f79061a0bc2e046024cb7ba44b08419ed238ecbd9adbd787434b9e8c25" -dependencies = [ - "autocfg", -] +checksum = "00d63df3d41950fb462ed38308eea019113ad1508da725bbedcd0fa5a85ef5f7" [[package]] name = "heck" @@ -1438,9 +1435,9 @@ dependencies = [ [[package]] name = "indexmap" -version = "1.5.2" +version = "1.6.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4e47a3566dd4fd4eec714ae6ceabdee0caec795be835c223d92c2d40f1e8cf1c" +checksum = "55e2e4c765aa53a0424761bf9f41aa7a6ac1efa87238f59560640e27fca028f2" dependencies = [ "autocfg", "hashbrown", @@ -1654,7 +1651,6 @@ dependencies = [ "serde_json", "serde_yaml", "serial_test", - "serial_test_derive", "slog", "slog-async", "slog-envlogger", @@ -1770,7 +1766,7 @@ dependencies = [ "medea-jason", "proc-macro2 1.0.20", "quote 1.0.7", - "syn 1.0.39", + "syn 1.0.40", "synstructure", ] @@ -1862,9 +1858,9 @@ dependencies = [ [[package]] name = "mockall" -version = "0.8.0" +version = "0.8.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c73da7b57d2fc759ea8d17af2441d6d4e31926a9040a76dd67a09d52f59de657" +checksum = "68ecfc6340c5b98a9a270b56e5f43353d87ebb18d9458a9301344bc79317c563" dependencies = [ "cfg-if", "downcast", @@ -1877,14 +1873,14 @@ dependencies = [ [[package]] name = "mockall_derive" -version = "0.8.0" +version = "0.8.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fb6f291ea23c9f81892649dd70a472d2bcbf12a27db63af2821f44576f80b524" +checksum = "b873f753808fe0c3827ce76edb3ace27804966dfde3043adfac1c24d0a2559df" dependencies = [ "cfg-if", "proc-macro2 1.0.20", "quote 1.0.7", - "syn 1.0.39", + "syn 1.0.40", ] [[package]] @@ -1906,9 +1902,9 @@ dependencies = [ [[package]] name = "net2" -version = "0.2.34" +version = "0.2.35" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2ba7c918ac76704fb42afcbbb43891e72731f3dcca3bef2a19786297baf14af7" +checksum = "3ebc3ec692ed7c9a255596c67808dee269f64655d8baf7b4f0638e51ba1d6853" dependencies = [ "cfg-if", "libc", @@ -2077,7 +2073,7 @@ checksum = "2c0e815c3ee9a031fdf5af21c10aa17c573c9c6a566328d99e3936c34e36461f" dependencies = [ "proc-macro2 1.0.20", "quote 1.0.7", - "syn 1.0.39", + "syn 1.0.40", ] [[package]] @@ -2094,9 +2090,9 @@ checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184" [[package]] name = "polling" -version = "0.1.8" +version = "0.1.9" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b1e9fa0ab21ed700cf0c4ebec57ae5496bec942a0aef9545562979a9f75b97aa" +checksum = "8fffa183f6bd5f1a8a3e1f60ce2f8d5621e350eed84a62d6daaa5b9d1aaf6fbd" dependencies = [ "cfg-if", "libc", @@ -2217,7 +2213,7 @@ dependencies = [ "itertools", "proc-macro2 1.0.20", "quote 1.0.7", - "syn 1.0.39", + "syn 1.0.40", ] [[package]] @@ -2530,7 +2526,7 @@ checksum = "609feed1d0a73cc36a0182a840a9b37b4a82f0b1150369f0536a9e3f2a31dc48" dependencies = [ "proc-macro2 1.0.20", "quote 1.0.7", - "syn 1.0.39", + "syn 1.0.40", ] [[package]] @@ -2583,7 +2579,7 @@ checksum = "4070d2c9b9d258465ad1d82aabb985b84cd9a3afa94da25ece5a9938ba5f1606" dependencies = [ "proc-macro2 1.0.20", "quote 1.0.7", - "syn 1.0.39", + "syn 1.0.40", ] [[package]] @@ -2617,7 +2613,7 @@ checksum = "65f59259be9fc1bf677d06cc1456e97756004a1a5a577480f71430bd7c17ba33" dependencies = [ "proc-macro2 1.0.20", "quote 1.0.7", - "syn 1.0.39", + "syn 1.0.40", ] [[package]] @@ -2737,7 +2733,7 @@ checksum = "133659a15339456eeeb07572eb02a91c91e9815e9cbc89566944d2c8d3efdbf6" dependencies = [ "proc-macro2 1.0.20", "quote 1.0.7", - "syn 1.0.39", + "syn 1.0.40", ] [[package]] @@ -2783,9 +2779,9 @@ dependencies = [ [[package]] name = "syn" -version = "1.0.39" +version = "1.0.40" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "891d8d6567fe7c7f8835a3a98af4208f3846fba258c1bc3c31d6e506239f11f9" +checksum = "963f7d3cc59b59b9325165add223142bbf1df27655d07789f109896d353d8350" dependencies = [ "proc-macro2 1.0.20", "quote 1.0.7", @@ -2800,7 +2796,7 @@ checksum = "b834f2d66f734cb897113e34aaff2f1ab4719ca946f9a7358dba8f8064148701" dependencies = [ "proc-macro2 1.0.20", "quote 1.0.7", - "syn 1.0.39", + "syn 1.0.40", "unicode-xid 0.2.1", ] @@ -2908,7 +2904,7 @@ checksum = "f0c3acc6aa564495a0f2e1d59fab677cd7f81a19994cfc7f3ad0e64301560389" dependencies = [ "proc-macro2 1.0.20", "quote 1.0.7", - "syn 1.0.39", + "syn 1.0.40", ] [[package]] @@ -2987,7 +2983,7 @@ dependencies = [ "proc-macro2 1.0.20", "prost-build", "quote 1.0.7", - "syn 1.0.39", + "syn 1.0.40", ] [[package]] @@ -3197,7 +3193,7 @@ checksum = "80e0ccfc3378da0cce270c946b676a376943f5cd16aeba64568e7939806f4ada" dependencies = [ "proc-macro2 1.0.20", "quote 1.0.7", - "syn 1.0.39", + "syn 1.0.40", ] [[package]] @@ -3392,7 +3388,7 @@ dependencies = [ "log", "proc-macro2 1.0.20", "quote 1.0.7", - "syn 1.0.39", + "syn 1.0.40", "wasm-bindgen-shared", ] @@ -3426,7 +3422,7 @@ checksum = "841a6d1c35c6f596ccea1f82504a192a60378f64b3bb0261904ad8f2f5657556" dependencies = [ "proc-macro2 1.0.20", "quote 1.0.7", - "syn 1.0.39", + "syn 1.0.40", "wasm-bindgen-backend", "wasm-bindgen-shared", ] diff --git a/Cargo.toml b/Cargo.toml index 144b03c44..37b2acdc6 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -74,7 +74,7 @@ url = "2.1" features = ["tokio-comp"] [dev-dependencies] -# TODO: upgrade this dependency when awc will be updated +# TODO: upgrade this dependency when awc 2.0 will be released actix-codec = "0.2" actix-rt = "1.0" awc = "1.0" @@ -83,5 +83,4 @@ function_name = "0.2" lazy_static = "1.4" mockall = "0.8" serial_test = "0.5" -serial_test_derive = "0.5" tempfile = "3.1" diff --git a/jason/src/api/mod.rs b/jason/src/api/mod.rs index 9e3e20dee..0b558cbd2 100644 --- a/jason/src/api/mod.rs +++ b/jason/src/api/mod.rs @@ -65,7 +65,7 @@ impl Jason { spawn_local({ let rpc = Rc::clone(&rpc); - let inner = self.0.clone(); + let inner = Rc::clone(&self.0); async move { let reason = rpc.on_normal_close().await.unwrap_or_else(|_| { ClientDisconnect::RpcClientUnexpectedlyDropped.into() diff --git a/src/api/control/callback/service.rs b/src/api/control/callback/service.rs index 775b2dcd9..0cc03fa45 100644 --- a/src/api/control/callback/service.rs +++ b/src/api/control/callback/service.rs @@ -105,7 +105,7 @@ mod tests { use std::{convert::TryFrom as _, time::Duration}; use futures::{future, FutureExt}; - use serial_test_derive::serial; + use serial_test::serial; use tokio::time; use crate::api::control::callback::{ diff --git a/src/conf/control.rs b/src/conf/control.rs index 3500880e5..0c6bbf544 100644 --- a/src/conf/control.rs +++ b/src/conf/control.rs @@ -20,7 +20,7 @@ pub struct ControlApi { #[cfg(test)] mod spec { - use serial_test_derive::serial; + use serial_test::serial; use crate::{conf::Conf, overrided_by_env_conf}; diff --git a/src/conf/log.rs b/src/conf/log.rs index 8f6a34ff6..7f6469704 100644 --- a/src/conf/log.rs +++ b/src/conf/log.rs @@ -24,7 +24,7 @@ impl Log { #[cfg(test)] mod log_conf_specs { - use serial_test_derive::serial; + use serial_test::serial; use crate::{conf::Conf, overrided_by_env_conf}; diff --git a/src/conf/media.rs b/src/conf/media.rs index 0f741758c..701a0dab6 100644 --- a/src/conf/media.rs +++ b/src/conf/media.rs @@ -24,7 +24,7 @@ pub struct Media { mod spec { use std::time::Duration; - use serial_test_derive::serial; + use serial_test::serial; use crate::{conf::Conf, overrided_by_env_conf}; diff --git a/src/conf/mod.rs b/src/conf/mod.rs index 2adae3f48..e929bcaf1 100644 --- a/src/conf/mod.rs +++ b/src/conf/mod.rs @@ -106,7 +106,7 @@ where #[cfg(test)] pub(crate) mod spec { - use serial_test_derive::serial; + use serial_test::serial; use super::*; diff --git a/src/conf/rpc.rs b/src/conf/rpc.rs index f3caa7ca9..3f38531d5 100644 --- a/src/conf/rpc.rs +++ b/src/conf/rpc.rs @@ -46,7 +46,7 @@ pub struct Rpc { mod spec { use std::{fs, time::Duration}; - use serial_test_derive::serial; + use serial_test::serial; use crate::{ conf::{Conf, APP_CONF_PATH_ENV_VAR_NAME}, diff --git a/src/conf/server.rs b/src/conf/server.rs index 8f443066e..4d5a7cf30 100644 --- a/src/conf/server.rs +++ b/src/conf/server.rs @@ -112,7 +112,7 @@ pub struct Server { mod client_http_spec { use std::net::Ipv4Addr; - use serial_test_derive::serial; + use serial_test::serial; use crate::{conf::Conf, overrided_by_env_conf}; @@ -151,7 +151,7 @@ mod client_http_spec { mod control_grpc_spec { use std::net::Ipv4Addr; - use serial_test_derive::serial; + use serial_test::serial; use crate::{conf::Conf, overrided_by_env_conf}; diff --git a/src/conf/shutdown.rs b/src/conf/shutdown.rs index 6900500f8..1767b9a82 100644 --- a/src/conf/shutdown.rs +++ b/src/conf/shutdown.rs @@ -19,7 +19,7 @@ pub struct Shutdown { mod spec { use std::time::Duration; - use serial_test_derive::serial; + use serial_test::serial; use crate::{conf::Conf, overrided_by_env_conf}; diff --git a/src/conf/turn.rs b/src/conf/turn.rs index 716118807..d8b93dd1d 100644 --- a/src/conf/turn.rs +++ b/src/conf/turn.rs @@ -207,7 +207,7 @@ impl From for deadpool::managed::PoolConfig { #[cfg(test)] mod spec { - use serial_test_derive::serial; + use serial_test::serial; use crate::{conf::Conf, overrided_by_env_conf}; diff --git a/src/signalling/peers/traffic_watcher.rs b/src/signalling/peers/traffic_watcher.rs index c4fcd3251..8b81b7517 100644 --- a/src/signalling/peers/traffic_watcher.rs +++ b/src/signalling/peers/traffic_watcher.rs @@ -35,8 +35,8 @@ use std::{ use actix::{ Actor, Addr, AsyncContext, Handler, MailboxError, Message, SpawnHandle, }; +use async_trait::async_trait; use chrono::{DateTime, Utc}; -use futures::future::LocalBoxFuture; use medea_client_api_proto::PeerId; use crate::{ @@ -64,16 +64,17 @@ pub fn build_peers_traffic_watcher( } /// Consumes `Peer` traffic metrics for further processing. +#[async_trait(?Send)] #[cfg_attr(test, mockall::automock)] pub trait PeerTrafficWatcher: Debug + Send + Sync { /// Registers provided [`PeerConnectionStateEventsHandler`] as `Peer`s state /// messages listener, preparing [`PeerTrafficWatcher`] for registering /// `Peer`s from this [`PeerConnectionStateEventsHandler`]. - fn register_room( + async fn register_room( &self, room_id: RoomId, handler: Box, - ) -> LocalBoxFuture<'static, Result<(), MailboxError>>; + ) -> Result<(), MailboxError>; /// Unregisters [`Room`] as `Peer`s state messages listener. /// @@ -82,12 +83,12 @@ pub trait PeerTrafficWatcher: Debug + Send + Sync { /// Registers `Peer`, so that [`PeerTrafficWatcher`] will be able to /// process traffic flow events of this `Peer`. - fn register_peer( + async fn register_peer( &self, room_id: RoomId, peer_id: PeerId, should_watch_turn: bool, - ) -> LocalBoxFuture<'static, Result<(), MailboxError>>; + ) -> Result<(), MailboxError>; /// Unregisters `Peer`s, so that [`PeerTrafficWatcher`] will not be able /// to process traffic flow events of this `Peer` anymore. @@ -125,15 +126,16 @@ fn build_flow_sources(should_watch_turn: bool) -> HashSet { sources } +#[async_trait(?Send)] impl PeerTrafficWatcher for Addr { /// Sends [`RegisterRoom`] message to the [`PeersTrafficWatcherImpl`] /// returning send result. - fn register_room( + async fn register_room( &self, room_id: RoomId, handler: Box, - ) -> LocalBoxFuture<'static, Result<(), MailboxError>> { - Box::pin(self.send(RegisterRoom { room_id, handler })) + ) -> Result<(), MailboxError> { + self.send(RegisterRoom { room_id, handler }).await } /// Sends [`UnregisterRoom`] message to [`PeersTrafficWatcherImpl`]. @@ -143,17 +145,18 @@ impl PeerTrafficWatcher for Addr { /// Sends [`RegisterPeer`] message to [`PeersTrafficWatcherImpl`] returning /// send result. - fn register_peer( + async fn register_peer( &self, room_id: RoomId, peer_id: PeerId, should_watch_turn: bool, - ) -> LocalBoxFuture<'static, Result<(), MailboxError>> { - Box::pin(self.send(RegisterPeer { + ) -> Result<(), MailboxError> { + self.send(RegisterPeer { room_id, peer_id, flow_metrics_sources: build_flow_sources(should_watch_turn), - })) + }) + .await } /// Sends [`UnregisterPeers`] message to [`PeersTrafficWatcherImpl`]. From 8fbe6d2d8387fa06e4f31fddb0ea0df0d054a3ee Mon Sep 17 00:00:00 2001 From: alexlapa Date: Tue, 8 Sep 2020 10:44:18 +0300 Subject: [PATCH 8/9] corrections --- jason/src/rpc/mod.rs | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/jason/src/rpc/mod.rs b/jason/src/rpc/mod.rs index 8ecafe8f8..b02b3bd09 100644 --- a/jason/src/rpc/mod.rs +++ b/jason/src/rpc/mod.rs @@ -515,10 +515,10 @@ impl WebSocketRpcClient { // subscribe to transport close let mut transport_state_changes = transport.on_state_change(); - let weak_inner = Rc::downgrade(&self); + let weak_this = Rc::downgrade(&self); spawn_local(async move { while let Some(state) = transport_state_changes.next().await { - if let Some(this) = weak_inner.upgrade() { + if let Some(this) = weak_this.upgrade() { if let TransportState::Closed(msg) = state { this.handle_close_message(msg); } @@ -527,11 +527,11 @@ impl WebSocketRpcClient { }); // subscribe to transport message received - let this_clone = Rc::downgrade(&self); + let weak_this = Rc::downgrade(&self); let mut on_socket_message = transport.on_message(); spawn_local(async move { while let Some(msg) = on_socket_message.next().await { - if let Some(this) = this_clone.upgrade() { + if let Some(this) = weak_this.upgrade() { this.on_transport_message(msg) } } From 0c21cc1d1f040b9f7a81cdec5666b1cd6db0d6d5 Mon Sep 17 00:00:00 2001 From: tyranron Date: Tue, 8 Sep 2020 12:43:16 +0300 Subject: [PATCH 9/9] Corrections --- CHANGELOG.md | 2 +- Cargo.toml | 2 +- config.toml | 4 ++-- jason/src/api/mod.rs | 2 +- src/conf/turn.rs | 6 +++--- 5 files changed, 8 insertions(+), 8 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 27385b185..8eb7966c0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -51,7 +51,7 @@ All user visible changes to this project will be documented in this file. This p - `[media]` section to configure timeouts involved for determining media flow liveness ([#98]): - `max_lag`; - `init_timeout`. - - `turn.db.redis.username` option to configure username to authenticate on Redis database server with ([#135]). + - `turn.db.redis.user` option to configure user to authenticate on [Coturn]'s Redis database server as ([#135]). - Testing: - E2E tests for signalling ([#28]). diff --git a/Cargo.toml b/Cargo.toml index 37b2acdc6..7fe4f9089 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -74,7 +74,7 @@ url = "2.1" features = ["tokio-comp"] [dev-dependencies] -# TODO: upgrade this dependency when awc 2.0 will be released +# TODO: Upgrade this dependency when `awc` 2.0 will be released. actix-codec = "0.2" actix-rt = "1.0" awc = "1.0" diff --git a/config.toml b/config.toml index 3a51d0005..8d83ec60e 100644 --- a/config.toml +++ b/config.toml @@ -117,11 +117,11 @@ # Default: # port = 6379 -# Username to authenticate on Redis database server with. +# User to authenticate on Coturn's Redis database server as. # # Env var: MEDEA_TURN__DB__REDIS__USERNAME # Default: -# username = "" +# user = "" # Password to authenticate on Coturn's Redis database server with. # diff --git a/jason/src/api/mod.rs b/jason/src/api/mod.rs index 0b558cbd2..c9a09b6e6 100644 --- a/jason/src/api/mod.rs +++ b/jason/src/api/mod.rs @@ -70,7 +70,7 @@ impl Jason { let reason = rpc.on_normal_close().await.unwrap_or_else(|_| { ClientDisconnect::RpcClientUnexpectedlyDropped.into() }); - // TODO: Don't close all rooms when multiple rpc connections + // TODO: Don't close all rooms when multiple RPC connections // will be supported. inner .borrow_mut() diff --git a/src/conf/turn.rs b/src/conf/turn.rs index d8b93dd1d..d019e75e9 100644 --- a/src/conf/turn.rs +++ b/src/conf/turn.rs @@ -77,11 +77,11 @@ pub struct Redis { #[default = 6379] pub port: u16, - /// Username to authenticate on Redis database server with. + /// User to authenticate on Redis database server as. /// /// Defaults to empty value. #[default = ""] - pub username: Cow<'static, str>, + pub user: Cow<'static, str>, /// Password to authenticate on Redis database server with. /// @@ -105,7 +105,7 @@ pub struct Redis { impl From<&Redis> for ConnectionInfo { fn from(cf: &Redis) -> Self { Self { - username: Some(cf.username.to_string()).filter(|u| !u.is_empty()), + username: Some(cf.user.to_string()).filter(|u| !u.is_empty()), addr: Box::new(redis::ConnectionAddr::Tcp( cf.host.to_string(), cf.port,