Implement FusedStream for FuturesOrdered (#2205)

Author: phil-opp

futures-util/src/stream/futures_ordered.rs

@@ -1,7 +1,7 @@
 use crate::stream::{FuturesUnordered, StreamExt};
 use futures_core::future::Future;
 use futures_core::stream::Stream;
-use futures_core::task::{Context, Poll};
+use futures_core::{FusedStream, task::{Context, Poll}};
 use pin_project::pin_project;
 use core::cmp::Ordering;
 use core::fmt::{self, Debug};
@@ -203,6 +203,12 @@ impl<Fut: Future> FromIterator<Fut> for FuturesOrdered<Fut> {
     }
 }
 
+impl<Fut: Future> FusedStream for FuturesOrdered<Fut> {
+    fn is_terminated(&self) -> bool {
+        self.in_progress_queue.is_terminated() && self.queued_outputs.is_empty()
+    }
+}
+
 impl<Fut: Future> Extend<Fut> for FuturesOrdered<Fut> {
     fn extend<I>(&mut self, iter: I)
     where