f - Implement async chunked transfer encoding using chunked_transfer::Decoder

Buffering asynchronously first using read_to_end won't work since the
server may not close the stream. Instead read the chunk header and use
the decoder only to get the chunk size.

Author: jkczyz

lightning-block-sync/Cargo.toml

@@ -17,7 +17,7 @@ bitcoin = "0.24"
 lightning = { version = "0.0.12", path = "../lightning" }
 tokio = { version = "1.0", features = [ "io-util", "net" ], optional = true }
 serde_json = { version = "1.0", optional = true }
-chunked_transfer = { version = "1.3.0", optional = true }
+chunked_transfer = { version = "1.4.0", optional = true }
 futures = { version = "0.3.8" }
 
 [dev-dependencies]

lightning-block-sync/src/http.rs

@@ -274,17 +274,46 @@ impl HttpClient {
 					Err(std::io::Error::new(
 							std::io::ErrorKind::InvalidInput, "unsupported transfer coding"))
 				} else {
-					#[cfg(feature = "tokio")]
-					let reader = {
-						// Read the stream into a buffer since chunked_transfer isn't async.
-						let mut buffer = Vec::new();
-						reader.read_to_end(&mut buffer).await?;
-						std::io::Cursor::new(buffer)
-					};
-					let mut decoder = chunked_transfer::Decoder::new(reader);
 					let mut content = Vec::new();
-					decoder.read_to_end(&mut content)?;
-					Ok(content)
+					#[cfg(feature = "tokio")]
+					{
+						// Since chunked_transfer doesn't have an async interface, only use it to
+						// determine the size of each chunk to read.
+						//
+						// TODO: Replace with an async interface when available.
+						// https://github.com/frewsxcv/rust-chunked-transfer/issues/7
+						loop {
+							// Read the chunk header which contains the chunk size.
+							let mut chunk_header = String::new();
+							reader.read_line(&mut chunk_header).await?;
+							if chunk_header == "0\r\n" {
+								// Read the terminator chunk since the decoder consumes the CRLF
+								// immediately when this chunk is encountered.
+								reader.read_line(&mut chunk_header).await?;
+							}
+
+							// Decode the chunk header to obtain the chunk size.
+							let mut buffer = Vec::new();
+							let mut decoder = chunked_transfer::Decoder::new(chunk_header.as_bytes());
+							decoder.read_to_end(&mut buffer)?;
+
+							// Read the chunk body.
+							let chunk_size = match decoder.remaining_chunks_size() {
+								None => break,
+								Some(chunk_size) => chunk_size,
+							};
+							let mut chunk_body = vec![0; chunk_size + "\r\n".len()];
+							reader.read_exact(&mut chunk_body[..]).await?;
+							content.extend_from_slice(&chunk_body[..chunk_size]);
+						}
+						Ok(content)
+					}
+					#[cfg(not(feature = "tokio"))]
+					{
+						let mut decoder = chunked_transfer::Decoder::new(reader);
+						decoder.read_to_end(&mut content)?;
+						Ok(content)
+					}
 				}
 			},
 		}