From 2477a0f945a367731928d84d02b580367d48f106 Mon Sep 17 00:00:00 2001
From: Maybe Waffle <waffle.lapkin@gmail.com>
Date: Sun, 10 Apr 2022 19:04:30 +0400
Subject: [PATCH] Fix infinite loop in `read_from_rx`

---
 CHANGELOG.md                    |  6 ++++++
 src/adaptors/throttle/worker.rs | 19 ++++++++++++++++++-
 2 files changed, 24 insertions(+), 1 deletion(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 5afb3b4e..daf670f4 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -25,6 +25,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   - Retry requests that previously returned `RetryAfter(_)` error
 - `RequestError::RetryAfter` now has a `Duration` field instead of `i32`
 
+### Fixed
+
+- Fix never ending loop that caused programs that used `Throttling` to never stop, see issue [#535][issue535]
+
+[issue535]: https://github.com/teloxide/teloxide/issues/535
+
 ### Added
 
 - `UserId::{url, is_anonymous, is_channel, is_telegram}` convenience functions ([#197][pr197]) 
diff --git a/src/adaptors/throttle/worker.rs b/src/adaptors/throttle/worker.rs
index 85bca5bd..4aa57099 100644
--- a/src/adaptors/throttle/worker.rs
+++ b/src/adaptors/throttle/worker.rs
@@ -365,9 +365,26 @@ async fn read_from_rx<T>(rx: &mut mpsc::Receiver<T>, queue: &mut Vec<T>, rx_is_c
     while queue.len() < queue.capacity() {
         match rx.try_recv() {
             Ok(req) => queue.push(req),
-            Err(TryRecvError::Disconnected) => *rx_is_closed = true,
+            Err(TryRecvError::Disconnected) => {
+                *rx_is_closed = true;
+                break;
+            }
             // There are no items in queue.
             Err(TryRecvError::Empty) => break,
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    #[tokio::test]
+    async fn issue_535() {
+        let (tx, mut rx) = tokio::sync::mpsc::channel(1);
+
+        // Close channel
+        drop(tx);
+
+        // Previously this caused an infinite loop
+        super::read_from_rx::<()>(&mut rx, &mut Vec::new(), &mut false).await;
+    }
+}