| 392 | type Item = Result<T>; |
| 393 | |
| 394 | fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> { |
| 395 | let self_project = self.project(); |
| 396 | match self_project.batch_rx.poll_recv(cx) { |
| 397 | Poll::Ready(Some(Ok((item, _semaphore_permit)))) => { |
| 398 | self_project.memory_reservation.shrink(item.size()); |
| 399 | Poll::Ready(Some(Ok(item))) |
| 400 | } |
| 401 | Poll::Ready(Some(Err(err))) => Poll::Ready(Some(Err(err))), |
| 402 | Poll::Ready(None) => Poll::Ready(None), |
| 403 | Poll::Pending => Poll::Pending, |
| 404 | } |
| 405 | } |
| 406 | |
| 407 | fn size_hint(&self) -> (usize, Option<usize>) { |
| 408 | if self.batch_rx.is_closed() { |