| 33 | W: AsyncWrite + Unpin, |
| 34 | { |
| 35 | async fn step(&mut self) -> Result<bool> { |
| 36 | match self.next_step { |
| 37 | NextStep::Read => { |
| 38 | let n = timeout(self.cfg.timeouts.idle, self.reader.read_buf(&mut self.buf)) |
| 39 | .await |
| 40 | .ok() |
| 41 | .context("idle timeout")? |
| 42 | .context("read error")?; |
| 43 | trace!(direction = %self.dir, "read: {n} bytes"); |
| 44 | if n == 0 { |
| 45 | self.next_step = NextStep::Shutdown; |
| 46 | } else { |
| 47 | self.next_step = NextStep::Write; |
| 48 | } |
| 49 | Ok(false) |
| 50 | } |
| 51 | NextStep::Write => { |
| 52 | timeout( |
| 53 | self.cfg.timeouts.write, |
| 54 | self.writer.write_buf(&mut self.buf), |
| 55 | ) |
| 56 | .await |
| 57 | .ok() |
| 58 | .context("write timeout")? |
| 59 | .context("write error")?; |
| 60 | if self.buf.is_empty() { |
| 61 | self.next_step = NextStep::Flush; |
| 62 | } |
| 63 | Ok(false) |
| 64 | } |
| 65 | NextStep::Flush => { |
| 66 | timeout(self.cfg.timeouts.write, self.writer.flush()) |
| 67 | .await |
| 68 | .ok() |
| 69 | .context("flush timeout")? |
| 70 | .context("flush error")?; |
| 71 | self.next_step = NextStep::Read; |
| 72 | Ok(false) |
| 73 | } |
| 74 | NextStep::Shutdown => { |
| 75 | timeout(self.cfg.timeouts.shutdown, self.writer.shutdown()) |
| 76 | .await |
| 77 | .ok() |
| 78 | .context("shutdown timeout")? |
| 79 | .context("shutdown error")?; |
| 80 | self.next_step = NextStep::Done; |
| 81 | Ok(true) |
| 82 | } |
| 83 | NextStep::Done => Ok(true), |
| 84 | } |
| 85 | } |
| 86 | } |
| 87 | |
| 88 | enum Rest<A, B> { |