Skip to content

Commit

Permalink
Chunked ReadFile from file manager
Browse files Browse the repository at this point in the history
  • Loading branch information
@izgzhen
izgzhen committed Jul 25, 2016
1 parent 2de3b11 commit 9c8ebd3
Show file tree
Hide file tree
Showing 7 changed files with 164 additions and 58 deletions.
40 changes: 34 additions & 6 deletions components/net/blob_loader.rs
View file
Expand Up @@ -10,7 +10,7 @@ use mime::{Mime, Attr};
use mime_classifier::MimeClassifier;
use net_traits::ProgressMsg::{Payload, Done};
use net_traits::blob_url_store::parse_blob_url;
use net_traits::filemanager_thread::{FileManagerThreadMsg, SelectedFileId};
use net_traits::filemanager_thread::{FileManagerThreadMsg, SelectedFileId, ReadFileProgress};
use net_traits::response::HttpsState;
use net_traits::{LoadConsumer, LoadData, Metadata, NetworkError};
use resource_thread::CancellationListener;
Expand All @@ -27,25 +27,28 @@ pub fn factory(filemanager_chan: IpcSender<FileManagerThreadMsg>)
LoadConsumer,
Arc<MimeClassifier>,
CancellationListener) + Send> {
box move |load_data: LoadData, start_chan, classifier, _cancel_listener| {
box move |load_data: LoadData, start_chan, classifier, cancel_listener| {
spawn_named(format!("blob loader for {}", load_data.url), move || {
load_blob(load_data, start_chan, classifier, filemanager_chan);
load_blob(load_data, start_chan, classifier, filemanager_chan, cancel_listener);
})
}
}

fn load_blob(load_data: LoadData, start_chan: LoadConsumer,
classifier: Arc<MimeClassifier>,
filemanager_chan: IpcSender<FileManagerThreadMsg>) {
filemanager_chan: IpcSender<FileManagerThreadMsg>,
// XXX(izgzhen): we should utilize cancel_listener, filed in #12589
_cancel_listener: CancellationListener) {
let (chan, recv) = ipc::channel().unwrap();
if let Ok((id, origin, _fragment)) = parse_blob_url(&load_data.url.clone()) {
let id = SelectedFileId(id.simple().to_string());
let check_url_validity = true;
let msg = FileManagerThreadMsg::ReadFile(chan, id, check_url_validity, origin);
let _ = filemanager_chan.send(msg);

// Receive first chunk
match recv.recv().unwrap() {
Ok(blob_buf) => {
Ok(ReadFileProgress::Meta(blob_buf)) => {
let content_type: Mime = blob_buf.type_string.parse().unwrap_or(mime!(Text / Plain));
let charset = content_type.get_param(Attr::Charset);

Expand Down Expand Up @@ -80,9 +83,34 @@ fn load_blob(load_data: LoadData, start_chan: LoadConsumer,
start_sending_sniffed_opt(start_chan, metadata, classifier,
&blob_buf.bytes, load_data.context.clone()) {
let _ = chan.send(Payload(blob_buf.bytes));
let _ = chan.send(Done(Ok(())));

loop {
match recv.recv().unwrap() {
Ok(ReadFileProgress::Partial(bytes)) => {
let _ = chan.send(Payload(bytes));
}
Ok(ReadFileProgress::EOF) => {
let _ = chan.send(Done(Ok(())));
return;
}
Ok(_) => {
let err = NetworkError::Internal("Invalid filemanager reply".to_string());
let _ = chan.send(Done(Err(err)));
return;
}
Err(e) => {
let err = NetworkError::Internal(format!("{:?}", e));
let _ = chan.send(Done(Err(err)));
return;
}
}
}
}
}
Ok(_) => {
let err = NetworkError::Internal("Invalid filemanager reply".to_string());
send_error(load_data.url, err, start_chan);
}
Err(e) => {
let err = NetworkError::Internal(format!("{:?}", e));
send_error(load_data.url, err, start_chan);
Expand Down
106 changes: 74 additions & 32 deletions components/net/filemanager_thread.rs
View file
Expand Up @@ -6,7 +6,8 @@ use ipc_channel::ipc::{self, IpcReceiver, IpcSender};
use mime_guess::guess_mime_type_opt;
use net_traits::blob_url_store::{BlobBuf, BlobURLStoreError};
use net_traits::filemanager_thread::{FileManagerThreadMsg, FileManagerResult, FilterPattern, FileOrigin};
use net_traits::filemanager_thread::{SelectedFile, RelativePos, FileManagerThreadError, SelectedFileId};
use net_traits::filemanager_thread::{SelectedFile, RelativePos, FileManagerThreadError};
use net_traits::filemanager_thread::{SelectedFileId, ReadFileProgress};
use std::collections::HashMap;
use std::fs::File;
use std::io::{Read, Seek, SeekFrom};
Expand Down Expand Up @@ -156,11 +157,8 @@ impl<UI: 'static + UIProvider> FileManager<UI> {
}
FileManagerThreadMsg::ReadFile(sender, id, check_url_validity, origin) => {
spawn_named("read file".to_owned(), move || {
match store.try_read_file(id, check_url_validity, origin) {
Ok(buffer) => { let _ = sender.send(Ok(buffer)); }
Err(e) => {
let _ = sender.send(Err(FileManagerThreadError::BlobURLStoreError(e)));
}
if let Err(e) = store.try_read_file(sender.clone(), id, check_url_validity, origin) {
let _ = sender.send(Err(FileManagerThreadError::BlobURLStoreError(e)));
}
})
}
Expand Down Expand Up @@ -370,8 +368,8 @@ impl <UI: 'static + UIProvider> FileManagerStore<UI> {
fn create_entry(&self, file_path: &Path, origin: &str) -> Result<SelectedFile, FileManagerThreadError> {
use net_traits::filemanager_thread::FileManagerThreadError::FileSystemError;

let handler = try!(File::open(file_path).map_err(|e| FileSystemError(e.to_string())));
let metadata = try!(handler.metadata().map_err(|e| FileSystemError(e.to_string())));
let file = try!(File::open(file_path).map_err(|e| FileSystemError(e.to_string())));
let metadata = try!(file.metadata().map_err(|e| FileSystemError(e.to_string())));
let modified = try!(metadata.modified().map_err(|e| FileSystemError(e.to_string())));
let elapsed = try!(modified.elapsed().map_err(|e| FileSystemError(e.to_string())));
// Unix Epoch: https://doc.servo.org/std/time/constant.UNIX_EPOCH.html
Expand Down Expand Up @@ -410,22 +408,28 @@ impl <UI: 'static + UIProvider> FileManagerStore<UI> {
})
}

fn get_blob_buf(&self, id: &Uuid, origin_in: &FileOrigin, rel_pos: RelativePos,
check_url_validity: bool) -> Result<BlobBuf, BlobURLStoreError> {
fn get_blob_buf(&self, sender: IpcSender<FileManagerResult<ReadFileProgress>>,
id: &Uuid, origin_in: &FileOrigin, rel_pos: RelativePos,
check_url_validity: bool) -> Result<(), BlobURLStoreError> {
let file_impl = try!(self.get_impl(id, origin_in, check_url_validity));
match file_impl {
FileImpl::Memory(buf) => {
let range = rel_pos.to_abs_range(buf.size as usize);
Ok(BlobBuf {
let buf = BlobBuf {
filename: None,
type_string: buf.type_string,
size: range.len() as u64,
bytes: buf.bytes.index(range).to_vec(),
})
};

let _ = sender.send(Ok(ReadFileProgress::Meta(buf)));
let _ = sender.send(Ok(ReadFileProgress::EOF));

Ok(())
}
FileImpl::MetaDataOnly(metadata) => {
/* XXX: Snapshot state check (optional) https://w3c.github.io/FileAPI/#snapshot-state.
Concretely, here we create another handler, and this handler might not
Concretely, here we create another file, and this file might not
has the same underlying file state (meta-info plus content) as the time
create_entry is called.
*/
Expand All @@ -437,42 +441,37 @@ impl <UI: 'static + UIProvider> FileManagerStore<UI> {
let mime = guess_mime_type_opt(metadata.path.clone());
let range = rel_pos.to_abs_range(metadata.size as usize);

let mut handler = try!(File::open(&metadata.path)
let mut file = try!(File::open(&metadata.path)
.map_err(|e| BlobURLStoreError::External(e.to_string())));
let seeked_start = try!(handler.seek(SeekFrom::Start(range.start as u64))
let seeked_start = try!(file.seek(SeekFrom::Start(range.start as u64))
.map_err(|e| BlobURLStoreError::External(e.to_string())));

if seeked_start == (range.start as u64) {
let mut bytes = vec![0; range.len()];
try!(handler.read_exact(&mut bytes)
.map_err(|e| BlobURLStoreError::External(e.to_string())));

Ok(BlobBuf {
filename: opt_filename,
type_string: match mime {
Some(x) => format!("{}", x),
None => "".to_string(),
},
size: range.len() as u64,
bytes: bytes,
})
let type_string = match mime {
Some(x) => format!("{}", x),
None => "".to_string(),
};

chunked_read(sender, &mut file, range.len(), opt_filename, type_string);
Ok(())
} else {
Err(BlobURLStoreError::InvalidEntry)
}
}
FileImpl::Sliced(parent_id, inner_rel_pos) => {
// Next time we don't need to check validity since
// we have already done that for requesting URL if necessary
self.get_blob_buf(&parent_id, origin_in, rel_pos.slice_inner(&inner_rel_pos), false)
self.get_blob_buf(sender, &parent_id, origin_in, rel_pos.slice_inner(&inner_rel_pos), false)
}
}
}

// Convenient wrapper over get_blob_buf
fn try_read_file(&self, id: SelectedFileId, check_url_validity: bool,
origin_in: FileOrigin) -> Result<BlobBuf, BlobURLStoreError> {
fn try_read_file(&self, sender: IpcSender<FileManagerResult<ReadFileProgress>>,
id: SelectedFileId, check_url_validity: bool,
origin_in: FileOrigin) -> Result<(), BlobURLStoreError> {
let id = try!(Uuid::parse_str(&id.0).map_err(|_| BlobURLStoreError::InvalidFileID));
self.get_blob_buf(&id, &origin_in, RelativePos::full_range(), check_url_validity)
self.get_blob_buf(sender, &id, &origin_in, RelativePos::full_range(), check_url_validity)
}

fn dec_ref(&self, id: &Uuid, origin_in: &FileOrigin,
Expand Down Expand Up @@ -563,3 +562,46 @@ fn select_files_pref_enabled() -> bool {
PREFS.get("dom.testing.htmlinputelement.select_files.enabled")
.as_boolean().unwrap_or(false)
}

const CHUNK_SIZE: usize = 8192;

fn chunked_read(sender: IpcSender<FileManagerResult<ReadFileProgress>>,
file: &mut File, size: usize, opt_filename: Option<String>, type_string: String) {
// First chunk
let mut buf = vec![0; CHUNK_SIZE];
match file.read(&mut buf) {
Ok(n) => {
buf.truncate(n);
let blob_buf = BlobBuf {
filename: opt_filename,
type_string: type_string,
size: size as u64,
bytes: buf,
};
let _ = sender.send(Ok(ReadFileProgress::Meta(blob_buf)));
}
Err(e) => {
let _ = sender.send(Err(FileManagerThreadError::FileSystemError(e.to_string())));
return;
}
}

// Send the remaining chunks
loop {
let mut buf = vec![0; CHUNK_SIZE];
match file.read(&mut buf) {
Ok(0) => {
let _ = sender.send(Ok(ReadFileProgress::EOF));
return;
}
Ok(n) => {
buf.truncate(n);
let _ = sender.send(Ok(ReadFileProgress::Partial(buf)));
}
Err(e) => {
let _ = sender.send(Err(FileManagerThreadError::FileSystemError(e.to_string())));
return;
}
}
}
}
12 changes: 9 additions & 3 deletions components/net_traits/filemanager_thread.rs
View file
Expand Up @@ -127,9 +127,8 @@ pub enum FileManagerThreadMsg {
/// Select multiple files, return a vector of triples
SelectFiles(Vec<FilterPattern>, IpcSender<FileManagerResult<Vec<SelectedFile>>>, FileOrigin, Option<Vec<String>>),

/// Read file by FileID, optionally check URL validity based on
/// third flag, return the blob buffer object
ReadFile(IpcSender<FileManagerResult<BlobBuf>>, SelectedFileId, bool, FileOrigin),
/// Read file in chunks by FileID, optionally check URL validity based on fourth flag
ReadFile(IpcSender<FileManagerResult<ReadFileProgress>>, SelectedFileId, bool, FileOrigin),

/// Add an entry as promoted memory-based blob and send back the associated FileID
/// as part of a valid/invalid Blob URL depending on the second bool flag
Expand All @@ -155,6 +154,13 @@ pub enum FileManagerThreadMsg {
Exit,
}

#[derive(Debug, Deserialize, Serialize)]
pub enum ReadFileProgress {
Meta(BlobBuf),
Partial(Vec<u8>),
EOF
}

pub type FileManagerResult<T> = Result<T, FileManagerThreadError>;

#[derive(Debug, Deserialize, Serialize)]
Expand Down
20 changes: 16 additions & 4 deletions components/script/dom/blob.rs
View file
Expand Up @@ -16,7 +16,7 @@ use encoding::types::{EncoderTrap, Encoding};
use ipc_channel::ipc;
use net_traits::IpcSend;
use net_traits::blob_url_store::{BlobBuf, get_blob_origin};
use net_traits::filemanager_thread::{FileManagerThreadMsg, SelectedFileId, RelativePos};
use net_traits::filemanager_thread::{FileManagerThreadMsg, SelectedFileId, RelativePos, ReadFileProgress};
use std::cell::Cell;
use std::ops::Index;
use std::path::PathBuf;
Expand Down Expand Up @@ -286,9 +286,21 @@ fn read_file(global: GlobalRef, id: SelectedFileId) -> Result<Vec<u8>, ()> {
let msg = FileManagerThreadMsg::ReadFile(chan, id, check_url_validity, origin);
let _ = file_manager.send(msg);

match recv.recv().unwrap() {
Ok(blob_buf) => Ok(blob_buf.bytes),
Err(_) => Err(()),
let mut bytes = vec![];

loop {
match recv.recv().unwrap() {
Ok(ReadFileProgress::Meta(mut blob_buf)) => {
bytes.append(&mut blob_buf.bytes);
}
Ok(ReadFileProgress::Partial(mut bytes_in)) => {
bytes.append(&mut bytes_in);
}
Ok(ReadFileProgress::EOF) => {
return Ok(bytes);
}
Err(_) => return Err(()),
}
}
}

Expand Down
43 changes: 31 additions & 12 deletions tests/unit/net/filemanager_thread.rs
View file
Expand Up @@ -5,7 +5,7 @@
use ipc_channel::ipc::{self, IpcSender};
use net::filemanager_thread::{FileManagerThreadFactory, UIProvider};
use net_traits::blob_url_store::BlobURLStoreError;
use net_traits::filemanager_thread::{FilterPattern, FileManagerThreadMsg, FileManagerThreadError};
use net_traits::filemanager_thread::{FilterPattern, FileManagerThreadMsg, FileManagerThreadError, ReadFileProgress};
use std::fs::File;
use std::io::Read;
use std::path::PathBuf;
Expand All @@ -16,38 +16,38 @@ struct TestProvider;

impl UIProvider for TestProvider {
fn open_file_dialog(&self, _path: &str, _patterns: Vec<FilterPattern>) -> Option<String> {
Some("test.txt".to_string())
Some("test.jpeg".to_string())
}

fn open_file_dialog_multi(&self, _path: &str, _patterns: Vec<FilterPattern>) -> Option<Vec<String>> {
Some(vec!["test.txt".to_string()])
Some(vec!["test.jpeg".to_string()])
}
}

#[test]
fn test_filemanager() {
let chan: IpcSender<FileManagerThreadMsg> = FileManagerThreadFactory::new(TEST_PROVIDER);

// Try to open a dummy file "tests/unit/net/test.txt" in tree
let mut handler = File::open("test.txt").expect("test.txt is stolen");
// Try to open a dummy file "tests/unit/net/test.jpeg" in tree
let mut handler = File::open("test.jpeg").expect("test.jpeg is stolen");
let mut test_file_content = vec![];

handler.read_to_end(&mut test_file_content)
.expect("Read tests/unit/net/test.txt error");
.expect("Read tests/unit/net/test.jpeg error");

let patterns = vec![FilterPattern(".txt".to_string())];
let origin = "test.com".to_string();

{
// Try to select a dummy file "tests/unit/net/test.txt"
// Try to select a dummy file "tests/unit/net/test.jpeg"
let (tx, rx) = ipc::channel().unwrap();
chan.send(FileManagerThreadMsg::SelectFile(patterns.clone(), tx, origin.clone(), None)).unwrap();
let selected = rx.recv().expect("Broken channel")
.expect("The file manager failed to find test.txt");
.expect("The file manager failed to find test.jpeg");

// Expecting attributes conforming the spec
assert!(selected.filename == PathBuf::from("test.txt"));
assert!(selected.type_string == "text/plain".to_string());
assert_eq!(selected.filename, PathBuf::from("test.jpeg"));
assert_eq!(selected.type_string, "image/jpeg".to_string());

// Test by reading, expecting same content
{
Expand All @@ -56,8 +56,27 @@ fn test_filemanager() {

let msg = rx2.recv().expect("Broken channel");

let blob_buf = msg.expect("File manager reading failure is unexpected");
assert_eq!(test_file_content, blob_buf.bytes, "Read content differs");
if let ReadFileProgress::Meta(blob_buf) = msg.expect("File manager reading failure is unexpected") {
let mut bytes = blob_buf.bytes;

loop {
match rx2.recv().expect("Broken channel").expect("File manager reading failure is unexpected") {
ReadFileProgress::Meta(_) => {
panic!("Invalid FileManager reply");
}
ReadFileProgress::Partial(mut bytes_in) => {
bytes.append(&mut bytes_in);
}
ReadFileProgress::EOF => {
break;
}
}
}

assert_eq!(test_file_content, bytes, "Read content differs");
} else {
panic!("Invalid FileManager reply");
}
}

// Delete the id
Expand Down
Binary file modified tests/unit/net/test.jpeg
View file
Sorry, something went wrong. Reload?
Sorry, we cannot display this file.
Sorry, this file is invalid so it cannot be displayed.

0 comments on commit 9c8ebd3

Please sign in to comment.