OpenSource/teloxide
OpenSource/teloxide
1
0
Fork 0
mirror of https://github.com/teloxide/teloxide.git synced 2025-03-14 11:44:04 +01:00
Code Issues Projects Releases Packages Wiki Activity

Throttle: log when queue is full

Browse source
This commit is contained in:
Waffle 2021-06-29 19:34:29 +03:00
parent 58cbf45bac
commit 07676a7156
1 changed files with 13 additions and 3 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

16
src/adaptors/throttle.rs
View file

@ -78,7 +78,9 @@ impl<B> Throttle<B> {
let (tx, rx) = mpsc::channel(limits.messages_per_sec_overall as usize);
let (info_tx, info_rx) = mpsc::channel(2);
let worker = worker(limits, rx, info_rx);
let worker = worker(limits, rx, info_rx, |pending| async move {
log::warn!("Throttle queue is full ({} pending requests)", pending);
});
let this = Self {
bot,
queue: tx,
@ -253,11 +255,15 @@ struct RequestsSentToChats {
// the request that it can be now executed, increase counts, add record to the
// history.
async fn worker(
async fn worker<F, Fut>(
mut limits: Limits,
mut rx: mpsc::Receiver<(ChatIdHash, RequestLock)>,
mut info_rx: mpsc::Receiver<InfoMessage>,
) {
queue_full: F,
) where
F: Fn(usize) -> Fut,
Fut: Future<Output = ()> + Send + 'static,
{
// FIXME(waffle): Make an research about data structures for this queue.
// Currently this is O(n) removing (n = number of elements
// stayed), amortized O(1) push (vec+vecrem).
@ -283,6 +289,10 @@ async fn worker(
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() {
tokio::spawn(queue_full(queue.len()));
}
// _Maybe_ we need to use `spawn_blocking` here, because there is
// decent amount of blocking work. However _for now_ I've decided not
// to use it here.