Relay chunked transfer encoding from reader to writer. Copies bytes verbatim (preserving chunk framing) while parsing the stream boundaries so we can stop exactly at the end of the current message body. Handles chunk extensions and trailers per RFC 7230. `already_forwarded` are overflow bytes that were already written to the writer during header parsing. They are seeded into the parser buffer so
(
reader: &mut R,
writer: &mut W,
already_forwarded: &[u8],
generation_guard: Option<&PolicyGenerationGuard>,
)
| 1773 | /// writer during header parsing. They are seeded into the parser buffer so |
| 1774 | /// termination can still be detected when boundaries span reads. |
| 1775 | async fn relay_chunked<R, W>( |
| 1776 | reader: &mut R, |
| 1777 | writer: &mut W, |
| 1778 | already_forwarded: &[u8], |
| 1779 | generation_guard: Option<&PolicyGenerationGuard>, |
| 1780 | ) -> Result<()> |
| 1781 | where |
| 1782 | R: AsyncRead + Unpin, |
| 1783 | W: AsyncWrite + Unpin, |
| 1784 | { |
| 1785 | let started_at = std::time::Instant::now(); |
| 1786 | let mut read_buf = [0u8; RELAY_BUF_SIZE]; |
| 1787 | let mut parse_buf = Vec::from(already_forwarded); |
| 1788 | let mut pos = 0usize; |
| 1789 | let mut chunk_count = 0usize; |
| 1790 | let mut chunk_payload_bytes = 0usize; |
| 1791 | |
| 1792 | // Parse chunk-size lines + chunk payloads until final 0-size chunk, then |
| 1793 | // parse trailers until the terminating empty trailer line. |
| 1794 | loop { |
| 1795 | // Parse one chunk size line: "<hex>[;extensions]\r\n" |
| 1796 | let size_line_end = loop { |
| 1797 | if let Some(end) = find_crlf(&parse_buf, pos) { |
| 1798 | break end; |
| 1799 | } |
| 1800 | let n = reader.read(&mut read_buf).await.into_diagnostic()?; |
| 1801 | if n == 0 { |
| 1802 | return Err(miette!("Chunked body ended before chunk-size line")); |
| 1803 | } |
| 1804 | if let Some(guard) = generation_guard { |
| 1805 | guard.ensure_current()?; |
| 1806 | } |
| 1807 | writer.write_all(&read_buf[..n]).await.into_diagnostic()?; |
| 1808 | parse_buf.extend_from_slice(&read_buf[..n]); |
| 1809 | }; |
| 1810 | |
| 1811 | let size_line = std::str::from_utf8(&parse_buf[pos..size_line_end]) |
| 1812 | .into_diagnostic() |
| 1813 | .map_err(|_| miette!("Invalid UTF-8 in chunk-size line"))?; |
| 1814 | let size_token = size_line |
| 1815 | .split(';') |
| 1816 | .next() |
| 1817 | .map(str::trim) |
| 1818 | .unwrap_or_default(); |
| 1819 | let chunk_size = usize::from_str_radix(size_token, 16) |
| 1820 | .into_diagnostic() |
| 1821 | .map_err(|_| miette!("Invalid chunk size token: {size_token:?}"))?; |
| 1822 | pos = size_line_end + 2; |
| 1823 | |
| 1824 | if chunk_size == 0 { |
| 1825 | // Parse trailers (if any). Terminates on empty trailer line. |
| 1826 | let mut trailer_count = 0usize; |
| 1827 | loop { |
| 1828 | let trailer_end = loop { |
| 1829 | if let Some(end) = find_crlf(&parse_buf, pos) { |
| 1830 | break end; |
| 1831 | } |
| 1832 | let n = reader.read(&mut read_buf).await.into_diagnostic()?; |