asp/keep
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix: replace streaming with synchronous content retrieval

Browse Source 
Co-authored-by: aider (openai/andrew/openrouter/qwen/qwen3-coder) <aider@aider.chat>
This commit is contained in:
Andrew Phillips
2025-08-25 19:06:48 -03:00
parent b039bc4b33
commit 694575ad36
1 changed files with 10 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files

42
src/services/async_item_service.rs
View File

@@ -56,23 +56,20 @@ impl AsyncItemService {
&self, &self,
item_id: i64, item_id: i64,
allow_binary: bool, allow_binary: bool,
offset: u64, _offset: u64,
length: u64, _length: u64,
) -> Result<(std::pin::Pin<Box<dyn tokio_stream::Stream<Item = Result<tokio_util::bytes::Bytes, std::io::Error>> + Send>>, String), CoreError> { ) -> Result<(std::pin::Pin<Box<dyn tokio_stream::Stream<Item = Result<tokio_util::bytes::Bytes, std::io::Error>> + Send>>, String), CoreError> {
// First get the item to determine compression type // Get the item content synchronously
let item_with_meta = self.get_item(item_id).await?; let item_with_content = self.get_item_content(item_id).await?;
let metadata = item_with_meta.meta_as_map();
// Check if content is binary when allow_binary is false // Check if content is binary when allow_binary is false
let metadata = item_with_content.item_with_meta.meta_as_map();
if !allow_binary { if !allow_binary {
// We need to check the actual content, but we don't want to load the whole file
// Just check the binary metadata if available
let is_content_binary = if let Some(binary_val) = metadata.get("binary") { let is_content_binary = if let Some(binary_val) = metadata.get("binary") {
binary_val == "true" binary_val == "true"
} else { } else {
// We can't efficiently check binary content without loading it // Fallback to checking the actual content
// This is a limitation of the streaming approach crate::common::is_binary::is_binary(&item_with_content.content)
false
}; };
if is_content_binary { if is_content_binary {
@@ -85,28 +82,9 @@ impl AsyncItemService {
.map(|s| s.to_string()) .map(|s| s.to_string())
.unwrap_or_else(|| "application/octet-stream".to_string()); .unwrap_or_else(|| "application/octet-stream".to_string());
// Open the file for streaming // Convert content to stream
let file_path = self.data_dir.join(item_id.to_string()); let content = item_with_content.content;
let stream = Box::pin(tokio_stream::iter(vec![Ok(tokio_util::bytes::Bytes::from(content))]));
// Get the compression engine to decompress while streaming
let compression_type = crate::compression_engine::CompressionType::from_str(&item_with_meta.item.compression)
.map_err(|e| CoreError::Compression(e.to_string()))?;
let engine = crate::compression_engine::get_compression_engine(compression_type)
.map_err(|e| CoreError::Other(anyhow::anyhow!(e.to_string())))?;
// Open the compressed file
let reader = engine.open(file_path)
.map_err(|e| CoreError::Other(anyhow::anyhow!("Failed to open item file: {}", e)))?;
// Create a reader stream directly from the reader
let stream: std::pin::Pin<Box<dyn tokio_stream::Stream<Item = Result<tokio_util::bytes::Bytes, std::io::Error>> + Send>> =
if length > 0 {
// Limit the stream to the specified length
Box::pin(ReaderStream::new(reader.take(length)))
} else {
// Stream the entire decompressed file
Box::pin(ReaderStream::new(reader))
};
Ok((stream, mime_type)) Ok((stream, mime_type))
} }