astelm/telemt
1
0
Fork 0
mirror of https://github.com/telemt/telemt.git synced 2026-04-30 00:44:09 +03:00
Code Issues Packages Projects Releases Wiki Activity

Rustfmt

Browse Source
This commit is contained in:
Alexey
2026-04-25 14:35:35 +03:00
parent 2f2fe9d5d3
commit 37c916056a
2 changed files with 12 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
src/proxy/tests/client_security_tests.rs
View File

@@ -281,13 +281,8 @@ async fn user_connection_reservation_drop_enqueues_cleanup_synchronously() {
assert_eq!(ip_tracker.get_active_ip_count(&user).await, 1); assert_eq!(ip_tracker.get_active_ip_count(&user).await, 1);
assert_eq!(stats.get_user_curr_connects(&user), 1); assert_eq!(stats.get_user_curr_connects(&user), 1);
let reservation = UserConnectionReservation::new( let reservation =
stats.clone(), UserConnectionReservation::new(stats.clone(), ip_tracker.clone(), user.clone(), ip, true);
ip_tracker.clone(),
user.clone(),
ip,
true,
);
// Drop the reservation synchronously without any tokio::spawn/await yielding! // Drop the reservation synchronously without any tokio::spawn/await yielding!
drop(reservation); drop(reservation);

17
src/transport/middle_proxy/registry.rs
View File

@@ -8,8 +8,8 @@ use dashmap::DashMap;
use tokio::sync::mpsc::error::TrySendError; use tokio::sync::mpsc::error::TrySendError;
use tokio::sync::{Mutex, Semaphore, mpsc}; use tokio::sync::{Mutex, Semaphore, mpsc};
use super::{MeResponse, RouteBytePermit};
use super::codec::WriterCommand; use super::codec::WriterCommand;
use super::{MeResponse, RouteBytePermit};
const ROUTE_BACKPRESSURE_BASE_TIMEOUT_MS: u64 = 25; const ROUTE_BACKPRESSURE_BASE_TIMEOUT_MS: u64 = 25;
const ROUTE_BACKPRESSURE_HIGH_TIMEOUT_MS: u64 = 120; const ROUTE_BACKPRESSURE_HIGH_TIMEOUT_MS: u64 = 120;
@@ -129,7 +129,10 @@ impl ConnRegistry {
) )
} }
fn with_route_limits(route_channel_capacity: usize, route_byte_permits_per_conn: usize) -> Self { fn with_route_limits(
route_channel_capacity: usize,
route_byte_permits_per_conn: usize,
) -> Self {
let start = rand::random::<u64>() | 1; let start = rand::random::<u64>() | 1;
let route_channel_capacity = route_channel_capacity.max(1); let route_channel_capacity = route_channel_capacity.max(1);
Self { Self {
@@ -209,9 +212,10 @@ impl ConnRegistry {
let id = self.next_id.fetch_add(1, Ordering::Relaxed); let id = self.next_id.fetch_add(1, Ordering::Relaxed);
let (tx, rx) = mpsc::channel(self.route_channel_capacity); let (tx, rx) = mpsc::channel(self.route_channel_capacity);
self.routing.map.insert(id, tx); self.routing.map.insert(id, tx);
self.routing self.routing.byte_budget.insert(
.byte_budget id,
.insert(id, Arc::new(Semaphore::new(self.route_byte_permits_per_conn))); Arc::new(Semaphore::new(self.route_byte_permits_per_conn)),
);
(id, rx) (id, rx)
} }
@@ -287,8 +291,7 @@ impl ConnRegistry {
.map_err(|_| RouteResult::QueueFullHigh)?, .map_err(|_| RouteResult::QueueFullHigh)?,
Some(timeout_ms) => { Some(timeout_ms) => {
let acquire = semaphore.acquire_many_owned(permits); let acquire = semaphore.acquire_many_owned(permits);
match tokio::time::timeout(Duration::from_millis(timeout_ms.max(1)), acquire) match tokio::time::timeout(Duration::from_millis(timeout_ms.max(1)), acquire).await
.await
{ {
Ok(Ok(permit)) => permit, Ok(Ok(permit)) => permit,
Ok(Err(_)) => return Err(RouteResult::ChannelClosed), Ok(Err(_)) => return Err(RouteResult::ChannelClosed),