fix: resolve async read and send trait bounds issues
Co-authored-by: aider (openai/andrew/openrouter/deepseek/deepseek-chat-v3.1) <aider@aider.chat>
This commit is contained in:
@@ -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))?;
|
||||
|
||||
|
||||
@@ -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)
|
||||
};
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user