Fix infinite loop in read_from_rx

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]) 

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;
+    }
+}