| 27 | } |
| 28 | } |
| 29 | fn poll_ready(&self, cx: &mut Context<'_>) -> Poll<()> { |
| 30 | // Lazily initialize the AsyncPollable |
| 31 | let subscription = self |
| 32 | .subscription |
| 33 | .get_or_init(|| AsyncPollable::new(self.stream.subscribe())); |
| 34 | // Lazily initialize the WaitFor. Clear it after it becomes ready. |
| 35 | let mut wait_for_slot = self.wait_for.lock().unwrap(); |
| 36 | let wait_for = wait_for_slot.get_or_insert_with(|| Box::pin(subscription.wait_for())); |
| 37 | match wait_for.as_mut().poll(cx) { |
| 38 | Poll::Pending => Poll::Pending, |
| 39 | Poll::Ready(()) => { |
| 40 | let _ = wait_for_slot.take(); |
| 41 | Poll::Ready(()) |
| 42 | } |
| 43 | } |
| 44 | } |
| 45 | /// Await for read readiness. |
| 46 | async fn ready(&self) { |
| 47 | poll_fn(|cx| self.poll_ready(cx)).await |