From 8de9032846b7a46553e9aaf1718903e53281c430 Mon Sep 17 00:00:00 2001
From: BiagioFesta <15035284+BiagioFesta@users.noreply.github.com>
Date: Wed, 22 Jan 2025 13:00:45 +0100
Subject: [PATCH] async/consumer: fix WaitOccupiedFuture fused impl

---
 async/src/traits/consumer.rs | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/async/src/traits/consumer.rs b/async/src/traits/consumer.rs
index 3680fcf..32591ec 100644
--- a/async/src/traits/consumer.rs
+++ b/async/src/traits/consumer.rs
@@ -248,12 +248,13 @@ impl<A: AsyncConsumer> FusedFuture for WaitOccupiedFuture<'_, A> {
 impl<A: AsyncConsumer> Future for WaitOccupiedFuture<'_, A> {
     type Output = ();
 
-    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
         let mut waker_registered = false;
         loop {
             assert!(!self.done);
             let closed = self.owner.is_closed();
             if self.count <= self.owner.occupied_len() || closed {
+                self.done = true;
                 break Poll::Ready(());
             }
             if waker_registered {