diff options
| author | Alex Auvolat <alex@adnab.me> | 2024-02-23 16:50:34 +0100 |
|---|---|---|
| committer | Alex Auvolat <alex@adnab.me> | 2024-02-23 16:50:34 +0100 |
| commit | 916c67ccf4c9d31c14088f2d775e15c64750458f (patch) | |
| tree | 63226f78b128f33b63d3b223283047372bcc31f2 /src/api/s3 | |
| parent | 81cebdd12415381f67747e96591e83b1a4a8cc0b (diff) | |
| parent | 61758ce0f91b930542bd2ee3c72735000cc12e75 (diff) | |
| download | garage-916c67ccf4c9d31c14088f2d775e15c64750458f.tar.gz garage-916c67ccf4c9d31c14088f2d775e15c64750458f.zip | |
Merge branch 'main' into next-0.10
Diffstat (limited to 'src/api/s3')
| -rw-r--r-- | src/api/s3/get.rs | 12 |
1 files changed, 6 insertions, 6 deletions
diff --git a/src/api/s3/get.rs b/src/api/s3/get.rs index 53f0a345..efb8d4ab 100644 --- a/src/api/s3/get.rs +++ b/src/api/s3/get.rs @@ -13,7 +13,7 @@ use http::header::{ use hyper::{body::Body, Request, Response, StatusCode}; use tokio::sync::mpsc; -use garage_block::manager::BlockStream; +use garage_net::stream::ByteStream; use garage_rpc::rpc_helper::OrderTag; use garage_table::EmptyKey; use garage_util::data::*; @@ -286,7 +286,7 @@ pub async fn handle_get( Ok(resp_builder.body(bytes_body(bytes.to_vec().into()))?) } ObjectVersionData::FirstBlock(_, first_block_hash) => { - let (tx, rx) = mpsc::channel::<BlockStream>(2); + let (tx, rx) = mpsc::channel::<ByteStream>(2); let order_stream = OrderTag::stream(); let first_block_hash = *first_block_hash; @@ -494,7 +494,7 @@ fn body_from_blocks_range( } let order_stream = OrderTag::stream(); - let (tx, rx) = mpsc::channel::<BlockStream>(2); + let (tx, rx) = mpsc::channel::<ByteStream>(2); tokio::spawn(async move { match async { @@ -542,7 +542,7 @@ fn body_from_blocks_range( }) .filter_map(futures::future::ready); - let block_stream: BlockStream = Box::pin(block_stream); + let block_stream: ByteStream = Box::pin(block_stream); tx.send(Box::pin(block_stream)) .await .ok_or_message("channel closed")?; @@ -562,7 +562,7 @@ fn body_from_blocks_range( response_body_from_block_stream(rx) } -fn response_body_from_block_stream(rx: mpsc::Receiver<BlockStream>) -> ResBody { +fn response_body_from_block_stream(rx: mpsc::Receiver<ByteStream>) -> ResBody { let body_stream = tokio_stream::wrappers::ReceiverStream::new(rx) .flatten() .map(|x| { @@ -572,7 +572,7 @@ fn response_body_from_block_stream(rx: mpsc::Receiver<BlockStream>) -> ResBody { ResBody::new(http_body_util::StreamBody::new(body_stream)) } -fn error_stream_item<E: std::fmt::Display>(e: E) -> BlockStream { +fn error_stream_item<E: std::fmt::Display>(e: E) -> ByteStream { let err = std::io::Error::new( std::io::ErrorKind::Other, format!("Error while getting object data: {}", e), |