OpenSource/teloxide
OpenSource/teloxide
1
0
Fork 0
mirror of https://github.com/teloxide/teloxide.git synced 2025-01-10 20:12:25 +01:00
Code Issues Projects Releases Packages Wiki Activity

Throttle: Avoid repeatedly calling queue_full

Browse source
This commit is contained in:
Waffle 2021-06-29 21:05:55 +03:00
parent 6450d9227d
commit 1661f9f955
1 changed files with 9 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
src/adaptors/throttle.rs
View file

@ -241,6 +241,9 @@ const SECOND: Duration = Duration::from_secs(1);
// want to change this.
const DELAY: Duration = Duration::from_millis(250);
/// Minimal time beetween calls to queue_full function
const QUEUE_FULL_DELAY: Duration = Duration::from_secs(4);
#[derive(Debug)]
enum InfoMessage {
GetLimits { response: Sender<Limits> },
@ -321,6 +324,10 @@ async fn worker<F, Fut>(
let mut rx_is_closed = false;
let mut last_queue_full = Instant::now()
.checked_sub(QUEUE_FULL_DELAY)
.unwrap_or_else(Instant::now);
while !rx_is_closed || !queue.is_empty() {
// FIXME(waffle):
// 1. If the `queue` is empty, `read_from_rx` call down below will 'block'
@ -335,7 +342,8 @@ async fn worker<F, Fut>(
read_from_rx(&mut rx, &mut queue, &mut rx_is_closed).await;
//debug_assert_eq!(queue.capacity(), limits.messages_per_sec_overall as usize);
if queue.len() == queue.capacity() {
if queue.len() == queue.capacity() && last_queue_full.elapsed() > QUEUE_FULL_DELAY {
last_queue_full = Instant::now();
tokio::spawn(queue_full(queue.len()));
}