Fix polling logic

Author: tyt2y3

sea-streamer-fuse/src/lib.rs

@@ -107,6 +107,7 @@ where
                 Poll::Ready(Some(Ok(mes))) => {
                     let key = mes.stream_key();
                     this.keys.entry(key).or_default().push_back(mes);
+                    // keep polling
                 }
                 Poll::Ready(Some(Err(err))) => {
                     *this.ended = true;
@@ -116,14 +117,18 @@ where
                     *this.ended = true;
                     break;
                 }
-                Poll::Pending => return Poll::Pending,
-            }
-            if !this.keys.values().any(|ms| ms.is_empty()) {
-                // if none of the streams are empty
-                break;
+                Poll::Pending => {
+                    // take a break
+                    break;
+                }
             }
         }
-        Poll::Ready(Self::next(this.keys).map(Ok))
+        if *this.ended || !this.keys.values().any(|ms| ms.is_empty()) {
+            // if none of the streams are empty
+            return Poll::Ready(Self::next(this.keys).map(Ok));
+        } else {
+            Poll::Pending
+        }
     }
 }