forked from mirror/async-std
		
	fix stream min_by_key mistake
This commit is contained in:
		
							parent
							
								
									667bbc1019
								
							
						
					
					
						commit
						ca71ad073b
					
				
					 1 changed files with 16 additions and 8 deletions
				
			
		|  | @ -1,6 +1,6 @@ | ||||||
| use std::cmp::Ordering; | use std::cmp::Ordering; | ||||||
| use std::pin::Pin; |  | ||||||
| use std::future::Future; | use std::future::Future; | ||||||
|  | use std::pin::Pin; | ||||||
| 
 | 
 | ||||||
| use pin_project_lite::pin_project; | use pin_project_lite::pin_project; | ||||||
| 
 | 
 | ||||||
|  | @ -13,7 +13,7 @@ pin_project! { | ||||||
|     pub struct MinByKeyFuture<S, T, K> { |     pub struct MinByKeyFuture<S, T, K> { | ||||||
|         #[pin] |         #[pin] | ||||||
|         stream: S, |         stream: S, | ||||||
|         min: Option<T>, |         min: Option<(T, T)>, | ||||||
|         key_by: K, |         key_by: K, | ||||||
|     } |     } | ||||||
| } | } | ||||||
|  | @ -37,24 +37,32 @@ where | ||||||
|     type Output = Option<S::Item>; |     type Output = Option<S::Item>; | ||||||
| 
 | 
 | ||||||
|     fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { |     fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { | ||||||
|  |         fn key<B, T>(mut f: impl FnMut(&T) -> B) -> impl FnMut(T) -> (B, T) { | ||||||
|  |             move |x| (f(&x), x) | ||||||
|  |         } | ||||||
|  | 
 | ||||||
|         let this = self.project(); |         let this = self.project(); | ||||||
|         let next = futures_core::ready!(this.stream.poll_next(cx)); |         let next = futures_core::ready!(this.stream.poll_next(cx)); | ||||||
| 
 | 
 | ||||||
|         match next { |         match next { | ||||||
|             Some(new) => { |             Some(new) => { | ||||||
|                 let new = (this.key_by)(&new); |                 let (key, value) = key(this.key_by)(new); | ||||||
|                 cx.waker().wake_by_ref(); |                 cx.waker().wake_by_ref(); | ||||||
|                 match this.min.take() { |  | ||||||
|                     None => *this.min = Some(new), |  | ||||||
| 
 | 
 | ||||||
|                     Some(old) => match new.cmp(&old) { |                 match this.min.take() { | ||||||
|                         Ordering::Less => *this.min = Some(new), |                     None => *this.min = Some((key, value)), | ||||||
|  | 
 | ||||||
|  |                     Some(old) => match key.cmp(&old.0) { | ||||||
|  |                         Ordering::Less => *this.min = Some((key, value)), | ||||||
|                         _ => *this.min = Some(old), |                         _ => *this.min = Some(old), | ||||||
|                     }, |                     }, | ||||||
|                 } |                 } | ||||||
|                 Poll::Pending |                 Poll::Pending | ||||||
|             } |             } | ||||||
|             None => Poll::Ready(this.min.take()), |             None => Poll::Ready(match this.min.take() { | ||||||
|  |                 None => None, | ||||||
|  |                 Some(max) => Some(max.1), | ||||||
|  |             }), | ||||||
|         } |         } | ||||||
|     } |     } | ||||||
| } | } | ||||||
|  |  | ||||||
		Loading…
	
		Reference in a new issue