OpenSource/teloxide
OpenSource/teloxide
1
0
Fork 0
mirror of https://github.com/teloxide/teloxide.git synced 2025-03-23 15:19:36 +01:00
Code Issues Projects Releases Packages Wiki Activity

Drop in-flight requests when polling is asked to stop

Browse source
This commit is contained in:
Maybe Waffle 2023-09-25 20:31:19 +04:00
parent d3c69eb0c9
commit 808a0cf4fb
1 changed files with 15 additions and 8 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -336,10 +336,23 @@ impl<B: Requester> Stream for PollingStream<'_, B> {
return Ready(None); return Ready(None);
} }
// If there are any buffered updates, return one
if let Some(upd) = this.buffer.next() {
return Ready(Some(Ok(upd)));
}
// Check if we should stop and if so — drop in flight request,
// we don't care about updates that happened *after* we started stopping
if !*this.stopping && this.polling.flag.is_stopped() {
*this.stopping = true;
log::trace!("dropping in-flight request");
this.in_flight.set(None);
}
// Poll in-flight future until completion // Poll in-flight future until completion
if let Some(in_flight) = this.in_flight.as_mut().as_pin_mut() { else if let Some(in_flight) = this.in_flight.as_mut().as_pin_mut() {
let res = ready!(in_flight.poll(cx)); let res = ready!(in_flight.poll(cx));
log::trace!("in-flight completed"); log::trace!("in-flight request completed");
this.in_flight.set(None); this.in_flight.set(None);
match res { match res {
@ -364,12 +377,6 @@ impl<B: Requester> Stream for PollingStream<'_, B> {
} }
} }
// If there are any buffered updates, return one
if let Some(upd) = this.buffer.next() {
return Ready(Some(Ok(upd)));
}
*this.stopping = this.polling.flag.is_stopped();
let (offset, limit, timeout) = match (this.stopping, this.drop_pending_updates) { let (offset, limit, timeout) = match (this.stopping, this.drop_pending_updates) {
// Normal `get_updates()` call // Normal `get_updates()` call
(false, false) => (*this.offset, this.polling.limit, *this.timeout), (false, false) => (*this.offset, this.polling.limit, *this.timeout),