grassrootseconomics/openethereum
5
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Merge pull request #3858 from ethcore/fetcher-ref

Browse Source 
Get rid of unecessary redirection while fetching content
This commit is contained in:
Gav Wood 2016-12-15 23:08:37 +01:00 committed by GitHub
commit 6ea5638240
5 changed files with 213 additions and 161 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

48
dapps/src/apps/fetcher.rs
View File

@ -120,7 +120,7 @@ impl<R: URLHint> ContentFetcher<R> {
// Content is already being fetched
Some(&mut ContentStatus::Fetching(ref fetch_control)) => {
trace!(target: "dapps", "Content fetching in progress. Waiting...");
(None, fetch_control.to_handler(control))
(None, fetch_control.to_async_handler(path, control))
},
// We need to start fetching the content
None => {
@ -129,11 +129,12 @@ impl<R: URLHint> ContentFetcher<R> {
let content = self.resolver.resolve(content_hex);
let cache = self.cache.clone();
let on_done = move |id: String, result: Option<LocalPageEndpoint>| {
let id = content_id.clone();
let on_done = move |result: Option<LocalPageEndpoint>| {
let mut cache = cache.lock();
match result {
Some(endpoint) => {
cache.insert(id, ContentStatus::Ready(endpoint));
cache.insert(id.clone(), ContentStatus::Ready(endpoint));
},
// In case of error
None => {
@ -150,6 +151,7 @@ impl<R: URLHint> ContentFetcher<R> {
Some(URLHintResult::Dapp(dapp)) => {
let (handler, fetch_control) = ContentFetcherHandler::new(
dapp.url(),
path,
control,
DappInstaller {
id: content_id.clone(),
@ -165,6 +167,7 @@ impl<R: URLHint> ContentFetcher<R> {
Some(URLHintResult::Content(content)) => {
let (handler, fetch_control) = ContentFetcherHandler::new(
content.url,
path,
control,
ContentInstaller {
id: content_id.clone(),
@ -248,13 +251,14 @@ struct ContentInstaller {
id: String,
mime: String,
content_path: PathBuf,
on_done: Box<Fn(String, Option<LocalPageEndpoint>) + Send>,
on_done: Box<Fn(Option<LocalPageEndpoint>) + Send>,
}
impl ContentValidator for ContentInstaller {
type Error = ValidationError;
fn validate_and_install(&self, path: PathBuf) -> Result<(String, LocalPageEndpoint), ValidationError> {
fn validate_and_install(&self, path: PathBuf) -> Result<LocalPageEndpoint, ValidationError> {
let validate = || {
// Create dir
try!(fs::create_dir_all(&self.content_path));
@ -279,12 +283,13 @@ impl ContentValidator for ContentInstaller {
}
try!(fs::copy(&path, &content_path));
Ok(LocalPageEndpoint::single_file(content_path, self.mime.clone(), PageCache::Enabled))
};
Ok((self.id.clone(), LocalPageEndpoint::single_file(content_path, self.mime.clone(), PageCache::Enabled)))
}
fn done(&self, endpoint: Option<LocalPageEndpoint>) {
(self.on_done)(self.id.clone(), endpoint)
// Make sure to always call on_done (even in case of errors)!
let result = validate();
(self.on_done)(result.as_ref().ok().cloned());
result
}
}
@ -292,7 +297,7 @@ impl ContentValidator for ContentInstaller {
struct DappInstaller {
id: String,
dapps_path: PathBuf,
on_done: Box<Fn(String, Option<LocalPageEndpoint>) + Send>,
on_done: Box<Fn(Option<LocalPageEndpoint>) + Send>,
embeddable_on: Option<(String, u16)>,
}
@ -331,9 +336,10 @@ impl DappInstaller {
impl ContentValidator for DappInstaller {
type Error = ValidationError;
fn validate_and_install(&self, app_path: PathBuf) -> Result<(String, LocalPageEndpoint), ValidationError> {
trace!(target: "dapps", "Opening dapp bundle at {:?}", app_path);
let mut file_reader = io::BufReader::new(try!(fs::File::open(app_path)));
fn validate_and_install(&self, path: PathBuf) -> Result<LocalPageEndpoint, ValidationError> {
trace!(target: "dapps", "Opening dapp bundle at {:?}", path);
let validate = || {
let mut file_reader = io::BufReader::new(try!(fs::File::open(path)));
let hash = try!(sha3(&mut file_reader));
let id = try!(self.id.as_str().parse().map_err(|_| ValidationError::InvalidContentId));
if id != hash {
@ -384,16 +390,14 @@ impl ContentValidator for DappInstaller {
let manifest_path = target.join(MANIFEST_FILENAME);
let mut manifest_file = try!(fs::File::create(manifest_path));
try!(manifest_file.write_all(manifest_str.as_bytes()));
// Create endpoint
let app = LocalPageEndpoint::new(target, manifest.clone().into(), PageCache::Enabled, self.embeddable_on.clone());
let endpoint = LocalPageEndpoint::new(target, manifest.clone().into(), PageCache::Enabled, self.embeddable_on.clone());
Ok(endpoint)
};
// Return modified app manifest
Ok((manifest.id.clone(), app))
}
fn done(&self, endpoint: Option<LocalPageEndpoint>) {
(self.on_done)(self.id.clone(), endpoint)
let result = validate();
(self.on_done)(result.as_ref().ok().cloned());
result
}
}

113
dapps/src/handlers/fetch.rs
View File

@ -22,35 +22,41 @@ use std::sync::{mpsc, Arc};
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::{Instant, Duration};
use util::Mutex;
use url::Url;
use fetch::{Client, Fetch, FetchResult};
use hyper::{server, Decoder, Encoder, Next, Method, Control};
use hyper::net::HttpStream;
use hyper::uri::RequestUri;
use hyper::status::StatusCode;
use handlers::{ContentHandler, Redirection, extract_url};
use page::LocalPageEndpoint;
use endpoint::EndpointPath;
use handlers::ContentHandler;
use page::{LocalPageEndpoint, PageHandlerWaiting};
const FETCH_TIMEOUT: u64 = 30;
enum FetchState {
Waiting,
NotStarted(String),
Error(ContentHandler),
InProgress(mpsc::Receiver<FetchResult>),
Done(String, LocalPageEndpoint, Redirection),
Done(LocalPageEndpoint, Box<PageHandlerWaiting>),
}
enum WaitResult {
Error(ContentHandler),
Done(LocalPageEndpoint),
}
pub trait ContentValidator {
type Error: fmt::Debug + fmt::Display;
fn validate_and_install(&self, app: PathBuf) -> Result<(String, LocalPageEndpoint), Self::Error>;
fn done(&self, Option<LocalPageEndpoint>);
fn validate_and_install(&self, path: PathBuf) -> Result<LocalPageEndpoint, Self::Error>;
}
pub struct FetchControl {
abort: Arc<AtomicBool>,
listeners: Mutex<Vec<(Control, mpsc::Sender<FetchState>)>>,
listeners: Mutex<Vec<(Control, mpsc::Sender<WaitResult>)>>,
deadline: Instant,
}
@ -65,9 +71,10 @@ impl Default for FetchControl {
}
impl FetchControl {
fn notify<F: Fn() -> FetchState>(&self, status: F) {
fn notify<F: Fn() -> WaitResult>(&self, status: F) {
let mut listeners = self.listeners.lock();
for (control, sender) in listeners.drain(..) {
trace!(target: "dapps", "Resuming request waiting for content...");
if let Err(e) = sender.send(status()) {
trace!(target: "dapps", "Waiting listener notification failed: {:?}", e);
} else {
@ -78,9 +85,9 @@ impl FetchControl {
fn set_status(&self, status: &FetchState) {
match *status {
FetchState::Error(ref handler) => self.notify(|| FetchState::Error(handler.clone())),
FetchState::Done(ref id, ref endpoint, ref handler) => self.notify(|| FetchState::Done(id.clone(), endpoint.clone(), handler.clone())),
FetchState::NotStarted(_) | FetchState::InProgress(_) => {},
FetchState::Error(ref handler) => self.notify(|| WaitResult::Error(handler.clone())),
FetchState::Done(ref endpoint, _) => self.notify(|| WaitResult::Done(endpoint.clone())),
FetchState::NotStarted(_) | FetchState::InProgress(_) | FetchState::Waiting => {},
}
}
@ -88,44 +95,66 @@ impl FetchControl {
self.abort.store(true, Ordering::SeqCst);
}
pub fn to_handler(&self, control: Control) -> Box<server::Handler<HttpStream> + Send> {
pub fn to_async_handler(&self, path: EndpointPath, control: Control) -> Box<server::Handler<HttpStream> + Send> {
let (tx, rx) = mpsc::channel();
self.listeners.lock().push((control, tx));
Box::new(WaitingHandler {
receiver: rx,
state: None,
state: FetchState::Waiting,
uri: RequestUri::default(),
path: path,
})
}
}
pub struct WaitingHandler {
receiver: mpsc::Receiver<FetchState>,
state: Option<FetchState>,
receiver: mpsc::Receiver<WaitResult>,
state: FetchState,
uri: RequestUri,
path: EndpointPath,
}
impl server::Handler<HttpStream> for WaitingHandler {
fn on_request(&mut self, _request: server::Request<HttpStream>) -> Next {
fn on_request(&mut self, request: server::Request<HttpStream>) -> Next {
self.uri = request.uri().clone();
Next::wait()
}
fn on_request_readable(&mut self, _decoder: &mut Decoder<HttpStream>) -> Next {
self.state = self.receiver.try_recv().ok();
Next::write()
fn on_request_readable(&mut self, decoder: &mut Decoder<HttpStream>) -> Next {
let result = self.receiver.try_recv().ok();
self.state = match result {
Some(WaitResult::Error(handler)) => FetchState::Error(handler),
Some(WaitResult::Done(endpoint)) => {
let mut page_handler = endpoint.to_page_handler(self.path.clone());
page_handler.set_uri(&self.uri);
FetchState::Done(endpoint, page_handler)
},
None => {
warn!("A result for waiting request was not received.");
FetchState::Waiting
},
};
match self.state {
FetchState::Done(_, ref mut handler) => handler.on_request_readable(decoder),
FetchState::Error(ref mut handler) => handler.on_request_readable(decoder),
_ => Next::write(),
}
}
fn on_response(&mut self, res: &mut server::Response) -> Next {
match self.state {
Some(FetchState::Done(_, _, ref mut handler)) => handler.on_response(res),
Some(FetchState::Error(ref mut handler)) => handler.on_response(res),
FetchState::Done(_, ref mut handler) => handler.on_response(res),
FetchState::Error(ref mut handler) => handler.on_response(res),
_ => Next::end(),
}
}
fn on_response_writable(&mut self, encoder: &mut Encoder<HttpStream>) -> Next {
match self.state {
Some(FetchState::Done(_, _, ref mut handler)) => handler.on_response_writable(encoder),
Some(FetchState::Error(ref mut handler)) => handler.on_response_writable(encoder),
FetchState::Done(_, ref mut handler) => handler.on_response_writable(encoder),
FetchState::Error(ref mut handler) => handler.on_response_writable(encoder),
_ => Next::end(),
}
}
@ -137,29 +166,19 @@ pub struct ContentFetcherHandler<H: ContentValidator> {
status: FetchState,
client: Option<Client>,
installer: H,
request_url: Option<Url>,
path: EndpointPath,
uri: RequestUri,
embeddable_on: Option<(String, u16)>,
}
impl<H: ContentValidator> Drop for ContentFetcherHandler<H> {
fn drop(&mut self) {
let result = match self.status {
FetchState::Done(_, ref result, _) => Some(result.clone()),
_ => None,
};
self.installer.done(result);
}
}
impl<H: ContentValidator> ContentFetcherHandler<H> {
pub fn new(
url: String,
path: EndpointPath,
control: Control,
handler: H,
embeddable_on: Option<(String, u16)>,
) -> (Self, Arc<FetchControl>) {
let fetch_control = Arc::new(FetchControl::default());
let client = Client::default();
let handler = ContentFetcherHandler {
@ -168,7 +187,8 @@ impl<H: ContentValidator> ContentFetcherHandler<H> {
client: Some(client),
status: FetchState::NotStarted(url),
installer: handler,
request_url: None,
path: path,
uri: RequestUri::default(),
embeddable_on: embeddable_on,
};
@ -192,7 +212,6 @@ impl<H: ContentValidator> ContentFetcherHandler<H> {
impl<H: ContentValidator> server::Handler<HttpStream> for ContentFetcherHandler<H> {
fn on_request(&mut self, request: server::Request<HttpStream>) -> Next {
self.request_url = extract_url(&request);
let status = if let FetchState::NotStarted(ref url) = self.status {
Some(match *request.method() {
// Start fetching content
@ -205,8 +224,8 @@ impl<H: ContentValidator> server::Handler<HttpStream> for ContentFetcherHandler<
Ok(receiver) => FetchState::InProgress(receiver),
Err(e) => FetchState::Error(ContentHandler::error(
StatusCode::BadGateway,
"Unable To Start Dapp Download",
"Could not initialize download of the dapp. It might be a problem with the remote server.",
"Unable To Start Content Download",
"Could not initialize download of the content. It might be a problem with the remote server.",
Some(&format!("{}", e)),
self.embeddable_on.clone(),
)),
@ -227,6 +246,7 @@ impl<H: ContentValidator> server::Handler<HttpStream> for ContentFetcherHandler<
self.fetch_control.set_status(&status);
self.status = status;
}
self.uri = request.uri().clone();
Next::read()
}
@ -266,11 +286,10 @@ impl<H: ContentValidator> server::Handler<HttpStream> for ContentFetcherHandler<
self.embeddable_on.clone(),
))
},
Ok((id, result)) => {
let url: String = self.request_url.take()
.map(|url| url.raw.into_string())
.expect("Request URL always read in on_request; qed");
FetchState::Done(id, result, Redirection::new(&url))
Ok(endpoint) => {
let mut handler = endpoint.to_page_handler(self.path.clone());
handler.set_uri(&self.uri);
FetchState::Done(endpoint, handler)
},
};
// Remove temporary zip file
@ -306,7 +325,7 @@ impl<H: ContentValidator> server::Handler<HttpStream> for ContentFetcherHandler<
fn on_response(&mut self, res: &mut server::Response) -> Next {
match self.status {
FetchState::Done(_, _, ref mut handler) => handler.on_response(res),
FetchState::Done(_, ref mut handler) => handler.on_response(res),
FetchState::Error(ref mut handler) => handler.on_response(res),
_ => Next::end(),
}
@ -314,7 +333,7 @@ impl<H: ContentValidator> server::Handler<HttpStream> for ContentFetcherHandler<
fn on_response_writable(&mut self, encoder: &mut Encoder<HttpStream>) -> Next {
match self.status {
FetchState::Done(_, _, ref mut handler) => handler.on_response_writable(encoder),
FetchState::Done(_, ref mut handler) => handler.on_response_writable(encoder),
FetchState::Error(ref mut handler) => handler.on_response_writable(encoder),
_ => Next::end(),
}

33
dapps/src/page/handler.rs
View File

@ -83,13 +83,19 @@ impl Default for PageCache {
}
}
/// A generic type for `PageHandler` allowing to set the URL.
/// Used by dapps fetching to set the URL after the content was downloaded.
pub trait PageHandlerWaiting: server::Handler<HttpStream> + Send {
fn set_uri(&mut self, uri: &RequestUri);
}
/// A handler for a single webapp.
/// Resolves correct paths and serves as a plumbing code between
/// hyper server and dapp.
pub struct PageHandler<T: Dapp> {
/// A Dapp.
pub app: T,
/// File currently being served (or `None` if file does not exist).
/// File currently being served
pub file: ServedFile<T>,
/// Optional prefix to strip from path.
pub prefix: Option<String>,
@ -101,6 +107,21 @@ pub struct PageHandler<T: Dapp> {
pub cache: PageCache,
}
impl<T: Dapp> PageHandlerWaiting for PageHandler<T> {
fn set_uri(&mut self, uri: &RequestUri) {
trace!(target: "dapps", "Setting URI: {:?}", uri);
self.file = match *uri {
RequestUri::AbsolutePath { ref path, .. } => {
self.app.file(&self.extract_path(path))
},
RequestUri::AbsoluteUri(ref url) => {
self.app.file(&self.extract_path(url.path()))
},
_ => None,
}.map_or_else(|| ServedFile::new(self.safe_to_embed_on.clone()), |f| ServedFile::File(f));
}
}
impl<T: Dapp> PageHandler<T> {
fn extract_path(&self, path: &str) -> String {
let app_id = &self.path.app_id;
@ -124,15 +145,7 @@ impl<T: Dapp> PageHandler<T> {
impl<T: Dapp> server::Handler<HttpStream> for PageHandler<T> {
fn on_request(&mut self, req: server::Request<HttpStream>) -> Next {
self.file = match *req.uri() {
RequestUri::AbsolutePath { ref path, .. } => {
self.app.file(&self.extract_path(path))
},
RequestUri::AbsoluteUri(ref url) => {
self.app.file(&self.extract_path(url.path()))
},
_ => None,
}.map_or_else(|| ServedFile::new(self.safe_to_embed_on.clone()), |f| ServedFile::File(f));
self.set_uri(req.uri());
Next::write()
}

50
dapps/src/page/local.rs
View File

@ -18,7 +18,7 @@ use mime_guess;
use std::io::{Seek, Read, SeekFrom};
use std::fs;
use std::path::{Path, PathBuf};
use page::handler::{self, PageCache};
use page::handler::{self, PageCache, PageHandlerWaiting};
use endpoint::{Endpoint, EndpointInfo, EndpointPath, Handler};
#[derive(Debug, Clone)]
@ -54,6 +54,36 @@ impl LocalPageEndpoint {
pub fn path(&self) -> PathBuf {
self.path.clone()
}
fn page_handler_with_mime(&self, path: EndpointPath, mime: &str) -> handler::PageHandler<LocalSingleFile> {
handler::PageHandler {
app: LocalSingleFile { path: self.path.clone(), mime: mime.into() },
prefix: None,
path: path,
file: handler::ServedFile::new(None),
safe_to_embed_on: self.embeddable_on.clone(),
cache: self.cache,
}
}
fn page_handler(&self, path: EndpointPath) -> handler::PageHandler<LocalDapp> {
handler::PageHandler {
app: LocalDapp { path: self.path.clone() },
prefix: None,
path: path,
file: handler::ServedFile::new(None),
safe_to_embed_on: self.embeddable_on.clone(),
cache: self.cache,
}
}
pub fn to_page_handler(&self, path: EndpointPath) -> Box<PageHandlerWaiting> {
if let Some(ref mime) = self.mime {
Box::new(self.page_handler_with_mime(path, mime))
} else {
Box::new(self.page_handler(path))
}
}
}
impl Endpoint for LocalPageEndpoint {
@ -63,23 +93,9 @@ impl Endpoint for LocalPageEndpoint {
fn to_handler(&self, path: EndpointPath) -> Box<Handler> {
if let Some(ref mime) = self.mime {
Box::new(handler::PageHandler {
app: LocalSingleFile { path: self.path.clone(), mime: mime.clone() },
prefix: None,
path: path,
file: handler::ServedFile::new(None),
safe_to_embed_on: self.embeddable_on.clone(),
cache: self.cache,
})
Box::new(self.page_handler_with_mime(path, mime))
} else {
Box::new(handler::PageHandler {
app: LocalDapp { path: self.path.clone() },
prefix: None,
path: path,
file: handler::ServedFile::new(None),
safe_to_embed_on: self.embeddable_on.clone(),
cache: self.cache,
})
Box::new(self.page_handler(path))
}
}
}

2
dapps/src/page/mod.rs
View File

@ -21,5 +21,5 @@ mod handler;
pub use self::local::LocalPageEndpoint;
pub use self::builtin::PageEndpoint;
pub use self::handler::PageCache;
pub use self::handler::{PageCache, PageHandlerWaiting};