diff --git a/src/modes/get.rs b/src/modes/get.rs index ba897a5..7722b0f 100644 --- a/src/modes/get.rs +++ b/src/modes/get.rs @@ -31,7 +31,7 @@ pub fn mode_get( } } - let item_service = ItemService::new(data_path); + let item_service = ItemService::new(data_path.clone()); let item_with_meta = item_service.find_item(conn, ids, tags, &meta) .map_err(|e| anyhow!("Unable to find matching item in database: {}", e))?; diff --git a/src/services/async_item_service.rs b/src/services/async_item_service.rs index c62e29e..7fb593f 100644 --- a/src/services/async_item_service.rs +++ b/src/services/async_item_service.rs @@ -208,12 +208,52 @@ impl AsyncItemService { } } - // Create a stream that reads the content in chunks - let stream = ReaderStream::new(reader); + // Convert the reader into an async stream manually + // Since ReaderStream requires AsyncRead, we'll create our own implementation + use tokio_stream::StreamExt; + use tokio_util::bytes::Bytes; + use std::io::Error; + + // Create a channel to stream data between the blocking thread and async runtime + let (tx, rx) = tokio::sync::mpsc::channel(1); + + // Spawn a blocking task to read from the reader and send chunks + tokio::task::spawn_blocking(move || { + let mut buffer = [0; 8192]; + loop { + match reader.read(&mut buffer) { + Ok(0) => break, // EOF + Ok(n) => { + let chunk = Bytes::copy_from_slice(&buffer[..n]); + // Block on sending to the channel + if tx.blocking_send(Ok(chunk)).is_err() { + break; // Receiver dropped + } + } + Err(e) => { + let _ = tx.blocking_send(Err(e)); + break; + } + } + } + }); + + // Convert the receiver into a stream + let stream = tokio_stream::wrappers::ReceiverStream::new(rx); // If length is specified, we need to limit the stream let limited_stream = if length > 0 { - Box::pin(stream.take(length as usize)) + // We need to track how many bytes we've sent + let mut bytes_sent = 0; + Box::pin(stream.take_while(move |result| { + if bytes_sent >= length { + return std::future::ready(false); + } + if let Ok(chunk) = result { + bytes_sent += chunk.len() as u64; + } + std::future::ready(true) + })) } else { Box::pin(stream) }; diff --git a/src/services/compression_service.rs b/src/services/compression_service.rs index 3a31b3e..ddfa328 100644 --- a/src/services/compression_service.rs +++ b/src/services/compression_service.rs @@ -37,7 +37,7 @@ impl CompressionService { let reader = engine.open(item_path.clone()) .map_err(|e| CoreError::Other(anyhow!("Failed to open item file {:?}: {}", item_path, e)))?; - Ok(Box::new(reader)) + Ok(reader) } }