fix: correct usage of EventListener for scenarios that involve multip…

…le wakes

This resolves a potential bug that has not yet been observed with Stopper. It would only be encountered if the Event were notified before the atomic boolean was stored. Although the usage of atomics should make guard against this, the future logic still should take into consideration the possibility that the future was erroneously woken.

I misunderstood the contract for EventListener, which turns out to be that each EventListener is only good for one wake, after which it needs to be replaced with a new listener. I thought that repeatedly polling the listener would repeatedly wake it on Event notification.

refs: smol-rs/event-listener#124

src/future_stopper.rs

@@ -40,12 +40,12 @@ impl<F: Future> Future for FutureStopper<F> {
             }
 
             match Pin::new(&mut *this.event_listener).poll(cx) {
-                Poll::Ready(()) => continue,
+                Poll::Ready(()) => {
+                    *this.event_listener = this.stopper.0.event.listen();
+                    continue;
+                }
                 Poll::Pending => {
-                    return match this.future.poll(cx) {
-                        Poll::Ready(output) => Poll::Ready(Some(output)),
-                        Poll::Pending => Poll::Pending,
-                    }
+                    return this.future.poll(cx).map(Some);
                 }
             };
         }

src/stopped.rs

@@ -39,7 +39,10 @@ impl Future for Stopped {
             }
 
             match Pin::new(&mut *event_listener).poll(cx) {
-                Poll::Ready(()) => continue,
+                Poll::Ready(()) => {
+                    *event_listener = stopper.0.event.listen();
+                    continue;
+                }
                 Poll::Pending => return Poll::Pending,
             };
         }

src/stream_stopper.rs

@@ -66,7 +66,10 @@ impl<S: Stream> Stream for StreamStopper<S> {
             }
 
             match Pin::new(&mut *this.event_listener).poll(cx) {
-                Poll::Ready(()) => continue,
+                Poll::Ready(()) => {
+                    *this.event_listener = this.stopper.0.event.listen();
+                    continue;
+                }
                 Poll::Pending => return this.stream.poll_next(cx),
             };
         }