Panic if All or Any are polled after completing due to a short-ci…

…rcuit

These futures should panic if they are polled after completing.
Currently they do so but only if they complete due to exhausting the
`Stream` that they pull data from. If they complete due to
short-circuiting, they are left in a state where `fut` and `accum` are
still `Some`. This means that if they are polled again, they end up
polling the inner `fut` again. That usually causes a panic but the error
message will likely reference the internal `Future`, not `All` / `Any`.

With this commit, `All` and `Any`'s internal state will be set such that
if they are polled again after completing, they will panic without
polling `fut`.

futures-util/src/stream/stream/all.rs

@@ -69,11 +69,12 @@ where
             if let Some(fut) = this.future.as_mut().as_pin_mut() {
                 // we're currently processing a future to produce a new accum value
                 let acc = this.accum.unwrap() && ready!(fut.poll(cx));
+                this.future.set(None);
                 if !acc {
+                    this.accum.take().unwrap();
                     break false;
                 } // early exit
                 *this.accum = Some(acc);
-                this.future.set(None);
             } else if this.accum.is_some() {
                 // we're waiting on a new item from the stream
                 match ready!(this.stream.as_mut().poll_next(cx)) {

futures-util/src/stream/stream/any.rs

@@ -69,11 +69,12 @@ where
             if let Some(fut) = this.future.as_mut().as_pin_mut() {
                 // we're currently processing a future to produce a new accum value
                 let acc = this.accum.unwrap() || ready!(fut.poll(cx));
+                this.future.set(None);
                 if acc {
+                    this.accum.take().unwrap();
                     break true;
                 } // early exit
                 *this.accum = Some(acc);
-                this.future.set(None);
             } else if this.accum.is_some() {
                 // we're waiting on a new item from the stream
                 match ready!(this.stream.as_mut().poll_next(cx)) {