Replaces waker function with loop, reomves Unpin trait bounds

Author: assemblaj

src/stream/stream/partial_cmp.rs

@@ -18,11 +18,11 @@ pub struct PartialCmpFuture<L: Stream, R: Stream> {
     r_cache: Option<R::Item>, 
 }
 
-impl<L: Stream + Unpin, R: Stream + Unpin> Unpin for PartialCmpFuture<L, R> {}
-
 impl<L: Stream, R: Stream> PartialCmpFuture<L, R> {
     pin_utils::unsafe_pinned!(l: Fuse<L>);
     pin_utils::unsafe_pinned!(r: Fuse<R>);
+    pin_utils::unsafe_unpinned!(l_cache: Option<L::Item>);
+    pin_utils::unsafe_unpinned!(r_cache: Option<R::Item>);
 
     pub(super) fn new(l: L, r: R) -> Self {
         PartialCmpFuture {
@@ -36,59 +36,57 @@ impl<L: Stream, R: Stream> PartialCmpFuture<L, R> {
 
 impl<L: Stream, R: Stream> Future for PartialCmpFuture<L, R> 
 where 
-    L: Stream + Unpin + Sized,
-    R: Stream + Unpin + Sized,
+    L: Stream + Sized,
+    R: Stream + Sized,
     L::Item: PartialOrd<R::Item>
 {
     type Output = Option<Ordering>;
 
     fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {        
-        // Short circuit logic 
-        // Stream that completes earliest can be considered Less, etc
-        let l_complete = self.l.done && self.as_mut().l_cache.is_none();
-        let r_complete = self.r.done && self.as_mut().r_cache.is_none();
+        loop {
+            // Short circuit logic 
+            // Stream that completes earliest can be considered Less, etc
+            let l_complete = self.l.done && self.as_mut().l_cache.is_none();
+            let r_complete = self.r.done && self.as_mut().r_cache.is_none();
 
-        if l_complete && r_complete {
-            return Poll::Ready(Some(Ordering::Equal))
-        } else if l_complete {
-            return Poll::Ready(Some(Ordering::Less))
-        } else if r_complete {
-            return Poll::Ready(Some(Ordering::Greater))
-        }
+            if l_complete && r_complete {
+                return Poll::Ready(Some(Ordering::Equal))
+            } else if l_complete {
+                return Poll::Ready(Some(Ordering::Less))
+            } else if r_complete {
+                return Poll::Ready(Some(Ordering::Greater))
+            }
 
-        // Get next value if possible and necesary 
-        if !self.l.done && self.as_mut().l_cache.is_none() {
-            let l_next = futures_core::ready!(self.as_mut().l().poll_next(cx));
-            if let Some(item) = l_next {
-                self.as_mut().l_cache = Some(item);
+            // Get next value if possible and necesary 
+            if !self.l.done && self.as_mut().l_cache.is_none() {
+                let l_next = futures_core::ready!(self.as_mut().l().poll_next(cx));
+                if let Some(item) = l_next {
+                    *self.as_mut().l_cache() = Some(item);
+                }
             }
-        }
 
-        if !self.r.done && self.as_mut().r_cache.is_none() {
-            let r_next = futures_core::ready!(self.as_mut().r().poll_next(cx));
-            if let Some(item) = r_next {
-                self.as_mut().r_cache = Some(item);
+            if !self.r.done && self.as_mut().r_cache.is_none() {
+                let r_next = futures_core::ready!(self.as_mut().r().poll_next(cx));
+                if let Some(item) = r_next {
+                    *self.as_mut().r_cache() = Some(item);
+                }
             }
-        }
 
-        // Compare if both values are available.
-        if self.as_mut().l_cache.is_some() && self.as_mut().r_cache.is_some() {
-            let l_value = self.as_mut().l_cache.take().unwrap();
-            let r_value = self.as_mut().r_cache.take().unwrap(); 
-            let result = l_value.partial_cmp(&r_value);
+            // Compare if both values are available.
+            if self.as_mut().l_cache.is_some() && self.as_mut().r_cache.is_some() {
+                let l_value = self.as_mut().l_cache().take().unwrap();
+                let r_value = self.as_mut().r_cache().take().unwrap(); 
+                let result = l_value.partial_cmp(&r_value);
 
-            if let Some(Ordering::Equal) = result {
-                // Reset cache to prepare for next comparison 
-                self.as_mut().l_cache = None;  
-                self.as_mut().r_cache = None;
-            } else {
-                // Return non equal value 
-                return Poll::Ready(result);
-            }            
+                if let Some(Ordering::Equal) = result {
+                    // Reset cache to prepare for next comparison 
+                    *self.as_mut().l_cache() = None;  
+                    *self.as_mut().r_cache() = None;
+                } else {
+                    // Return non equal value 
+                    return Poll::Ready(result);
+                }            
+            }
         }
-
-        // wakes task to pull the next item from stream
-        cx.waker().wake_by_ref();
-        Poll::Pending
     }
-}
+}