OpenSource/teloxide
OpenSource/teloxide
1
0
Fork 0
mirror of https://github.com/teloxide/teloxide.git synced 2024-12-22 14:35:36 +01:00
Code Issues Projects Releases Packages Wiki Activity

Slightly simplify Polling listener by moving update dropping

Browse source
This commit is contained in:
Maybe Waffle 2024-01-11 01:27:49 +01:00
parent d60552a878
commit ad303ef130
1 changed files with 9 additions and 17 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

26
crates/teloxide/src/update_listeners/polling.rs
View file

@ -13,6 +13,7 @@ use std::{
use futures::{ready, stream::Stream};
use crate::{
payloads::GetUpdatesSetters as _,
requests::{HasPayload, Request, Requester},
stop::{mk_stop_token, StopFlag, StopToken},
types::{AllowedUpdate, Update},
@ -286,9 +287,6 @@ pub struct PollingStream<'a, B: Requester> {
/// Parent structure
polling: &'a mut Polling<B>,
/// Whatever to drop pending updates or not.
drop_pending_updates: bool,
/// Timeout parameter for normal `get_updates()` calls.
timeout: Option<u32>,
/// Allowed updates parameter for the first `get_updates()` call.
@ -324,19 +322,19 @@ impl<B: Requester + Send + 'static> UpdateListener for Polling<B> {
Box::pin(async {
let timeout = self.timeout.map(|t| t.as_secs().try_into().expect("timeout is too big"));
let allowed_updates = self.allowed_updates.clone();
let drop_pending_updates = self.drop_pending_updates;
// FIXME: document that `listen` is a destructive operation, actually,
// and you need to call `stop_token` *again* after it
self.reinit_stop_flag_if_needed();
// FIXME: do update dropping *here*
if self.drop_pending_updates {
self.bot.get_updates().offset(-1).limit(1).timeout(0).await?;
}
// Unwrap: just called reinit
let flag = self.flag.take().unwrap();
let stream = PollingStream {
polling: self,
drop_pending_updates,
timeout,
allowed_updates,
offset: 0,
@ -409,29 +407,23 @@ impl<B: Requester> Stream for PollingStream<'_, B> {
*this.offset = upd.id.as_offset();
}
match *this.drop_pending_updates {
false => *this.buffer = updates.into_iter(),
true => *this.drop_pending_updates = false,
}
*this.buffer = updates.into_iter();
}
Err(err) => return Ready(Some(Err(err))),
}
}
let (offset, limit, timeout) = match (this.stopping, this.drop_pending_updates) {
let (offset, limit, timeout) = match this.stopping {
// Normal `get_updates()` call
(false, false) => (*this.offset, this.polling.limit, *this.timeout),
// Graceful shutdown `get_updates()` call (shutdown takes priority over dropping pending
// updates)
false => (*this.offset, this.polling.limit, *this.timeout),
// Graceful shutdown `get_updates()` call
//
// When stopping we set `timeout = 0` and `limit = 1` so that `get_updates()`
// set last seen update (offset) and return immediately
(true, _) => {
true => {
log::trace!("graceful shutdown `get_updates` call");
(*this.offset, Some(1), Some(0))
}
// Drop pending updates
(_, true) => (-1, Some(1), Some(0)),
};
let req = this