GRIN/nym
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Chore/dependency updates (#549)

Browse Source 
* Updated all non-breaking dependencies

* Updated common/crypto dependencies

* Updated all tokio [and associated] dependencies to most recent version

* Bumped version of rand_distr

* Fixed api changes in tests

* Made clippy happier about the acronym

* Fixed the type while trying to make clippy even happier

* nightly cargo fmt
This commit is contained in:
Jędrzej Stuczyński
2021-03-29 15:32:34 +01:00
committed by GitHub
parent cc1b80229c
commit 596bc76cc6
76 changed files with 926 additions and 1383 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+433 -917
View File
File diff suppressed because it is too large Load Diff
clients/client-core/Cargo.toml
+6 -6
View File
@@ -7,14 +7,14 @@ edition = "2018"
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
[dependencies]
dirs = "2.0.2"
futures = "0.3.1"
humantime-serde = "1.0.1"
dirs = "3.0"
futures = "0.3"
humantime-serde = "1.0"
log = "0.4"
rand = { version = "0.7.3", features = ["wasm-bindgen"] }
serde = { version = "1.0.104", features = ["derive"] }
sled = "0.33"
tokio = { version = "0.2", features = ["full"] }
serde = { version = "1.0", features = ["derive"] }
sled = "0.34"
tokio = { version = "1.4", features = ["macros"] }
# internal
config = { path = "../../common/config" }
clients/client-core/src/client/cover_traffic_stream.rs
+6 -6
View File
@@ -46,7 +46,7 @@ where
/// Internal state, determined by `average_message_sending_delay`,
/// used to keep track of when a next packet should be sent out.
next_delay: time::Delay,
next_delay: Pin<Box<time::Sleep>>,
/// Channel used for sending prepared sphinx packets to `MixTrafficController` that sends them
/// out to the network without any further delays.
@@ -74,7 +74,7 @@ where
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
// it is not yet time to return a message
if Pin::new(&mut self.next_delay).poll(cx).is_pending() {
if self.next_delay.as_mut().poll(cx).is_pending() {
return Poll::Pending;
};
@@ -87,7 +87,7 @@ where
// The next interval value is `next_poisson_delay` after the one that just
// yielded.
let next = now + next_poisson_delay;
self.next_delay.reset(next);
self.next_delay.as_mut().reset(next);
Poll::Ready(Some(()))
}
@@ -112,7 +112,7 @@ impl LoopCoverTrafficStream<OsRng> {
average_ack_delay,
average_packet_delay,
average_cover_message_sending_delay,
next_delay: time::delay_for(Default::default()),
next_delay: Box::pin(time::sleep(Default::default())),
mix_tx,
our_full_destination,
rng,
@@ -166,10 +166,10 @@ impl LoopCoverTrafficStream<OsRng> {
async fn run(&mut self) {
// we should set initial delay only when we actually start the stream
self.next_delay = time::delay_for(sample_poisson_duration(
self.next_delay = Box::pin(time::sleep(sample_poisson_duration(
&mut self.rng,
self.average_cover_message_sending_delay,
));
)));
while self.next().await.is_some() {
self.on_new_message().await;
clients/client-core/src/client/real_messages_control/acknowledgement_control/action_controller.rs
+3 -5
View File
@@ -15,21 +15,19 @@
use super::PendingAcknowledgement;
use crate::client::real_messages_control::acknowledgement_control::RetransmissionRequestSender;
use futures::channel::mpsc::{self, UnboundedReceiver, UnboundedSender};
use futures::StreamExt;
use log::*;
use nonexhaustive_delayqueue::NonExhaustiveDelayQueue;
use nonexhaustive_delayqueue::{Expired, NonExhaustiveDelayQueue, QueueKey, TimerError};
use nymsphinx::chunking::fragment::FragmentIdentifier;
use nymsphinx::Delay as SphinxDelay;
use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;
use tokio::stream::StreamExt;
use tokio::time::delay_queue::{self, Expired};
use tokio::time::Error as TimerError;
pub(crate) type ActionSender = UnboundedSender<Action>;
// The actual data being sent off as well as potential key to the delay queue
type PendingAckEntry = (Arc<PendingAcknowledgement>, Option<delay_queue::Key>);
type PendingAckEntry = (Arc<PendingAcknowledgement>, Option<QueueKey>);
// we can either:
// - have a completely new set of packets we just sent and need to create entries for
clients/client-core/src/client/real_messages_control/real_traffic_stream.rs
+6 -6
View File
@@ -73,7 +73,7 @@ where
/// Internal state, determined by `average_message_sending_delay`,
/// used to keep track of when a next packet should be sent out.
next_delay: time::Delay,
next_delay: Pin<Box<time::Sleep>>,
/// Channel used for sending prepared sphinx packets to `MixTrafficController` that sends them
/// out to the network without any further delays.
@@ -128,7 +128,7 @@ where
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
// it is not yet time to return a message
if Pin::new(&mut self.next_delay).poll(cx).is_pending() {
if self.next_delay.as_mut().poll(cx).is_pending() {
return Poll::Pending;
};
@@ -141,7 +141,7 @@ where
// The next interval value is `next_poisson_delay` after the one that just
// yielded.
let next = now + next_poisson_delay;
self.next_delay.reset(next);
self.next_delay.as_mut().reset(next);
// check if we have anything immediately available
if let Some(real_available) = self.received_buffer.pop_front() {
@@ -190,7 +190,7 @@ where
config,
ack_key,
sent_notifier,
next_delay: time::delay_for(Default::default()),
next_delay: Box::pin(time::sleep(Default::default())),
mix_tx,
real_receiver,
our_full_destination,
@@ -272,10 +272,10 @@ where
// Send messages at certain rate and if no real traffic is available, send cover message.
async fn run_normal_out_queue(&mut self) {
// we should set initial delay only when we actually start the stream
self.next_delay = time::delay_for(sample_poisson_duration(
self.next_delay = Box::pin(time::sleep(sample_poisson_duration(
&mut self.rng,
self.config.average_message_sending_delay,
));
)));
while let Some(next_message) = self.next().await {
self.on_message(next_message).await;
clients/client-core/src/client/topology_control.rs
+1 -1
View File
@@ -234,7 +234,7 @@ impl TopologyRefresher {
pub fn start(mut self, handle: &Handle) -> JoinHandle<()> {
handle.spawn(async move {
loop {
tokio::time::delay_for(self.refresh_rate).await;
tokio::time::sleep(self.refresh_rate).await;
self.refresh().await;
}
})
clients/native/Cargo.toml
+4 -4
View File
@@ -22,12 +22,12 @@ clap = "2.33.0" # for the command line arguments
dirs = "3.0" # for determining default store directories in config
dotenv = "0.15.0" # for obtaining environmental variables (only used for RUST_LOG for time being)
log = "0.4" # self explanatory
pretty_env_logger = "0.3" # for formatting log messages
pretty_env_logger = "0.4" # for formatting log messages
rand = {version = "0.7.3", features = ["wasm-bindgen"]} # rng-related traits + some rng implementation to use
serde = { version = "1.0.104", features = ["derive"] } # for config serialization/deserialization
sled = "0.33" # for storage of replySURB decryption keys
tokio = { version = "0.2", features = ["full"] } # async runtime
tokio-tungstenite = "0.11.0" # websocket
sled = "0.34" # for storage of replySURB decryption keys
tokio = { version = "1.4", features = ["rt-multi-thread", "net", "signal"] } # async runtime
tokio-tungstenite = "0.14" # websocket
## internal
client-core = { path = "../client-core" }
clients/native/examples/websocket_binarysend.rs
+5 -3
View File
@@ -1,13 +1,15 @@
use futures::{SinkExt, StreamExt};
use nymsphinx::addressing::clients::Recipient;
use tokio::net::TcpStream;
use tokio_tungstenite::{connect_async, tungstenite::protocol::Message, WebSocketStream};
use tokio_tungstenite::{
connect_async, tungstenite::protocol::Message, MaybeTlsStream, WebSocketStream,
};
use websocket_requests::{requests::ClientRequest, responses::ServerResponse};
// just helpers functions that work in this very particular context because we are sending to ourselves
// and hence will always get a response back (i.e. the message we sent)
async fn send_message_and_get_response(
ws_stream: &mut WebSocketStream<TcpStream>,
ws_stream: &mut WebSocketStream<MaybeTlsStream<TcpStream>>,
req: Vec<u8>,
) -> ServerResponse {
ws_stream.send(Message::Binary(req)).await.unwrap();
@@ -18,7 +20,7 @@ async fn send_message_and_get_response(
}
}
async fn get_self_address(ws_stream: &mut WebSocketStream<TcpStream>) -> Recipient {
async fn get_self_address(ws_stream: &mut WebSocketStream<MaybeTlsStream<TcpStream>>) -> Recipient {
let self_address_request = ClientRequest::SelfAddress.serialize();
let response = send_message_and_get_response(ws_stream, self_address_request).await;
clients/native/examples/websocket_textsend.rs
+5 -3
View File
@@ -1,7 +1,9 @@
use futures::{SinkExt, StreamExt};
use serde_json::json;
use tokio::net::TcpStream;
use tokio_tungstenite::{connect_async, tungstenite::protocol::Message, WebSocketStream};
use tokio_tungstenite::{
connect_async, tungstenite::protocol::Message, MaybeTlsStream, WebSocketStream,
};
// PREFACE: in practice I don't see why you would ever want to use text api while in Rust, but example
// is here for the completion sake
@@ -9,7 +11,7 @@ use tokio_tungstenite::{connect_async, tungstenite::protocol::Message, WebSocket
// just helpers functions that work in this very particular context because we are sending to ourselves
// and hence will always get a response back (i.e. the message we sent)
async fn send_message_and_get_json_response(
ws_stream: &mut WebSocketStream<TcpStream>,
ws_stream: &mut WebSocketStream<MaybeTlsStream<TcpStream>>,
text_req: String,
) -> serde_json::Value {
ws_stream.send(Message::Text(text_req)).await.unwrap();
@@ -20,7 +22,7 @@ async fn send_message_and_get_json_response(
}
}
async fn get_self_address(ws_stream: &mut WebSocketStream<TcpStream>) -> String {
async fn get_self_address(ws_stream: &mut WebSocketStream<MaybeTlsStream<TcpStream>>) -> String {
let self_address_request = json!({ "type": "selfAddress" }).to_string();
let response = send_message_and_get_json_response(ws_stream, self_address_request).await;
clients/native/src/client/mod.rs
+25 -27
View File
@@ -106,21 +106,20 @@ impl NymClient {
info!("Starting loop cover traffic stream...");
// we need to explicitly enter runtime due to "next_delay: time::delay_for(Default::default())"
// set in the constructor which HAS TO be called within context of a tokio runtime
self.runtime
.enter(|| {
LoopCoverTrafficStream::new(
self.key_manager.ack_key(),
self.config.get_base().get_average_ack_delay(),
self.config.get_base().get_average_packet_delay(),
self.config
.get_base()
.get_loop_cover_traffic_average_delay(),
mix_tx,
self.as_mix_recipient(),
topology_accessor,
)
})
.start(self.runtime.handle());
let _guard = self.runtime.enter();
LoopCoverTrafficStream::new(
self.key_manager.ack_key(),
self.config.get_base().get_average_ack_delay(),
self.config.get_base().get_average_packet_delay(),
self.config
.get_base()
.get_loop_cover_traffic_average_delay(),
mix_tx,
self.as_mix_recipient(),
topology_accessor,
)
.start(self.runtime.handle());
}
fn start_real_traffic_controller(
@@ -153,18 +152,17 @@ impl NymClient {
// we need to explicitly enter runtime due to "next_delay: time::delay_for(Default::default())"
// set in the constructor [of OutQueueControl] which HAS TO be called within context of a tokio runtime
// When refactoring this restriction should definitely be removed.
let real_messages_controller = self.runtime.enter(|| {
RealMessagesController::new(
controller_config,
ack_receiver,
input_receiver,
mix_sender,
topology_accessor,
reply_key_storage,
)
});
real_messages_controller
.start(self.runtime.handle(), self.config.get_base().get_vpn_mode());
let _guard = self.runtime.enter();
RealMessagesController::new(
controller_config,
ack_receiver,
input_receiver,
mix_sender,
topology_accessor,
reply_key_storage,
)
.start(self.runtime.handle(), self.config.get_base().get_vpn_mode());
}
// buffer controlling all messages fetched from provider
clients/native/src/commands/init.rs
+1 -1
View File
@@ -182,7 +182,7 @@ pub fn execute(matches: &ArgMatches) {
};
// TODO: is there perhaps a way to make it work without having to spawn entire runtime?
let mut rt = tokio::runtime::Runtime::new().unwrap();
let rt = tokio::runtime::Runtime::new().unwrap();
let (shared_keys, gateway_listener) = rt.block_on(registration_fut);
config
.get_base_mut()
clients/native/src/websocket/listener.rs
+6 -8
View File
@@ -14,10 +14,8 @@
use super::handler::Handler;
use log::*;
use std::{
net::{Shutdown, SocketAddr},
sync::Arc,
};
use std::{net::SocketAddr, sync::Arc};
use tokio::io::AsyncWriteExt;
use tokio::runtime;
use tokio::{sync::Notify, task::JoinHandle};
@@ -47,7 +45,7 @@ impl Listener {
}
pub(crate) async fn run(&mut self, handler: Handler) {
let mut tcp_listener = tokio::net::TcpListener::bind(self.address)
let tcp_listener = tokio::net::TcpListener::bind(self.address)
.await
.expect("Failed to start websocket listener");
@@ -61,7 +59,7 @@ impl Listener {
}
new_conn = tcp_listener.accept() => {
match new_conn {
Ok((socket, remote_addr)) => {
Ok((mut socket, remote_addr)) => {
debug!("Received connection from {:?}", remote_addr);
if self.state.is_connected() {
warn!("tried to duplicate!");
@@ -70,7 +68,7 @@ impl Listener {
// while we only ever want to accept a single connection, we don't want
// to leave clients hanging (and also allow for reconnection if it somehow
// was dropped)
match socket.shutdown(Shutdown::Both) {
match socket.shutdown().await {
Ok(_) => trace!(
"closed the connection between attempting websocket handshake"
),
@@ -84,7 +82,7 @@ impl Listener {
let fresh_handler = handler.clone();
tokio::spawn(async move {
fresh_handler.handle_connection(socket).await;
notify_clone.notify();
notify_clone.notify_one();
});
self.state = State::Connected;
}
clients/socks5/Cargo.toml
+4 -4
View File
@@ -14,12 +14,12 @@ dirs = "3.0" # for determining default store directories in config
dotenv = "0.15.0"
futures = "0.3"
log = "0.4"
pin-project = "0.4"
pretty_env_logger = "0.3"
pin-project = "1.0"
pretty_env_logger = "0.4"
rand = { version = "0.7.3", features = ["wasm-bindgen"] }
serde = { version = "1.0", features = ["derive"] } # for config serialization/deserialization
snafu = "0.4.1"
tokio = { version = "0.2", features = ["rt-threaded"] }
snafu = "0.6"
tokio = { version = "1.4", features = ["rt-multi-thread", "net", "signal"] }
# internal
client-core = { path = "../client-core" }
clients/socks5/src/client/mod.rs
+25 -27
View File
@@ -94,21 +94,20 @@ impl NymClient {
info!("Starting loop cover traffic stream...");
// we need to explicitly enter runtime due to "next_delay: time::delay_for(Default::default())"
// set in the constructor which HAS TO be called within context of a tokio runtime
self.runtime
.enter(|| {
LoopCoverTrafficStream::new(
self.key_manager.ack_key(),
self.config.get_base().get_average_ack_delay(),
self.config.get_base().get_average_packet_delay(),
self.config
.get_base()
.get_loop_cover_traffic_average_delay(),
mix_tx,
self.as_mix_recipient(),
topology_accessor,
)
})
.start(self.runtime.handle());
let _guard = self.runtime.enter();
LoopCoverTrafficStream::new(
self.key_manager.ack_key(),
self.config.get_base().get_average_ack_delay(),
self.config.get_base().get_average_packet_delay(),
self.config
.get_base()
.get_loop_cover_traffic_average_delay(),
mix_tx,
self.as_mix_recipient(),
topology_accessor,
)
.start(self.runtime.handle());
}
fn start_real_traffic_controller(
@@ -141,18 +140,17 @@ impl NymClient {
// we need to explicitly enter runtime due to "next_delay: time::delay_for(Default::default())"
// set in the constructor [of OutQueueControl] which HAS TO be called within context of a tokio runtime
// When refactoring this restriction should definitely be removed.
let real_messages_controller = self.runtime.enter(|| {
RealMessagesController::new(
controller_config,
ack_receiver,
input_receiver,
mix_sender,
topology_accessor,
reply_key_storage,
)
});
real_messages_controller
.start(self.runtime.handle(), self.config.get_base().get_vpn_mode());
let _guard = self.runtime.enter();
RealMessagesController::new(
controller_config,
ack_receiver,
input_receiver,
mix_sender,
topology_accessor,
reply_key_storage,
)
.start(self.runtime.handle(), self.config.get_base().get_vpn_mode());
}
// buffer controlling all messages fetched from provider
clients/socks5/src/commands/init.rs
+1 -1
View File
@@ -184,7 +184,7 @@ pub fn execute(matches: &ArgMatches) {
};
// TODO: is there perhaps a way to make it work without having to spawn entire runtime?
let mut rt = tokio::runtime::Runtime::new().unwrap();
let rt = tokio::runtime::Runtime::new().unwrap();
let (shared_keys, gateway_listener) = rt.block_on(registration_fut);
config
.get_base_mut()
clients/socks5/src/socks/client.rs
+12 -11
View File
@@ -17,9 +17,10 @@ use proxy_helpers::connection_controller::{
use proxy_helpers::proxy_runner::ProxyRunner;
use rand::RngCore;
use socks5_requests::{ConnectionId, RemoteAddress, Request};
use std::net::{Shutdown, SocketAddr};
use std::io;
use std::net::SocketAddr;
use std::pin::Pin;
use tokio::prelude::*;
use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, ReadBuf};
use tokio::{self, net::TcpStream};
#[pin_project(project = StateProject)]
@@ -61,12 +62,12 @@ impl StreamState {
}
}
fn shutdown(&self, how: Shutdown) -> io::Result<()> {
async fn shutdown(&mut self) -> io::Result<()> {
// shutdown should only be called if proxy is not being run. If it is, there's some bug
// somewhere
match self {
StreamState::RunningProxy => panic!("Tried to shutdown stream while proxy is running"),
StreamState::Available(ref stream) => TcpStream::shutdown(stream, how),
StreamState::Available(ref mut stream) => TcpStream::shutdown(stream).await,
}
}
}
@@ -76,8 +77,8 @@ impl AsyncRead for StreamState {
fn poll_read(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &mut [u8],
) -> Poll<io::Result<usize>> {
buf: &mut ReadBuf<'_>,
) -> Poll<io::Result<()>> {
match self.project() {
StateProject::RunningProxy => Poll::Ready(Err(io::Error::new(
io::ErrorKind::NotFound,
@@ -190,9 +191,9 @@ impl SocksClient {
}
/// Shutdown the TcpStream to the client and end the session
pub fn shutdown(&mut self) -> Result<(), SocksProxyError> {
pub async fn shutdown(&mut self) -> Result<(), SocksProxyError> {
info!("client is shutting down its TCP stream");
self.stream.shutdown(Shutdown::Both)?;
self.stream.shutdown().await?;
Ok(())
}
@@ -210,7 +211,7 @@ impl SocksClient {
// Handle SOCKS4 requests
if header[0] != SOCKS_VERSION {
warn!("Init: Unsupported version: SOCKS{}", self.socks_version);
self.shutdown()
self.shutdown().await
}
// Valid SOCKS5
else {
@@ -390,7 +391,7 @@ impl SocksClient {
self.stream.write_all(&response).await?;
// Shutdown
self.shutdown()?;
self.shutdown().await?;
}
Ok(())
@@ -404,7 +405,7 @@ impl SocksClient {
warn!("Client has no suitable authentication methods!");
response[1] = AuthenticationMethods::NoMethods as u8;
self.stream.write_all(&response).await?;
self.shutdown()?;
self.shutdown().await?;
Err(ResponseCode::Failure.into())
}
}
clients/socks5/src/socks/request.rs
+1 -1
View File
@@ -2,7 +2,7 @@ use super::types::{AddrType, ResponseCode, SocksProxyError};
use super::{utils as socks_utils, SOCKS_VERSION};
use log::*;
use std::fmt::{self, Display};
use tokio::prelude::*;
use tokio::io::{AsyncRead, AsyncReadExt};
/// A Socks5 request hitting the proxy.
pub(crate) struct SocksRequest {
clients/socks5/src/socks/server.rs
+2 -2
View File
@@ -48,7 +48,7 @@ impl SphinxSocksServer {
input_sender: InputMessageSender,
buffer_requester: ReceivedBufferRequestSender,
) -> Result<(), SocksProxyError> {
let mut listener = TcpListener::bind(self.listening_address).await.unwrap();
let listener = TcpListener::bind(self.listening_address).await.unwrap();
info!("Serving Connections...");
// controller for managing all active connections
@@ -100,7 +100,7 @@ impl SphinxSocksServer {
if client.error(response).await.is_err() {
warn!("Failed to send error code");
};
if client.shutdown().is_err() {
if client.shutdown().await.is_err() {
warn!("Failed to shutdown TcpStream");
};
}
clients/webassembly/Cargo.toml
+1 -1
View File
@@ -45,7 +45,7 @@ console_error_panic_hook = { version = "0.1", optional = true }
wee_alloc = { version = "0.4", optional = true }
[dev-dependencies]
wasm-bindgen-test = "0.2"
wasm-bindgen-test = "0.3"
[package.metadata.wasm-pack.profile.release]
wasm-opt = false
common/client-libs/gateway-client/Cargo.toml
+13 -4
View File
@@ -19,16 +19,16 @@ gateway-requests = { path = "../../../gateway/gateway-requests" }
nymsphinx = { path = "../../nymsphinx" }
[dependencies.tungstenite]
version = "0.11"
version = "0.13"
default-features = false
# non-wasm-only dependencies
[target."cfg(not(target_arch = \"wasm32\"))".dependencies.tokio]
version = "0.2"
features = ["macros", "rt-core", "stream", "sync", "time"]
version = "1.4"
features = ["macros", "rt", "net", "sync", "time"]
[target."cfg(not(target_arch = \"wasm32\"))".dependencies.tokio-tungstenite]
version = "0.11"
version = "0.14"
# wasm-only dependencies
[target."cfg(target_arch = \"wasm32\")".dependencies.wasm-bindgen]
@@ -44,6 +44,15 @@ path = "../../wasm-utils"
[target."cfg(target_arch = \"wasm32\")".dependencies.wasm-timer]
version = "0.2"
# this is due to tungstenite using `rand` 0.8 and associated changes in `getrandom` crate
# which now does not support wasm32-unknown-unknown target by default.
# using the below, we assume our client is going to be run in environment
# containing javascript (such as a web browser or node.js).
# refer to https://docs.rs/getrandom/0.2.2/getrandom/#webassembly-support for more information
[target."cfg(target_arch = \"wasm32\")".dependencies.getrandom]
version = "0.2"
features = ["js"]
[dev-dependencies]
# for tests
#url = "2.1"
common/client-libs/gateway-client/src/client.rs
+4 -2
View File
@@ -203,7 +203,7 @@ impl GatewayClient {
}
#[cfg(not(target_arch = "wasm32"))]
tokio::time::delay_for(self.reconnection_backoff).await;
tokio::time::sleep(self.reconnection_backoff).await;
#[cfg(target_arch = "wasm32")]
if let Err(err) = wasm_timer::Delay::new(self.reconnection_backoff).await {
@@ -241,7 +241,9 @@ impl GatewayClient {
};
#[cfg(not(target_arch = "wasm32"))]
let timeout = tokio::time::delay_for(self.response_timeout_duration);
let timeout = tokio::time::sleep(self.response_timeout_duration);
#[cfg(not(target_arch = "wasm32"))]
tokio::pin!(timeout);
// technically the `wasm_timer` also works outside wasm, but unless required,
// I really prefer to just stick to tokio
common/client-libs/gateway-client/src/socket_state.rs
+7 -4
View File
@@ -27,7 +27,7 @@ use tungstenite::Message;
#[cfg(not(target_arch = "wasm32"))]
use tokio::net::TcpStream;
#[cfg(not(target_arch = "wasm32"))]
use tokio_tungstenite::WebSocketStream;
use tokio_tungstenite::{MaybeTlsStream, WebSocketStream};
#[cfg(target_arch = "wasm32")]
use wasm_bindgen_futures;
@@ -37,7 +37,7 @@ use wasm_utils::websocket::JSWebsocket;
// type alias for not having to type the whole thing every single time (and now it makes it easier
// to use different types based on compilation target)
#[cfg(not(target_arch = "wasm32"))]
type WsConn = WebSocketStream<TcpStream>;
type WsConn = WebSocketStream<MaybeTlsStream<TcpStream>>;
#[cfg(target_arch = "wasm32")]
type WsConn = JSWebsocket;
@@ -124,11 +124,14 @@ impl PartiallyDelegated {
};
};
match ret_err {
if match ret_err {
Err(err) => stream_sender.send(Err(err)),
Ok(_) => stream_sender.send(Ok(stream)),
}
.unwrap();
.is_err()
{
panic!("failed to send back `mixnet_receiver_future` result on the oneshot channel")
}
};
#[cfg(target_arch = "wasm32")]
common/client-libs/metrics-client/Cargo.toml
+3 -3
View File
@@ -8,8 +8,8 @@ edition = "2018"
[dependencies]
serde = { version = "1.0", features = ["derive"] }
reqwest = { version = "0.10", features = ["json"] }
reqwest = { version = "0.11", features = ["json"] }
[dev-dependencies]
mockito = "0.23.0"
tokio = { version = "0.2", features = ["macros"] }
mockito = "0.30"
tokio = "1.4"
common/client-libs/mixnet-client/Cargo.toml
+2 -2
View File
@@ -9,8 +9,8 @@ edition = "2018"
[dependencies]
futures = "0.3"
log = "0.4.8"
tokio = { version = "0.2", features = ["full"] }
tokio-util = { version = "0.3.1", features = ["codec"] }
tokio = { version = "1.4", features = ["time", "net", "rt"] }
tokio-util = { version = "0.6", features = ["codec"] }
# internal
nymsphinx = {path = "../../nymsphinx" }
common/client-libs/mixnet-client/src/client.rs
+2 -2
View File
@@ -26,7 +26,7 @@ use std::sync::atomic::{AtomicU32, Ordering};
use std::sync::Arc;
use std::time::Duration;
use tokio::net::TcpStream;
use tokio::time::delay_for;
use tokio::time::sleep;
use tokio_util::codec::Framed;
pub struct Config {
@@ -185,7 +185,7 @@ impl Client {
// before executing the manager, wait for what was specified, if anything
if let Some(backoff) = backoff {
trace!("waiting for {:?} before attempting connection", backoff);
delay_for(backoff).await;
sleep(backoff).await;
}
Self::manage_connection(
common/client-libs/validator-client/Cargo.toml
+4 -4
View File
@@ -8,13 +8,13 @@ edition = "2018"
[dependencies]
log = "0.4"
reqwest = { version = "0.10", features = ["json"] }
schemars = "0.7.6"
reqwest = { version = "0.11", features = ["json"] }
schemars = "0.7"
serde = { version = "1.0", features = ["derive"] }
crypto = { path = "../../crypto" }
topology = { path = "../../topology" }
[dev-dependencies]
mockito = "0.23.0"
tokio = { version = "0.2", features = ["macros"] }
mockito = "0.30"
tokio = "1.4"
common/config/Cargo.toml
+1 -1
View File
@@ -8,5 +8,5 @@ edition = "2018"
[dependencies]
handlebars = "3.0.1"
serde = { version = "1.0.104", features = ["derive"] }
serde = { version = "1.0", features = ["derive"] }
toml = "0.5.6"
common/crypto/Cargo.toml
+6 -6
View File
@@ -7,18 +7,18 @@ edition = "2018"
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
[dependencies]
aes-ctr = "0.4"
bs58 = "0.3.0"
blake3 = "0.3.5"
aes-ctr = "0.6.0"
bs58 = "0.4"
blake3 = "0.3"
#blake3 = { version = "0.3", features = ["traits-preview"]}
digest = "0.9"
generic-array = "0.14"
hkdf = "0.9"
hkdf = "0.10"
hmac = "0.8"
stream-cipher = "0.4"
cipher = "0.2"
x25519-dalek = "1.1"
ed25519-dalek = "1.0"
log = "0.4"
pretty_env_logger = "0.3"
rand = { version = "0.7.3", features = ["wasm-bindgen"] }
# internal
common/crypto/src/shared_key.rs
+1 -1
View File
@@ -14,10 +14,10 @@
use crate::asymmetric::encryption;
use crate::hkdf;
use cipher::stream::{Key, NewStreamCipher, SyncStreamCipher};
use digest::{BlockInput, FixedOutput, Reset, Update};
use generic_array::{typenum::Unsigned, ArrayLength};
use rand::{CryptoRng, RngCore};
use stream_cipher::{Key, NewStreamCipher, SyncStreamCipher};
/// Generate an ephemeral encryption keypair and perform diffie-hellman to establish
/// shared key with the remote.
common/crypto/src/symmetric/stream_cipher.rs
+2 -2
View File
@@ -4,12 +4,12 @@
#![allow(renamed_and_removed_lints)]
#![allow(unknown_lints)] // beta-nightly
#![allow(clippy::unknown_clippy_lints)] // `clippy::upper_case_acronyms` does not exist on stable just yet
use cipher::stream::{Nonce, StreamCipher, SyncStreamCipher};
use generic_array::{typenum::Unsigned, GenericArray};
use rand::{CryptoRng, RngCore};
use stream_cipher::{Nonce, StreamCipher, SyncStreamCipher};
// re-export this for ease of use
pub use stream_cipher::{Key, NewStreamCipher};
pub use cipher::stream::{Key, NewStreamCipher};
// TODO: note that this is not the most secure approach here
// we are not using nonces properly but instead "kinda" thinking of them as IVs.
common/mixnode-common/Cargo.toml
+1 -1
View File
@@ -17,4 +17,4 @@ nymsphinx-forwarding = { path = "../nymsphinx/forwarding" }
nymsphinx-framing = { path = "../nymsphinx/framing" }
nymsphinx-params = { path = "../nymsphinx/params" }
nymsphinx-types = { path = "../nymsphinx/types" }
tokio = { version = "0.2", features = ["time", "macros", "rt-core"] }
tokio = { version = "1.4", features = ["time", "macros", "rt"] }
common/mixnode-common/src/cached_packet_processor/cache.rs
+4 -4
View File
@@ -15,13 +15,13 @@
use dashmap::mapref::one::Ref;
use dashmap::DashMap;
use futures::channel::mpsc;
use futures::StreamExt;
use log::*;
use nonexhaustive_delayqueue::{Expired, NonExhaustiveDelayQueue};
use nonexhaustive_delayqueue::{Expired, NonExhaustiveDelayQueue, TimerError};
use nymsphinx_types::header::keys::RoutingKeys;
use nymsphinx_types::SharedSecret;
use std::sync::Arc;
use tokio::stream::StreamExt;
use tokio::time::{Duration, Error as TimeError};
use tokio::time::Duration;
type CachedKeys = (Option<SharedSecret>, RoutingKeys);
@@ -128,7 +128,7 @@ impl CacheInvalidator {
// pros: the lock situation will be spread more in time
// cons: possibly less efficient?
fn handle_expired(&mut self, expired: Option<Result<Expired<SharedSecret>, TimeError>>) {
fn handle_expired(&mut self, expired: Option<Result<Expired<SharedSecret>, TimerError>>) {
let expired = expired.expect("the queue has unexpectedly terminated!");
let expired_entry = expired.expect("Encountered timer issue within the runtime!");
common/nonexhaustive-delayqueue/Cargo.toml
+3 -1
View File
@@ -7,4 +7,6 @@ edition = "2018"
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
[dependencies]
tokio = { version = "0.2", features = ["stream", "time"] }
tokio = { version = "1.4", features = [] }
tokio-stream = "0.1" # this one seems to be a thing until `Stream` trait is stabilised in stdlib
tokio-util = { version = "0.6", features = ["time"] }
common/nonexhaustive-delayqueue/src/lib.rs
+5 -3
View File
@@ -15,10 +15,12 @@
use std::pin::Pin;
use std::task::{Context, Poll, Waker};
use std::time::Duration;
use tokio::stream::Stream;
pub use tokio::time::delay_queue::Expired;
use tokio::time::{delay_queue, DelayQueue, Instant};
use tokio::time::Instant;
use tokio_stream::Stream;
use tokio_util::time::{delay_queue, DelayQueue};
pub use tokio::time::error::Error as TimerError;
pub use tokio_util::time::delay_queue::Expired;
pub type QueueKey = delay_queue::Key;
/// A variant of tokio's `DelayQueue`, such that its `Stream` implementation will never return a 'None'.
common/nymsphinx/Cargo.toml
+2 -2
View File
@@ -8,7 +8,7 @@ edition = "2018"
[dependencies]
rand = { version = "0.7.3", features = ["wasm-bindgen"] }
rand_distr = "0.2.2"
rand_distr = "0.3"
nymsphinx-acknowledgements = { path = "acknowledgements" }
nymsphinx-addressing = { path = "addressing" }
@@ -30,5 +30,5 @@ topology = { path = "../topology" }
path = "framing"
[target."cfg(not(target_arch = \"wasm32\"))".dependencies.tokio]
version = "0.2"
version = "1.4"
features = ["sync"]
common/nymsphinx/acknowledgements/src/identifier.rs
+1 -2
View File
@@ -13,9 +13,8 @@
// limitations under the License.
use crate::AckKey;
use crypto::aes_ctr::stream_cipher::NewStreamCipher;
use crypto::generic_array::typenum::Unsigned;
use crypto::symmetric::stream_cipher::{self, encrypt, iv_from_slice, random_iv};
use crypto::symmetric::stream_cipher::{self, encrypt, iv_from_slice, random_iv, NewStreamCipher};
use nymsphinx_params::{
packet_sizes::PacketSize, AckEncryptionAlgorithm, SerializedFragmentIdentifier, FRAG_ID_LEN,
};
common/nymsphinx/anonymous-replies/Cargo.toml
+1 -1
View File
@@ -8,7 +8,7 @@ edition = "2018"
[dependencies]
rand = {version = "0.7.3", features = ["wasm-bindgen"]}
bs58 = "0.3"
bs58 = "0.4"
serde = "1.0"
crypto = { path = "../../crypto" }
common/nymsphinx/framing/Cargo.toml
+2 -2
View File
@@ -7,8 +7,8 @@ edition = "2018"
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
[dependencies]
bytes = "0.5"
tokio-util = { version = "0.3.1", features = ["codec"] }
bytes = "1.0"
tokio-util = { version = "0.6", features = ["codec"] }
nymsphinx-types = { path = "../types" }
nymsphinx-params = { path = "../params" }
common/pemstore/Cargo.toml
+1 -1
View File
@@ -7,4 +7,4 @@ edition = "2018"
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
[dependencies]
pem = "0.7.0"
pem = "0.8"
common/socks5/proxy-helpers/Cargo.toml
+7 -5
View File
@@ -7,14 +7,16 @@ edition = "2018"
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
[dependencies]
bytes = "0.5"
# TODO: "time" feature is only required for the delay loop which is going to go away very soon!
tokio = { version = "0.2", features = [ "tcp", "io-util", "sync", "macros", "time" ] }
bytes = "1.0"
tokio = { version = "1.4", features = [ "net", "io-util", "sync", "macros", "time" ] }
tokio-util = { version = "0.6", features = [ "io" ] } # reason for getting this guy is to to able to port to tokio 1.X more quickly by being able to use
# their `read_buf` [from the util crate] replacement rather than having to rethink/reimplement `AvailableReader` with the new AsyncRead trait definition.
# In the long run, the dependency should probably get removed in favour of pure-tokio implementation, but for time being it's fine.
futures = "0.3"
log = "0.4"
socks5-requests = { path = "../requests" }
ordered-buffer = { path = "../ordered-buffer" }
[dev-dependencies]
tokio = { version = "0.2", features = ["rt-threaded"] }
tokio-test = "0.2"
tokio = { version = "1.4", features = ["rt-multi-thread"] }
tokio-test = "0.4"
common/socks5/proxy-helpers/src/available_reader.rs
+17 -9
View File
@@ -13,6 +13,7 @@
// limitations under the License.
use bytes::{BufMut, Bytes, BytesMut};
use futures::Stream;
use std::cell::RefCell;
use std::future::Future;
use std::io;
@@ -20,8 +21,8 @@ use std::ops::DerefMut;
use std::pin::Pin;
use std::task::{Context, Poll};
use tokio::io::AsyncRead;
use tokio::stream::Stream;
use tokio::time::{delay_for, Delay, Duration, Instant};
use tokio::time::{sleep, Duration, Instant, Sleep};
use tokio_util::io::poll_read_buf;
const MAX_READ_AMOUNT: usize = 500 * 1000; // 0.5MB
const GRACE_DURATION: Duration = Duration::from_millis(1);
@@ -31,7 +32,7 @@ pub struct AvailableReader<'a, R: AsyncRead + Unpin> {
// mutably borrow both inner reader and buffer at the same time)
buf: RefCell<BytesMut>,
inner: RefCell<&'a mut R>,
grace_period: Option<Delay>,
grace_period: Option<Pin<Box<Sleep>>>,
}
impl<'a, R> AvailableReader<'a, R>
@@ -44,7 +45,7 @@ where
AvailableReader {
buf: RefCell::new(BytesMut::with_capacity(Self::BUF_INCREMENT)),
inner: RefCell::new(reader),
grace_period: Some(delay_for(GRACE_DURATION)),
grace_period: Some(Box::pin(sleep(GRACE_DURATION))),
}
}
}
@@ -59,8 +60,11 @@ impl<'a, R: AsyncRead + Unpin> Stream for AvailableReader<'a, R> {
}
// note: poll_read_buf calls `buf.advance_mut(n)`
let poll_res = Pin::new(self.inner.borrow_mut().deref_mut())
.poll_read_buf(cx, self.buf.borrow_mut().deref_mut());
let poll_res = poll_read_buf(
Pin::new(self.inner.borrow_mut().deref_mut()),
cx,
self.buf.borrow_mut().deref_mut(),
);
match poll_res {
Poll::Pending => {
@@ -84,7 +88,7 @@ impl<'a, R: AsyncRead + Unpin> Stream for AvailableReader<'a, R> {
// if exists - reset grace period
if let Some(grace_period) = self.grace_period.as_mut() {
let now = Instant::now();
grace_period.reset(now + GRACE_DURATION);
grace_period.as_mut().reset(now + GRACE_DURATION);
}
// if we read a non-0 amount, we're not done yet!
@@ -115,10 +119,10 @@ impl<'a, R: AsyncRead + Unpin> Stream for AvailableReader<'a, R> {
#[cfg(test)]
mod tests {
use super::*;
use futures::poll;
use futures::{poll, StreamExt};
use std::io::Cursor;
use std::time::Duration;
use tokio::stream::StreamExt;
use tokio::io::AsyncReadExt;
use tokio_test::assert_pending;
#[tokio::test]
@@ -161,6 +165,10 @@ mod tests {
assert_eq!(read_data, first_data_chunk);
assert_pending!(poll!(available_reader.next()));
// before dropping the mock, we need to empty it
let mut buf = vec![0u8; second_data_chunk.len()];
reader_mock.read(&mut buf).await.unwrap();
}
#[tokio::test]
common/socks5/proxy-helpers/src/connection_controller.rs
+1 -1
View File
@@ -13,11 +13,11 @@
// limitations under the License.
use futures::channel::mpsc;
use futures::StreamExt;
use log::*;
use ordered_buffer::{OrderedMessage, OrderedMessageBuffer};
use socks5_requests::ConnectionId;
use std::collections::{HashMap, HashSet};
use tokio::stream::StreamExt;
/// A generic message produced after reading from a socket/connection. It includes data that was
/// actually read alongside boolean indicating whether the connection got closed so that
common/socks5/proxy-helpers/src/proxy_runner/inbound.rs
+4 -6
View File
@@ -17,13 +17,13 @@ use super::SHUTDOWN_TIMEOUT;
use crate::available_reader::AvailableReader;
use bytes::Bytes;
use futures::FutureExt;
use futures::StreamExt;
use log::*;
use ordered_buffer::OrderedMessageSender;
use socks5_requests::ConnectionId;
use std::{io, sync::Arc};
use tokio::select;
use tokio::stream::StreamExt;
use tokio::{net::tcp::OwnedReadHalf, sync::Notify, time::delay_for};
use tokio::{net::tcp::OwnedReadHalf, sync::Notify, time::sleep};
fn send_empty_close<F, S>(
connection_id: ConnectionId,
@@ -99,9 +99,7 @@ where
{
let mut available_reader = AvailableReader::new(&mut reader);
let mut message_sender = OrderedMessageSender::new();
let shutdown_future = shutdown_notify
.notified()
.then(|_| delay_for(SHUTDOWN_TIMEOUT));
let shutdown_future = shutdown_notify.notified().then(|_| sleep(SHUTDOWN_TIMEOUT));
tokio::pin!(shutdown_future);
@@ -122,7 +120,7 @@ where
}
}
trace!("{} - inbound closed", connection_id);
shutdown_notify.notify();
shutdown_notify.notify_one();
reader
}
common/socks5/proxy-helpers/src/proxy_runner/outbound.rs
+7 -14
View File
@@ -15,17 +15,13 @@
use super::SHUTDOWN_TIMEOUT;
use crate::connection_controller::{ConnectionMessage, ConnectionReceiver};
use futures::FutureExt;
use futures::StreamExt;
use log::*;
use socks5_requests::ConnectionId;
use std::{sync::Arc, time::Duration};
use tokio::prelude::*;
use tokio::io::AsyncWriteExt;
use tokio::select;
use tokio::stream::StreamExt;
use tokio::{
net::tcp::OwnedWriteHalf,
sync::Notify,
time::{delay_for, Instant},
};
use tokio::{net::tcp::OwnedWriteHalf, sync::Notify, time::sleep, time::Instant};
const MIX_TTL: Duration = Duration::from_secs(5 * 60);
@@ -66,13 +62,10 @@ pub(super) async fn run_outbound(
connection_id: ConnectionId,
shutdown_notify: Arc<Notify>,
) -> (OwnedWriteHalf, ConnectionReceiver) {
let shutdown_future = shutdown_notify
.notified()
.then(|_| delay_for(SHUTDOWN_TIMEOUT));
let shutdown_future = shutdown_notify.notified().then(|_| sleep(SHUTDOWN_TIMEOUT));
tokio::pin!(shutdown_future);
let mut mix_timeout = delay_for(MIX_TTL);
let mut mix_timeout = Box::pin(sleep(MIX_TTL));
loop {
select! {
@@ -81,7 +74,7 @@ pub(super) async fn run_outbound(
if deal_with_message(connection_message, &mut writer, &local_destination_address, &remote_source_address, connection_id).await {
break;
}
mix_timeout.reset(Instant::now() + MIX_TTL);
mix_timeout.as_mut().reset(Instant::now() + MIX_TTL);
} else {
warn!("mix receiver is none so we already got removed somewhere. This isn't really a warning, but shouldn't happen to begin with, so please say if you see this message");
break;
@@ -100,7 +93,7 @@ pub(super) async fn run_outbound(
}
trace!("{} - outbound closed", connection_id);
shutdown_notify.notify();
shutdown_notify.notify_one();
(writer, mix_receiver)
}
common/topology/Cargo.toml
+1 -2
View File
@@ -7,9 +7,8 @@ edition = "2018"
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
[dependencies]
bs58 = "0.3.0"
bs58 = "0.4"
log = "0.4"
pretty_env_logger = "0.3"
rand = { version = "0.7.3", features = ["wasm-bindgen"] }
## internal
common/version-checker/Cargo.toml
+1 -1
View File
@@ -7,4 +7,4 @@ edition = "2018"
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
[dependencies]
semver = "0.9.0"
semver = "0.11"
common/wasm-utils/Cargo.toml
+1 -1
View File
@@ -14,7 +14,7 @@ wasm-bindgen-futures = "0.4"
# we don't want entire tokio-tungstenite, tungstenite itself is just fine - we just want message and error enums
[dependencies.tungstenite]
version = "0.11"
version = "0.13"
default-features = false
[dependencies.web-sys]
common/wasm-utils/src/websocket/mod.rs
+3 -4
View File
@@ -15,7 +15,6 @@
use crate::websocket::state::State;
use crate::{console_error, console_log};
use futures::{Sink, Stream};
use std::borrow::Cow;
use std::cell::RefCell;
use std::collections::VecDeque;
use std::io;
@@ -49,9 +48,8 @@ fn try_message_event_into_ws_message(msg_event: MessageEvent) -> Result<WsMessag
Some(text) => Ok(WsMessage::Text(text)),
None => Err(WsError::Utf8),
},
_ => Err(WsError::Protocol(Cow::from(
"received a websocket message that is neither a String, ArrayBuffer or a Blob",
))),
// "received a websocket message that is neither a String, ArrayBuffer or a Blob"
_ => Err(WsError::Io(io::Error::from(io::ErrorKind::InvalidInput))),
}
}
@@ -71,6 +69,7 @@ fn try_message_event_into_ws_message(msg_event: MessageEvent) -> Result<WsMessag
unsafe impl Send for JSWebsocket {}
#[derive(Debug)]
#[allow(clippy::upper_case_acronyms)]
pub struct JSWebsocket {
socket: web_sys::WebSocket,
explorer/Cargo.lock
Generated
+197 -181
View File
@@ -1,5 +1,7 @@
# This file is automatically @generated by Cargo.
# It is not intended for manual editing.
version = 3
[[package]]
name = "aead"
version = "0.2.0"
@@ -97,6 +99,12 @@ version = "0.12.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3441f0f7b02788e948e47f457ca01f1d7e6d92c693bc132c22b087d3141c03ff"
[[package]]
name = "base64"
version = "0.13.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "904dfeac50f3cdaba28fc6f57fdcddb75f49ed61346676a78c4ffe55877802fd"
[[package]]
name = "bitflags"
version = "1.2.1"
@@ -166,6 +174,12 @@ version = "0.5.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0e4cec68f03f32e44924783795810fa50a7035d8c8ebe78580ad7e6c703fba38"
[[package]]
name = "bytes"
version = "1.0.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b700ce4376041dcd0a327fd0097c41095743c4c8af8887265942faf1100bd040"
[[package]]
name = "cc"
version = "1.0.61"
@@ -210,7 +224,7 @@ dependencies = [
"hkdf",
"hmac",
"percent-encoding 2.1.0",
"rand",
"rand 0.7.3",
"sha2",
"time",
]
@@ -297,12 +311,6 @@ dependencies = [
"generic-array 0.14.4",
]
[[package]]
name = "dtoa"
version = "0.4.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "134951f4028bdadb9b84baf4232681efbf277da25144b9b0ad65df75946c422b"
[[package]]
name = "encoding_rs"
version = "0.8.26"
@@ -448,7 +456,7 @@ dependencies = [
"futures-macro",
"futures-sink",
"futures-task",
"pin-project 1.0.1",
"pin-project",
"pin-utils",
"proc-macro-hack",
"proc-macro-nested",
@@ -485,6 +493,17 @@ dependencies = [
"wasi 0.9.0+wasi-snapshot-preview1",
]
[[package]]
name = "getrandom"
version = "0.2.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c9495705279e7140bf035dde1f6e750c162df8b625267cd52cc44e0b156732c8"
dependencies = [
"cfg-if 1.0.0",
"libc",
"wasi 0.10.0+wasi-snapshot-preview1",
]
[[package]]
name = "ghash"
version = "0.2.3"
@@ -502,11 +521,11 @@ checksum = "9b919933a397b79c37e33b77bb2aa3dc8eb6e165ad809e58ff75bc7db2e34574"
[[package]]
name = "h2"
version = "0.2.7"
version = "0.3.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5e4728fd124914ad25e99e3d15a9361a879f6620f63cb56bbb08f95abb97a535"
checksum = "fc018e188373e2777d0ef2467ebff62a08e66c3f5857b23c8fbec3018210dc00"
dependencies = [
"bytes",
"bytes 1.0.1",
"fnv",
"futures-core",
"futures-sink",
@@ -517,7 +536,6 @@ dependencies = [
"tokio",
"tokio-util",
"tracing",
"tracing-futures",
]
[[package]]
@@ -561,19 +579,20 @@ version = "0.2.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "28d569972648b2c512421b5f2a405ad6ac9666547189d0c5477a3f200f3e02f9"
dependencies = [
"bytes",
"bytes 0.5.6",
"fnv",
"itoa",
]
[[package]]
name = "http-body"
version = "0.3.1"
version = "0.4.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "13d5ff830006f7646652e057693569bfe0d51760c0085a071769d142a205111b"
checksum = "5dfb77c123b4e2f72a2069aeae0b4b4949cc7e966df277813fc16347e7549737"
dependencies = [
"bytes",
"bytes 1.0.1",
"http",
"pin-project-lite 0.2.6",
]
[[package]]
@@ -603,17 +622,17 @@ dependencies = [
"time",
"traitobject",
"typeable",
"unicase 1.4.2",
"unicase",
"url 1.7.2",
]
[[package]]
name = "hyper"
version = "0.13.9"
version = "0.14.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f6ad767baac13b44d4529fcf58ba2cd0995e36e7b435bc5b039de6f47e880dbf"
checksum = "8bf09f61b52cfcf4c00de50df88ae423d6c02354e385a86341133b5338630ad1"
dependencies = [
"bytes",
"bytes 1.0.1",
"futures-channel",
"futures-core",
"futures-util",
@@ -623,7 +642,7 @@ dependencies = [
"httparse",
"httpdate",
"itoa",
"pin-project 1.0.1",
"pin-project",
"socket2",
"tokio",
"tower-service",
@@ -633,15 +652,15 @@ dependencies = [
[[package]]
name = "hyper-tls"
version = "0.4.3"
version = "0.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d979acc56dcb5b8dddba3917601745e877576475aa046df3226eabdecef78eed"
checksum = "d6183ddfa99b85da61a140bea0efc93fdf56ceaa041b37d553518030827f9905"
dependencies = [
"bytes",
"hyper 0.13.9",
"bytes 1.0.1",
"hyper 0.14.5",
"native-tls",
"tokio",
"tokio-tls",
"tokio-native-tls",
]
[[package]]
@@ -698,11 +717,11 @@ dependencies = [
[[package]]
name = "input_buffer"
version = "0.3.1"
version = "0.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "19a8a95243d5a0398cae618ec29477c6e3cb631152be5c19481f80bc71559754"
checksum = "f97967975f448f1a7ddb12b0bc41069d09ed6a1c161a92687e057325db35d413"
dependencies = [
"bytes",
"bytes 1.0.1",
]
[[package]]
@@ -765,9 +784,9 @@ checksum = "830d08ce1d1d941e6b30645f1a0eb5643013d835ce3779a5fc208261dbe10f55"
[[package]]
name = "libc"
version = "0.2.80"
version = "0.2.91"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4d58d1b70b004888f764dfbf6a26a3b0342a1632d33968e4a179d8011c760614"
checksum = "8916b1f6ca17130ec6568feccee27c156ad12037880833a3b842a823236502e7"
[[package]]
name = "log"
@@ -814,16 +833,6 @@ version = "0.3.16"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2a60c7ce501c71e03a9c9c0d35b861413ae925bd979cc7a4e30d060069aaac8d"
[[package]]
name = "mime_guess"
version = "2.0.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2684d4c2e97d99848d30b324b00c8fcc7e5c897b7cbb5819b09e7c90e8baf212"
dependencies = [
"mime 0.3.16",
"unicase 2.6.0",
]
[[package]]
name = "mio"
version = "0.6.22"
@@ -843,6 +852,19 @@ dependencies = [
"winapi 0.2.8",
]
[[package]]
name = "mio"
version = "0.7.11"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cf80d3e903b34e0bd7282b218398aec54e082c840d9baf8339e0080a0c542956"
dependencies = [
"libc",
"log 0.4.11",
"miow 0.3.7",
"ntapi",
"winapi 0.3.9",
]
[[package]]
name = "mio-extras"
version = "2.0.6"
@@ -851,33 +873,10 @@ checksum = "52403fe290012ce777c4626790c8951324a2b9e3316b3143779c72b029742f19"
dependencies = [
"lazycell",
"log 0.4.11",
"mio",
"mio 0.6.22",
"slab",
]
[[package]]
name = "mio-named-pipes"
version = "0.1.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0840c1c50fd55e521b247f949c241c9997709f23bd7f023b9762cd561e935656"
dependencies = [
"log 0.4.11",
"mio",
"miow 0.3.5",
"winapi 0.3.9",
]
[[package]]
name = "mio-uds"
version = "0.6.8"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "afcb699eb26d4332647cc848492bbc15eafb26f08d0304550d5aa1f612e066f0"
dependencies = [
"iovec",
"libc",
"mio",
]
[[package]]
name = "miow"
version = "0.2.1"
@@ -892,19 +891,18 @@ dependencies = [
[[package]]
name = "miow"
version = "0.3.5"
version = "0.3.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "07b88fb9795d4d36d62a012dfbf49a8f5cf12751f36d31a9dbe66d528e58979e"
checksum = "b9f1c5b025cda876f66ef43a113f91ebc9f4ccef34843000e0adf6ebbab84e21"
dependencies = [
"socket2",
"winapi 0.3.9",
]
[[package]]
name = "native-tls"
version = "0.2.5"
version = "0.2.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1a1cda389c26d6b88f3d2dc38aa1b750fe87d298cc5d795ec9e975f402f00372"
checksum = "b8d96b2e1c8da3957d58100b09f102c6d9cfdfced01b7ec5a8974044bb09dbd4"
dependencies = [
"lazy_static",
"libc",
@@ -941,12 +939,21 @@ dependencies = [
"fsevent-sys",
"inotify",
"libc",
"mio",
"mio 0.6.22",
"mio-extras",
"walkdir",
"winapi 0.3.9",
]
[[package]]
name = "ntapi"
version = "0.3.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3f6bb902e437b6d86e03cce10a7e2af662292c5dfef23b65899ea3ac9354ad44"
dependencies = [
"winapi 0.3.9",
]
[[package]]
name = "num_cpus"
version = "1.13.0"
@@ -968,7 +975,7 @@ dependencies = [
"rocket",
"rocket_contrib",
"tokio",
"tokio-native-tls",
"tokio-stream",
"tokio-tungstenite",
]
@@ -1057,33 +1064,13 @@ version = "2.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d4fd5641d01c8f18a23da7b6fe29298ff4b55afcccdf78973b24cf3175fee32e"
[[package]]
name = "pin-project"
version = "0.4.27"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2ffbc8e94b38ea3d2d8ba92aea2983b503cd75d0888d75b86bb37970b5698e15"
dependencies = [
"pin-project-internal 0.4.27",
]
[[package]]
name = "pin-project"
version = "1.0.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ee41d838744f60d959d7074e3afb6b35c7456d0f61cad38a24e35e6553f73841"
dependencies = [
"pin-project-internal 1.0.1",
]
[[package]]
name = "pin-project-internal"
version = "0.4.27"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "65ad2ae56b6abe3a1ee25f15ee605bacadb9a764edaba9c2bf4103800d4a1895"
dependencies = [
"proc-macro2 1.0.24",
"quote 1.0.7",
"syn 1.0.48",
"pin-project-internal",
]
[[package]]
@@ -1103,6 +1090,12 @@ version = "0.1.11"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c917123afa01924fc84bb20c4c03f004d9c38e5127e3c039bbf7f4b9c76a2f6b"
[[package]]
name = "pin-project-lite"
version = "0.2.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "dc0e1f259c92177c30a4c9d177246edd0a3568b25756a977d0632cf8fa37e905"
[[package]]
name = "pin-utils"
version = "0.1.0"
@@ -1185,11 +1178,23 @@ version = "0.7.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6a6b1679d49b24bbfe0c803429aa1874472f50d9b363131f0e89fc356b544d03"
dependencies = [
"getrandom",
"getrandom 0.1.15",
"libc",
"rand_chacha",
"rand_core",
"rand_hc",
"rand_chacha 0.2.2",
"rand_core 0.5.1",
"rand_hc 0.2.0",
]
[[package]]
name = "rand"
version = "0.8.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0ef9e7e66b4468674bfcb0c81af8b7fa0bb154fa9f28eb840da5c447baeb8d7e"
dependencies = [
"libc",
"rand_chacha 0.3.0",
"rand_core 0.6.2",
"rand_hc 0.3.0",
]
[[package]]
@@ -1199,7 +1204,17 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f4c8ed856279c9737206bf725bf36935d8666ead7aa69b52be55af369d193402"
dependencies = [
"ppv-lite86",
"rand_core",
"rand_core 0.5.1",
]
[[package]]
name = "rand_chacha"
version = "0.3.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e12735cf05c9e10bf21534da50a147b924d555dc7a547c42e6bb2d5b6017ae0d"
dependencies = [
"ppv-lite86",
"rand_core 0.6.2",
]
[[package]]
@@ -1208,7 +1223,16 @@ version = "0.5.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "90bde5296fc891b0cef12a6d03ddccc162ce7b2aff54160af9338f8d40df6d19"
dependencies = [
"getrandom",
"getrandom 0.1.15",
]
[[package]]
name = "rand_core"
version = "0.6.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "34cf66eb183df1c5876e2dcf6b13d57340741e8dc255b48e40a26de954d06ae7"
dependencies = [
"getrandom 0.2.2",
]
[[package]]
@@ -1217,7 +1241,16 @@ version = "0.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ca3129af7b92a17112d59ad498c6f81eaf463253766b90396d39ea7a39d6613c"
dependencies = [
"rand_core",
"rand_core 0.5.1",
]
[[package]]
name = "rand_hc"
version = "0.3.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3190ef7066a446f2e7f42e239d161e905420ccab01eb967c9eb27d21b2322a73"
dependencies = [
"rand_core 0.6.2",
]
[[package]]
@@ -1237,32 +1270,31 @@ dependencies = [
[[package]]
name = "reqwest"
version = "0.10.8"
version = "0.11.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e9eaa17ac5d7b838b7503d118fa16ad88f440498bf9ffe5424e621f93190d61e"
checksum = "bf12057f289428dbf5c591c74bf10392e4a8003f993405a902f20117019022d4"
dependencies = [
"base64 0.12.3",
"bytes",
"base64 0.13.0",
"bytes 1.0.1",
"encoding_rs",
"futures-core",
"futures-util",
"http",
"http-body",
"hyper 0.13.9",
"hyper 0.14.5",
"hyper-tls",
"ipnet",
"js-sys",
"lazy_static",
"log 0.4.11",
"mime 0.3.16",
"mime_guess",
"native-tls",
"percent-encoding 2.1.0",
"pin-project-lite",
"pin-project-lite 0.2.6",
"serde",
"serde_urlencoded",
"tokio",
"tokio-tls",
"tokio-native-tls",
"url 2.2.0",
"wasm-bindgen",
"wasm-bindgen-futures",
@@ -1409,14 +1441,14 @@ dependencies = [
[[package]]
name = "serde_urlencoded"
version = "0.6.1"
version = "0.7.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9ec5d77e2d4c73717816afac02670d5c4f534ea95ed430442cad02e7a6e32c97"
checksum = "edfa57a7f8d9c1d260a549e7224100f6c43d43f9103e06dd8b4095a9b2b43ce9"
dependencies = [
"dtoa",
"form_urlencoded",
"itoa",
"ryu",
"serde",
"url 2.2.0",
]
[[package]]
@@ -1444,15 +1476,6 @@ dependencies = [
"opaque-debug 0.2.3",
]
[[package]]
name = "signal-hook-registry"
version = "1.2.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ce32ea0c6c56d5eacaeb814fbed9960547021d3edd010ded1425f180536b20ab"
dependencies = [
"libc",
]
[[package]]
name = "slab"
version = "0.4.2"
@@ -1467,13 +1490,11 @@ checksum = "fbee7696b84bbf3d89a1c2eccff0850e3047ed46bfcd2e92c29a2d074d57e252"
[[package]]
name = "socket2"
version = "0.3.15"
version = "0.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b1fa70dc5c8104ec096f4fe7ede7a221d35ae13dcd19ba1ad9a81d2cab9a1c44"
checksum = "9e3dfc207c526015c632472a77be09cf1b6e46866581aecae5cc38fb4235dea2"
dependencies = [
"cfg-if 0.1.10",
"libc",
"redox_syscall",
"winapi 0.3.9",
]
@@ -1531,7 +1552,7 @@ checksum = "7a6e24d9338a0a5be79593e2fa15a648add6138caa803e2d5bc782c371732ca9"
dependencies = [
"cfg-if 0.1.10",
"libc",
"rand",
"rand 0.7.3",
"redox_syscall",
"remove_dir_all",
"winapi 0.3.9",
@@ -1546,6 +1567,26 @@ dependencies = [
"unicode-width",
]
[[package]]
name = "thiserror"
version = "1.0.24"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e0f4a65597094d4483ddaed134f409b2cb7c1beccf25201a9f73c719254fa98e"
dependencies = [
"thiserror-impl",
]
[[package]]
name = "thiserror-impl"
version = "1.0.24"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7765189610d8241a44529806d6fd1f2e0a08734313a35d5b3a556f92b381f3c0"
dependencies = [
"proc-macro2 1.0.24",
"quote 1.0.7",
"syn 1.0.48",
]
[[package]]
name = "time"
version = "0.1.44"
@@ -1565,33 +1606,25 @@ checksum = "238ce071d267c5710f9d31451efec16c5ee22de34df17cc05e56cbc92e967117"
[[package]]
name = "tokio"
version = "0.2.22"
version = "1.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5d34ca54d84bf2b5b4d7d31e901a8464f7b60ac145a284fba25ceb801f2ddccd"
checksum = "134af885d758d645f0f0505c9a8b3f9bf8a348fd822e112ab5248138348f1722"
dependencies = [
"bytes",
"fnv",
"futures-core",
"iovec",
"lazy_static",
"autocfg",
"bytes 1.0.1",
"libc",
"memchr",
"mio",
"mio-named-pipes",
"mio-uds",
"mio 0.7.11",
"num_cpus",
"pin-project-lite",
"signal-hook-registry",
"slab",
"pin-project-lite 0.2.6",
"tokio-macros",
"winapi 0.3.9",
]
[[package]]
name = "tokio-macros"
version = "0.2.5"
version = "1.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f0c3acc6aa564495a0f2e1d59fab677cd7f81a19994cfc7f3ad0e64301560389"
checksum = "caf7b11a536f46a809a8a9f0bb4237020f70ecbf115b842360afb127ea2fda57"
dependencies = [
"proc-macro2 1.0.24",
"quote 1.0.7",
@@ -1600,34 +1633,36 @@ dependencies = [
[[package]]
name = "tokio-native-tls"
version = "0.1.0"
version = "0.3.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cd608593a919a8e05a7d1fc6df885e40f6a88d3a70a3a7eff23ff27964eda069"
checksum = "f7d995660bd2b7f8c1568414c1126076c13fbb725c40112dc0120b78eb9b717b"
dependencies = [
"native-tls",
"tokio",
]
[[package]]
name = "tokio-tls"
version = "0.3.1"
name = "tokio-stream"
version = "0.1.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9a70f4fcd7b3b24fb194f837560168208f669ca8cb70d0c4b862944452396343"
checksum = "e177a5d8c3bf36de9ebe6d58537d8879e964332f93fb3339e43f618c81361af0"
dependencies = [
"native-tls",
"futures-core",
"pin-project-lite 0.2.6",
"tokio",
"tokio-util",
]
[[package]]
name = "tokio-tungstenite"
version = "0.11.0"
version = "0.14.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6d9e878ad426ca286e4dcae09cbd4e1973a7f8987d97570e2469703dd7f5720c"
checksum = "1e96bb520beab540ab664bd5a9cfeaa1fcd846fa68c830b42e2c8963071251d2"
dependencies = [
"futures-util",
"log 0.4.11",
"native-tls",
"pin-project 0.4.27",
"pin-project",
"tokio",
"tokio-native-tls",
"tungstenite",
@@ -1635,15 +1670,15 @@ dependencies = [
[[package]]
name = "tokio-util"
version = "0.3.1"
version = "0.6.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "be8242891f2b6cbef26a2d7e8605133c2c554cd35b3e4948ea892d6d68436499"
checksum = "5143d049e85af7fbc36f5454d990e62c2df705b3589f123b71f441b6b59f443f"
dependencies = [
"bytes",
"bytes 1.0.1",
"futures-core",
"futures-sink",
"log 0.4.11",
"pin-project-lite",
"pin-project-lite 0.2.6",
"tokio",
]
@@ -1669,8 +1704,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b0987850db3733619253fe60e17cb59b82d37c7e6c0236bb81e4d6b87c879f27"
dependencies = [
"cfg-if 0.1.10",
"log 0.4.11",
"pin-project-lite",
"pin-project-lite 0.1.11",
"tracing-core",
]
@@ -1683,16 +1717,6 @@ dependencies = [
"lazy_static",
]
[[package]]
name = "tracing-futures"
version = "0.2.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ab7bb6f14721aa00656086e9335d363c5c8747bae02ebe32ea2c7dece5689b4c"
dependencies = [
"pin-project 0.4.27",
"tracing",
]
[[package]]
name = "traitobject"
version = "0.1.0"
@@ -1707,20 +1731,21 @@ checksum = "59547bce71d9c38b83d9c0e92b6066c4253371f15005def0c30d9657f50c7642"
[[package]]
name = "tungstenite"
version = "0.11.1"
version = "0.13.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f0308d80d86700c5878b9ef6321f020f29b1bb9d5ff3cab25e75e23f3a492a23"
checksum = "5fe8dada8c1a3aeca77d6b51a4f1314e0f4b8e438b7b1b71e3ddaca8080e4093"
dependencies = [
"base64 0.12.3",
"base64 0.13.0",
"byteorder",
"bytes",
"bytes 1.0.1",
"http",
"httparse",
"input_buffer",
"log 0.4.11",
"native-tls",
"rand",
"rand 0.8.3",
"sha-1",
"thiserror",
"url 2.2.0",
"utf-8",
]
@@ -1746,15 +1771,6 @@ dependencies = [
"version_check 0.1.5",
]
[[package]]
name = "unicase"
version = "2.6.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "50f37be617794602aabbeee0be4f259dc1778fabe05e2d67ee8f79326d5cb4f6"
dependencies = [
"version_check 0.9.2",
]
[[package]]
name = "unicode-bidi"
version = "0.3.4"
explorer/Cargo.toml
+6 -8
View File
@@ -11,11 +11,9 @@ clap = "2.33"
# no point in importing entire futures crate
futures-util = "0.3"
log = "0.4"
reqwest = "0.10.8"
rocket = "0.4.5"
rocket_contrib = "0.4.5"
tokio = { version = "0.2", features = ["full"] }
tokio-native-tls = "0.1.0"
tokio-tungstenite = {version = "0.11", features = ["tls"] }
# tungstenite = "0.11"
reqwest = "0.11"
rocket = "0.4"
rocket_contrib = "0.4"
tokio = { version = "1.4", features = ["net", "rt-multi-thread", "macros"] }
tokio-tungstenite = {version = "0.14", features = ["native-tls"] }
tokio-stream = { version = "0.1", features = [ "sync" ] }
explorer/src/websockets/client.rs
+6 -6
View File
@@ -1,12 +1,11 @@
use futures_util::StreamExt;
use log::*;
use tokio::net::TcpStream;
use tokio::stream::StreamExt;
use tokio::sync::broadcast;
use tokio::time::{delay_for, Duration};
use tokio_native_tls::TlsStream;
use tokio::time::{sleep, Duration};
use tokio_tungstenite::tungstenite::{Error as WsError, Message};
use tokio_tungstenite::WebSocketStream;
use tokio_tungstenite::{connect_async, stream::Stream};
use tokio_tungstenite::{connect_async, MaybeTlsStream};
pub(crate) type WsItem = Result<Message, WsError>;
const MAX_RECONNECTION_ATTEMPTS: u32 = 10;
@@ -16,7 +15,8 @@ const RECONNECTION_BACKOFF: Duration = Duration::from_secs(10);
/// All metrics messages get copied out to this dashboard instance's clients.
pub(crate) struct MetricsWebsocketClient {
metrics_address: String,
metrics_upstream: WebSocketStream<Stream<TcpStream, TlsStream<TcpStream>>>,
// metrics_upstream: WebSocketStream<Stream<TcpStream, TlsStream<TcpStream>>>,
metrics_upstream: WebSocketStream<MaybeTlsStream<TcpStream>>,
broadcaster: broadcast::Sender<Message>,
reconnection_attempt: u32,
@@ -52,7 +52,7 @@ impl MetricsWebsocketClient {
}
// use linear backoff to try to reconnect asap
delay_for(RECONNECTION_BACKOFF * self.reconnection_attempt).await;
sleep(RECONNECTION_BACKOFF * self.reconnection_attempt).await;
let ws_stream = match connect_async(&self.metrics_address).await {
Ok((ws_stream, _)) => ws_stream,
explorer/src/websockets/server.rs
+6 -4
View File
@@ -3,6 +3,7 @@ use log::*;
use std::{io::Error as IoError, net::SocketAddr};
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::broadcast;
use tokio_stream::wrappers::BroadcastStream;
use tokio_tungstenite::accept_async;
use tokio_tungstenite::tungstenite::Message;
@@ -14,13 +15,13 @@ pub struct DashboardWebsocketServer {
impl DashboardWebsocketServer {
pub fn new(port: u16, sender: broadcast::Sender<Message>) -> DashboardWebsocketServer {
let addr = format!("[::]:{}", port);
DashboardWebsocketServer { addr, sender }
DashboardWebsocketServer { sender, addr }
}
pub async fn start(self) -> Result<(), IoError> {
let try_socket = TcpListener::bind(&self.addr).await;
let mut listener = try_socket?;
let listener = try_socket?;
info!("starting to listen on {}", self.addr);
while let Ok((stream, addr)) = listener.accept().await {
tokio::spawn(Self::handle_connection(
@@ -36,7 +37,7 @@ impl DashboardWebsocketServer {
async fn handle_connection(
stream: TcpStream,
addr: SocketAddr,
mut receiver: broadcast::Receiver<Message>,
receiver: broadcast::Receiver<Message>,
) {
let mut ws_stream = match accept_async(stream).await {
Ok(ws_stream) => ws_stream,
@@ -50,7 +51,8 @@ impl DashboardWebsocketServer {
};
info!("client connected from {}", addr);
while let Some(message) = receiver.next().await {
let mut broadcast_stream = BroadcastStream::new(receiver);
while let Some(message) = broadcast_stream.next().await {
let message = message.expect("the websocket broadcaster is dead!");
if let Err(err) = ws_stream.send(message).await {
warn!(
gateway/Cargo.toml
+7 -10
View File
@@ -11,19 +11,20 @@ edition = "2018"
[dependencies]
clap = "2.33.0"
dirs = "2.0.2"
dirs = "3.0"
dashmap = "4.0"
dotenv = "0.15.0"
futures = "0.3"
humantime-serde = "1.0.1"
log = "0.4"
pretty_env_logger = "0.3"
pretty_env_logger = "0.4"
rand = "0.7"
serde = { version = "1.0.104", features = ["derive"] }
sled = "0.31"
tokio = { version = "0.2", features = ["full"] }
tokio-util = { version = "0.3.1", features = ["codec"] }
tokio-tungstenite = "0.11"
sled = "0.34"
tokio = { version = "1.4", features = [ "rt-multi-thread", "net", "signal", "fs" ] }
tokio-util = { version = "0.6", features = [ "codec" ] }
tokio-stream = { version = "0.1", features = [ "fs" ] }
tokio-tungstenite = "0.14"
# internal
config = { path = "../common/config" }
@@ -35,7 +36,3 @@ nymsphinx = { path = "../common/nymsphinx" }
pemstore = { path = "../common/pemstore" }
validator-client-rest = { path = "../common/client-libs/validator-client-rest" }
version-checker = { path = "../common/version-checker" }
[dependencies.tungstenite]
version = "0.11"
default-features = false
gateway/gateway-requests/Cargo.toml
+1 -1
View File
@@ -22,7 +22,7 @@ crypto = { path = "../../common/crypto" }
pemstore = { path = "../../common/pemstore" }
[dependencies.tungstenite]
version = "0.11"
version = "0.13"
default-features = false
gateway/src/node/client_handling/websocket/connection_handler.rs
+2 -2
View File
@@ -10,7 +10,7 @@ use crate::node::client_handling::websocket::message_receiver::{
use crypto::asymmetric::identity;
use futures::{
channel::{mpsc, oneshot},
SinkExt,
SinkExt, StreamExt,
};
use gateway_requests::authentication::encrypted_address::EncryptedAddressBytes;
use gateway_requests::authentication::iv::AuthenticationIV;
@@ -24,7 +24,7 @@ use nymsphinx::DestinationAddressBytes;
use rand::{CryptoRng, Rng};
use std::convert::TryFrom;
use std::sync::Arc;
use tokio::{prelude::*, stream::StreamExt};
use tokio::io::{AsyncRead, AsyncWrite};
use tokio_tungstenite::{
tungstenite::{protocol::Message, Error as WsError},
WebSocketStream,
gateway/src/node/client_handling/websocket/listener.rs
+1 -1
View File
@@ -30,7 +30,7 @@ impl Listener {
outbound_mix_sender: MixForwardingSender,
) {
info!("Starting websocket listener at {}", self.address);
let mut tcp_listener = tokio::net::TcpListener::bind(self.address)
let tcp_listener = tokio::net::TcpListener::bind(self.address)
.await
.expect("Failed to start websocket listener");
gateway/src/node/mixnet_handling/receiver/connection_handler.rs
+2 -2
View File
@@ -9,6 +9,7 @@ use crate::node::mixnet_handling::receiver::packet_processing::PacketProcessor;
use crate::node::storage::inboxes::{ClientStorage, StoreData};
use dashmap::DashMap;
use futures::channel::oneshot;
use futures::StreamExt;
use log::*;
use mixnet_client::forwarder::MixForwardingSender;
use mixnode_common::cached_packet_processor::processor::ProcessedFinalHop;
@@ -16,11 +17,10 @@ use nymsphinx::forwarding::packet::MixPacket;
use nymsphinx::framing::codec::SphinxCodec;
use nymsphinx::framing::packet::FramedSphinxPacket;
use nymsphinx::DestinationAddressBytes;
use std::io;
use std::net::SocketAddr;
use std::sync::Arc;
use tokio::net::TcpStream;
use tokio::prelude::*;
use tokio::stream::StreamExt;
use tokio_util::codec::Framed;
pub(crate) struct ConnectionHandler {
gateway/src/node/mixnet_handling/receiver/listener.rs
+1 -1
View File
@@ -18,7 +18,7 @@ impl Listener {
pub(crate) async fn run(&mut self, connection_handler: ConnectionHandler) {
info!("Starting mixnet listener at {}", self.address);
let mut tcp_listener = tokio::net::TcpListener::bind(self.address)
let tcp_listener = tokio::net::TcpListener::bind(self.address)
.await
.expect("Failed to start mixnet listener");
gateway/src/node/mod.rs
+1 -1
View File
@@ -159,7 +159,7 @@ impl Gateway {
pub fn run(&mut self) {
info!("Starting nym gateway!");
let mut runtime = Runtime::new().unwrap();
let runtime = Runtime::new().unwrap();
runtime.block_on(async {
if let Some(duplicate_node_key) = self.check_if_same_ip_gateway_exists().await {
gateway/src/node/storage/inboxes.rs
+3 -3
View File
@@ -12,7 +12,8 @@ use std::path::PathBuf;
use std::sync::Arc;
use tokio::fs;
use tokio::fs::File;
use tokio::prelude::*;
use tokio::io::AsyncWriteExt;
use tokio_stream::wrappers::ReadDirStream;
fn dummy_message() -> ClientFile {
ClientFile {
@@ -135,8 +136,7 @@ impl ClientStorage {
}
let mut msgs = Vec::new();
let mut read_dir = fs::read_dir(full_store_dir).await?;
let mut read_dir = ReadDirStream::new(fs::read_dir(full_store_dir).await?);
while let Some(dir_entry) = read_dir.next().await {
if let Ok(dir_entry) = dir_entry {
if !Self::is_valid_file(&dir_entry).await {
mixnode/Cargo.toml
+7 -8
View File
@@ -10,19 +10,18 @@ edition = "2018"
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
[dependencies]
bs58 = "0.3.0"
bs58 = "0.4.0"
clap = "2.33.0"
curve25519-dalek = "2.0.0"
dirs = "2.0.2"
dirs = "3.0"
dotenv = "0.15.0"
futures = "0.3.1"
futures = "0.3"
humantime-serde = "1.0.1"
log = "0.4"
pretty_env_logger = "0.3"
pretty_env_logger = "0.4"
rand = "0.7"
serde = { version = "1.0.104", features = ["derive"] }
tokio = { version = "0.2", features = ["full"] }
tokio-util = { version = "0.3.1", features = ["codec"] }
serde = { version = "1.0", features = ["derive"] }
tokio = { version = "1.4", features = [ "rt-multi-thread", "net", "signal" ] }
tokio-util = { version = "0.6", features = ["codec"] }
## internal
config = { path = "../common/config" }
mixnode/src/commands/init.rs
+1 -1
View File
@@ -131,7 +131,7 @@ async fn choose_layer(
pub fn execute(matches: &ArgMatches) {
// TODO: this should probably be made implicit by slapping `#[tokio::main]` on our main method
// and then removing runtime from mixnode itself in `run`
let mut rt = Runtime::new().unwrap();
let rt = Runtime::new().unwrap();
rt.block_on(async {
let id = matches.value_of("id").unwrap();
println!("Initialising mixnode {}...", id);
mixnode/src/node/listener/connection_handler/mod.rs
+1 -1
View File
@@ -5,6 +5,7 @@ use crate::node::listener::connection_handler::packet_processing::{
MixProcessingResult, PacketProcessor,
};
use crate::node::packet_delayforwarder::PacketDelayForwardSender;
use futures::StreamExt;
use log::*;
use nymsphinx::forwarding::packet::MixPacket;
use nymsphinx::framing::codec::SphinxCodec;
@@ -12,7 +13,6 @@ use nymsphinx::framing::packet::FramedSphinxPacket;
use nymsphinx::Delay as SphinxDelay;
use std::net::SocketAddr;
use tokio::net::TcpStream;
use tokio::stream::StreamExt;
use tokio::time::Instant;
use tokio_util::codec::Framed;
mixnode/src/node/listener/mod.rs
+1 -1
View File
@@ -19,7 +19,7 @@ impl Listener {
}
async fn run(&mut self, connection_handler: ConnectionHandler) {
let mut listener = TcpListener::bind(self.address)
let listener = TcpListener::bind(self.address)
.await
.expect("Failed to create TCP listener");
loop {
mixnode/src/node/metrics.rs
+2 -2
View File
@@ -138,11 +138,11 @@ impl MetricsSender {
{
Err(err) => {
error!("failed to send metrics - {:?}", err);
tokio::time::delay_for(METRICS_FAILURE_BACKOFF)
tokio::time::sleep(METRICS_FAILURE_BACKOFF)
}
Ok(new_interval) => {
debug!("sent metrics information");
tokio::time::delay_for(Duration::from_secs(new_interval.next_report_in))
tokio::time::sleep(Duration::from_secs(new_interval.next_report_in))
}
};
mixnode/src/node/mod.rs
+1 -1
View File
@@ -123,7 +123,7 @@ impl MixNode {
pub fn run(&mut self) {
info!("Starting nym mixnode");
let mut runtime = Runtime::new().unwrap();
let runtime = Runtime::new().unwrap();
runtime.block_on(async {
if let Some(duplicate_node_key) = self.check_if_same_ip_node_exists().await {
mixnode/src/node/packet_delayforwarder.rs
+4 -4
View File
@@ -3,11 +3,11 @@
use crate::node::metrics::MetricsReporter;
use futures::channel::mpsc;
use futures::StreamExt;
use log::*;
use nonexhaustive_delayqueue::{Expired, NonExhaustiveDelayQueue};
use nonexhaustive_delayqueue::{Expired, NonExhaustiveDelayQueue, TimerError};
use nymsphinx::forwarding::packet::MixPacket;
use tokio::stream::StreamExt;
use tokio::time::{Duration, Error as TimeError, Instant};
use tokio::time::{Duration, Instant};
// Delay + MixPacket vs Instant + MixPacket
@@ -71,7 +71,7 @@ impl DelayForwarder {
}
/// Upon packet being finished getting delayed, forward it to the mixnet.
fn handle_done_delaying(&mut self, packet: Option<Result<Expired<MixPacket>, TimeError>>) {
fn handle_done_delaying(&mut self, packet: Option<Result<Expired<MixPacket>, TimerError>>) {
// those are critical errors that I don't think can be recovered from.
let delayed = packet.expect("the queue has unexpectedly terminated!");
let delayed_packet = delayed
network-monitor/Cargo.toml
+2 -2
View File
@@ -15,11 +15,11 @@ dotenv = "0.15.0"
futures = "0.3"
log = "0.4"
pin-project = "1.0"
pretty_env_logger = "0.3"
pretty_env_logger = "0.4"
rand = "0.7"
serde = "1.0"
serde_json = "1.0"
tokio = { version = "0.2", features = ["signal", "rt-threaded", "macros"] }
tokio = { version = "1.4", features = ["rt-multi-thread", "macros", "signal", "time"] }
## internal
crypto = { path = "../common/crypto" }
network-monitor/src/monitor/mod.rs
+2 -2
View File
@@ -8,7 +8,7 @@ use crate::monitor::summary_producer::SummaryProducer;
use log::*;
use std::collections::HashSet;
use std::sync::Arc;
use tokio::time::{delay_for, interval_at, Duration, Instant};
use tokio::time::{interval_at, sleep, Duration, Instant};
use validator_client::models::mixmining::BatchMixStatus;
pub(crate) mod preparer;
@@ -91,7 +91,7 @@ impl Monitor {
debug!(target: "Monitor", "sending is over, waiting for {:?} before checking what we received", PACKET_DELIVERY_TIMEOUT);
// give the packets some time to traverse the network
delay_for(PACKET_DELIVERY_TIMEOUT).await;
sleep(PACKET_DELIVERY_TIMEOUT).await;
let received = self.received_processor.return_received().await;
network-monitor/src/monitor/receiver.rs
+1 -1
View File
@@ -5,8 +5,8 @@ use crate::gateways_reader::{GatewayChannel, GatewayMessages, GatewaysReader};
use crate::monitor::processor::ReceivedProcessorSender;
use crypto::asymmetric::identity;
use futures::channel::mpsc;
use futures::StreamExt;
use gateway_client::{AcknowledgementReceiver, MixnetMessageReceiver};
use tokio::stream::StreamExt;
pub(crate) type GatewayClientUpdateSender = mpsc::UnboundedSender<GatewayClientUpdate>;
pub(crate) type GatewayClientUpdateReceiver = mpsc::UnboundedReceiver<GatewayClientUpdate>;
network-monitor/src/monitor/sender.rs
+1 -1
View File
@@ -167,7 +167,7 @@ impl PacketSender {
client.batch_send_mix_packets(mix_packets).await?;
}
tokio::time::delay_for(TIME_CHUNK_SIZE).await;
tokio::time::sleep(TIME_CHUNK_SIZE).await;
mix_packets = retained;
}
service-providers/network-requester/Cargo.toml
+4 -4
View File
@@ -11,12 +11,12 @@ edition = "2018"
[dependencies]
clap = "2.33.0"
dirs = "2.0.2"
dirs = "3.0"
futures = "0.3"
log = "0.4"
pretty_env_logger = "0.3"
tokio = { version = "0.2", features = ["stream", "tcp", "rt-threaded", "macros"] }
tokio-tungstenite = "0.11.0"
pretty_env_logger = "0.4"
tokio = { version = "1.4", features = [ "net", "rt-multi-thread", "macros" ] }
tokio-tungstenite = "0.14"
publicsuffix = "1.5"
ipnetwork = "0.17"
service-providers/network-requester/src/allowed_hosts.rs
+2 -2
View File
@@ -48,8 +48,8 @@ impl OutboundRequestFilter {
}
}
fn fetch_domain_list() -> Result<List, errors::ErrorKind> {
Ok(publicsuffix::List::fetch()?)
fn fetch_domain_list() -> Result<List, errors::Error> {
publicsuffix::List::fetch()
}
/// Returns `true` if a host's root domain is in the `allowed_hosts` list.
service-providers/network-requester/src/connection.rs
+1 -1
View File
@@ -6,8 +6,8 @@ use nymsphinx::addressing::clients::Recipient;
use proxy_helpers::connection_controller::ConnectionReceiver;
use proxy_helpers::proxy_runner::ProxyRunner;
use socks5_requests::{ConnectionId, RemoteAddress, Response};
use std::io;
use tokio::net::TcpStream;
use tokio::prelude::*;
/// A TCP connection between the Socks5 service provider, which makes
/// outbound requests on behalf of users and returns the responses through
service-providers/network-requester/src/core.rs
+4 -5
View File
@@ -4,6 +4,7 @@
use crate::allowed_hosts::{HostsStore, OutboundRequestFilter};
use crate::connection::Connection;
use crate::websocket;
use crate::websocket::TSWebsocketStream;
use futures::channel::mpsc;
use futures::stream::{SplitSink, SplitStream};
use futures::{SinkExt, StreamExt};
@@ -14,9 +15,7 @@ use proxy_helpers::connection_controller::{Controller, ControllerCommand, Contro
use socks5_requests::{ConnectionId, Request, Response};
use std::path::PathBuf;
use std::sync::atomic::{AtomicUsize, Ordering};
use tokio::net::TcpStream;
use tokio_tungstenite::tungstenite::protocol::Message;
use tokio_tungstenite::WebSocketStream;
use websocket::WebsocketConnectionError;
use websocket_requests::{requests::ClientRequest, responses::ServerResponse};
@@ -51,7 +50,7 @@ impl ServiceProvider {
/// Listens for any messages from `mix_reader` that should be written back to the mix network
/// via the `websocket_writer`.
async fn mixnet_response_listener(
mut websocket_writer: SplitSink<WebSocketStream<TcpStream>, Message>,
mut websocket_writer: SplitSink<TSWebsocketStream, Message>,
mut mix_reader: mpsc::UnboundedReceiver<(Response, Recipient)>,
) {
// TODO: wire SURBs in here once they're available
@@ -69,7 +68,7 @@ impl ServiceProvider {
}
async fn read_websocket_message(
websocket_reader: &mut SplitStream<WebSocketStream<TcpStream>>,
websocket_reader: &mut SplitStream<TSWebsocketStream>,
) -> Option<ReconstructedMessage> {
while let Some(msg) = websocket_reader.next().await {
let data = msg
@@ -270,7 +269,7 @@ impl ServiceProvider {
}
// Make the websocket connection so we can receive incoming Mixnet messages.
async fn connect_websocket(&self, uri: &str) -> WebSocketStream<TcpStream> {
async fn connect_websocket(&self, uri: &str) -> TSWebsocketStream {
let ws_stream = match websocket::Connection::new(uri).connect().await {
Ok(ws_stream) => {
info!("* connected to local websocket server at {}", uri);
service-providers/network-requester/src/websocket.rs
+5 -2
View File
@@ -2,8 +2,11 @@
// SPDX-License-Identifier: Apache-2.0
use tokio::net::TcpStream;
use tokio_tungstenite::connect_async;
use tokio_tungstenite::WebSocketStream;
use tokio_tungstenite::{connect_async, MaybeTlsStream};
#[allow(clippy::upper_case_acronyms)]
pub(crate) type TSWebsocketStream = WebSocketStream<MaybeTlsStream<TcpStream>>;
pub struct Connection {
uri: String,
@@ -16,7 +19,7 @@ impl Connection {
}
}
pub async fn connect(&self) -> Result<WebSocketStream<TcpStream>, WebsocketConnectionError> {
pub async fn connect(&self) -> Result<TSWebsocketStream, WebsocketConnectionError> {
match connect_async(&self.uri).await {
Ok((ws_stream, _)) => Ok(ws_stream),
Err(_e) => Err(WebsocketConnectionError::ConnectionNotEstablished),