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

Get rid of 'Dapp is being downloaded' page

Browse Source
This commit is contained in:
Tomasz Drwięga 2016-09-06 22:12:52 +02:00
parent 9655920896
commit 6f88b7f084
8 changed files with 145 additions and 76 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
dapps/src/apps/cache.rs
View File

@ -18,13 +18,13 @@
use std::fs; use std::fs;
use std::sync::{Arc}; use std::sync::{Arc};
use std::sync::atomic::{AtomicBool, Ordering};
use linked_hash_map::LinkedHashMap; use linked_hash_map::LinkedHashMap;
use page::LocalPageEndpoint; use page::LocalPageEndpoint;
use handlers::FetchControl;
pub enum ContentStatus { pub enum ContentStatus {
Fetching(Arc<AtomicBool>), Fetching(Arc<FetchControl>),
Ready(LocalPageEndpoint), Ready(LocalPageEndpoint),
} }
@ -57,10 +57,10 @@ impl ContentCache {
while len > expected_size { while len > expected_size {
let entry = self.cache.pop_front().unwrap(); let entry = self.cache.pop_front().unwrap();
match entry.1 { match entry.1 {
ContentStatus::Fetching(ref abort) => { ContentStatus::Fetching(ref fetch) => {
trace!(target: "dapps", "Aborting {} because of limit.", entry.0); trace!(target: "dapps", "Aborting {} because of limit.", entry.0);
// Mark as aborted // Mark as aborted
abort.store(true, Ordering::SeqCst); fetch.abort()
}, },
ContentStatus::Ready(ref endpoint) => { ContentStatus::Ready(ref endpoint) => {
trace!(target: "dapps", "Removing {} because of limit.", entry.0); trace!(target: "dapps", "Removing {} because of limit.", entry.0);

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

@ -23,7 +23,6 @@ use std::{fs, env, fmt};
use std::io::{self, Read, Write}; use std::io::{self, Read, Write};
use std::path::PathBuf; use std::path::PathBuf;
use std::sync::Arc; use std::sync::Arc;
use std::sync::atomic::{AtomicBool};
use rustc_serialize::hex::FromHex; use rustc_serialize::hex::FromHex;
use hyper; use hyper;
@ -76,10 +75,12 @@ impl<R: URLHint> AppFetcher<R> {
} }
pub fn contains(&self, app_id: &str) -> bool { pub fn contains(&self, app_id: &str) -> bool {
let mut dapps = self.dapps.lock(); {
// Check if we already have the app let mut dapps = self.dapps.lock();
if dapps.get(app_id).is_some() { // Check if we already have the app
return true; if dapps.get(app_id).is_some() {
return true;
}
} }
// fallback to resolver // fallback to resolver
if let Ok(app_id) = app_id.from_hex() { if let Ok(app_id) = app_id.from_hex() {
@ -115,25 +116,19 @@ impl<R: URLHint> AppFetcher<R> {
(None, endpoint.to_async_handler(path, control)) (None, endpoint.to_async_handler(path, control))
}, },
// App is already being fetched // App is already being fetched
Some(&mut ContentStatus::Fetching(_)) => { Some(&mut ContentStatus::Fetching(ref fetch_control)) => {
(None, Box::new(ContentHandler::error_with_refresh( trace!(target: "dapps", "Content fetching in progress. Waiting...");
StatusCode::ServiceUnavailable, (None, fetch_control.to_handler(control))
"Download In Progress",
"This dapp is already being downloaded. Please wait...",
None,
)) as Box<Handler>)
}, },
// We need to start fetching app // We need to start fetching app
None => { None => {
trace!(target: "dapps", "Content fetching unavailable. Fetching...");
let app_hex = app_id.from_hex().expect("to_handler is called only when `contains` returns true."); let app_hex = app_id.from_hex().expect("to_handler is called only when `contains` returns true.");
let app = self.resolver.resolve(app_hex); let app = self.resolver.resolve(app_hex);
if let Some(app) = app { if let Some(app) = app {
let abort = Arc::new(AtomicBool::new(false)); let (handler, fetch_control) = ContentFetcherHandler::new(
(Some(ContentStatus::Fetching(abort.clone())), Box::new(ContentFetcherHandler::new(
app, app,
abort,
control, control,
path.using_dapps_domains, path.using_dapps_domains,
DappInstaller { DappInstaller {
@ -141,7 +136,9 @@ impl<R: URLHint> AppFetcher<R> {
dapps_path: self.dapps_path.clone(), dapps_path: self.dapps_path.clone(),
dapps: self.dapps.clone(), dapps: self.dapps.clone(),
} }
)) as Box<Handler>) );
(Some(ContentStatus::Fetching(fetch_control)), Box::new(handler) as Box<Handler>)
} else { } else {
// This may happen when sync status changes in between // This may happen when sync status changes in between
// `contains` and `to_handler` // `contains` and `to_handler`

1
dapps/src/error_tpl.html
View File

@ -3,7 +3,6 @@
<head> <head>
<meta charset="utf-8"> <meta charset="utf-8">
<meta name="viewport" content="width=device-width"> <meta name="viewport" content="width=device-width">
{meta}
<title>{title}</title> <title>{title}</title>
<link rel="stylesheet" href="/parity-utils/styles.css"> <link rel="stylesheet" href="/parity-utils/styles.css">
</head> </head>

13
dapps/src/handlers/content.rs
View File

@ -23,6 +23,7 @@ use hyper::status::StatusCode;
use util::version; use util::version;
#[derive(Clone)]
pub struct ContentHandler { pub struct ContentHandler {
code: StatusCode, code: StatusCode,
content: String, content: String,
@ -57,18 +58,6 @@ impl ContentHandler {
Self::html(code, format!( Self::html(code, format!(
include_str!("../error_tpl.html"), include_str!("../error_tpl.html"),
title=title, title=title,
meta="",
message=message,
details=details.unwrap_or_else(|| ""),
version=version(),
))
}
pub fn error_with_refresh(code: StatusCode, title: &str, message: &str, details: Option<&str>) -> Self {
Self::html(code, format!(
include_str!("../error_tpl.html"),
title=title,
meta="<meta http-equiv=\"refresh\" content=\"1\">",
message=message, message=message,
details=details.unwrap_or_else(|| ""), details=details.unwrap_or_else(|| ""),
version=version(), version=version(),

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

@ -16,17 +16,18 @@
//! Hyper Server Handler that fetches a file during a request (proxy). //! Hyper Server Handler that fetches a file during a request (proxy).
use std::fmt; use std::{fs, fmt};
use std::path::PathBuf; use std::path::PathBuf;
use std::sync::{mpsc, Arc}; use std::sync::{mpsc, Arc};
use std::sync::atomic::AtomicBool; use std::sync::atomic::{AtomicBool, Ordering};
use std::time::{Instant, Duration}; use std::time::{Instant, Duration};
use util::Mutex;
use hyper::{header, server, Decoder, Encoder, Next, Method, Control}; use hyper::{server, Decoder, Encoder, Next, Method, Control};
use hyper::net::HttpStream; use hyper::net::HttpStream;
use hyper::status::StatusCode; use hyper::status::StatusCode;
use handlers::ContentHandler; use handlers::{ContentHandler, Redirection};
use handlers::client::{Client, FetchResult}; use handlers::client::{Client, FetchResult};
use apps::redirection_address; use apps::redirection_address;
use apps::urlhint::GithubApp; use apps::urlhint::GithubApp;
@ -36,12 +37,9 @@ const FETCH_TIMEOUT: u64 = 30;
enum FetchState { enum FetchState {
NotStarted(GithubApp), NotStarted(GithubApp),
InProgress(mpsc::Receiver<FetchResult>),
Error(ContentHandler), Error(ContentHandler),
InProgress { Done(Manifest, Redirection),
deadline: Instant,
receiver: mpsc::Receiver<FetchResult>,
},
Done(Manifest),
} }
pub trait ContentValidator { pub trait ContentValidator {
@ -51,10 +49,93 @@ pub trait ContentValidator {
fn done(&self, Option<&Manifest>); fn done(&self, Option<&Manifest>);
} }
pub struct ContentFetcherHandler<H: ContentValidator> { pub struct FetchControl {
abort: Arc<AtomicBool>, abort: Arc<AtomicBool>,
control: Option<Control>, listeners: Mutex<Vec<(Control, mpsc::Sender<FetchState>)>>,
deadline: Instant,
}
impl Default for FetchControl {
fn default() -> Self {
FetchControl {
abort: Arc::new(AtomicBool::new(false)),
listeners: Mutex::new(Vec::new()),
deadline: Instant::now() + Duration::from_secs(FETCH_TIMEOUT),
}
}
}
impl FetchControl {
fn notify<F: Fn() -> FetchState>(&self, status: F) {
let mut listeners = self.listeners.lock();
for (control, sender) in listeners.drain(..) {
if let Err(e) = sender.send(status()) {
trace!(target: "dapps", "Waiting listener notification failed: {:?}", e);
} else {
let _ = control.ready(Next::read());
}
}
}
fn set_status(&self, status: &FetchState) {
match *status {
FetchState::Error(ref handler) => self.notify(|| FetchState::Error(handler.clone())),
FetchState::Done(ref manifest, ref handler) => self.notify(|| FetchState::Done(manifest.clone(), handler.clone())),
FetchState::NotStarted(_) | FetchState::InProgress(_) => {},
}
}
pub fn abort(&self) {
self.abort.store(true, Ordering::SeqCst);
}
pub fn to_handler(&self, 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,
})
}
}
pub struct WaitingHandler {
receiver: mpsc::Receiver<FetchState>,
state: Option<FetchState>,
}
impl server::Handler<HttpStream> for WaitingHandler {
fn on_request(&mut self, _request: server::Request<HttpStream>) -> Next {
Next::wait()
}
fn on_request_readable(&mut self, _decoder: &mut Decoder<HttpStream>) -> Next {
self.state = self.receiver.try_recv().ok();
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),
_ => 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),
_ => Next::end(),
}
}
}
pub struct ContentFetcherHandler<H: ContentValidator> {
fetch_control: Arc<FetchControl>,
status: FetchState, status: FetchState,
control: Option<Control>,
client: Option<Client>, client: Option<Client>,
using_dapps_domains: bool, using_dapps_domains: bool,
dapp: H, dapp: H,
@ -63,7 +144,7 @@ pub struct ContentFetcherHandler<H: ContentValidator> {
impl<H: ContentValidator> Drop for ContentFetcherHandler<H> { impl<H: ContentValidator> Drop for ContentFetcherHandler<H> {
fn drop(&mut self) { fn drop(&mut self) {
let manifest = match self.status { let manifest = match self.status {
FetchState::Done(ref manifest) => Some(manifest), FetchState::Done(ref manifest, _) => Some(manifest),
_ => None, _ => None,
}; };
self.dapp.done(manifest); self.dapp.done(manifest);
@ -74,20 +155,22 @@ impl<H: ContentValidator> ContentFetcherHandler<H> {
pub fn new( pub fn new(
app: GithubApp, app: GithubApp,
abort: Arc<AtomicBool>,
control: Control, control: Control,
using_dapps_domains: bool, using_dapps_domains: bool,
handler: H) -> Self { handler: H) -> (Self, Arc<FetchControl>) {
let fetch_control = Arc::new(FetchControl::default());
let client = Client::new(); let client = Client::new();
ContentFetcherHandler { let handler = ContentFetcherHandler {
abort: abort, fetch_control: fetch_control.clone(),
control: Some(control), control: Some(control),
client: Some(client), client: Some(client),
status: FetchState::NotStarted(app), status: FetchState::NotStarted(app),
using_dapps_domains: using_dapps_domains, using_dapps_domains: using_dapps_domains,
dapp: handler, dapp: handler,
} };
(handler, fetch_control)
} }
fn close_client(client: &mut Option<Client>) { fn close_client(client: &mut Option<Client>) {
@ -96,13 +179,13 @@ impl<H: ContentValidator> ContentFetcherHandler<H> {
.close(); .close();
} }
fn fetch_app(client: &mut Client, app: &GithubApp, abort: Arc<AtomicBool>, control: Control) -> Result<mpsc::Receiver<FetchResult>, String> { fn fetch_app(client: &mut Client, app: &GithubApp, abort: Arc<AtomicBool>, control: Control) -> Result<mpsc::Receiver<FetchResult>, String> {
client.request(app.url(), abort, Box::new(move || { let res = client.request(app.url(), abort, Box::new(move || {
trace!(target: "dapps", "Fetching finished."); trace!(target: "dapps", "Fetching finished.");
// Ignoring control errors // Ignoring control errors
let _ = control.ready(Next::read()); let _ = control.ready(Next::read());
})).map_err(|e| format!("{:?}", e)) })).map_err(|e| format!("{:?}", e));
res
} }
} }
@ -115,12 +198,9 @@ impl<H: ContentValidator> server::Handler<HttpStream> for ContentFetcherHandler<
trace!(target: "dapps", "Fetching dapp: {:?}", app); trace!(target: "dapps", "Fetching dapp: {:?}", app);
let control = self.control.take().expect("on_request is called only once, thus control is always Some"); let control = self.control.take().expect("on_request is called only once, thus control is always Some");
let client = self.client.as_mut().expect("on_request is called before client is closed."); let client = self.client.as_mut().expect("on_request is called before client is closed.");
let fetch = Self::fetch_app(client, app, self.abort.clone(), control); let fetch = Self::fetch_app(client, app, self.fetch_control.abort.clone(), control);
match fetch { match fetch {
Ok(receiver) => FetchState::InProgress { Ok(receiver) => FetchState::InProgress(receiver),
deadline: Instant::now() + Duration::from_secs(FETCH_TIMEOUT),
receiver: receiver,
},
Err(e) => FetchState::Error(ContentHandler::error( Err(e) => FetchState::Error(ContentHandler::error(
StatusCode::BadGateway, StatusCode::BadGateway,
"Unable To Start Dapp Download", "Unable To Start Dapp Download",
@ -140,6 +220,7 @@ impl<H: ContentValidator> server::Handler<HttpStream> for ContentFetcherHandler<
} else { None }; } else { None };
if let Some(status) = status { if let Some(status) = status {
self.fetch_control.set_status(&status);
self.status = status; self.status = status;
} }
@ -149,7 +230,7 @@ impl<H: ContentValidator> server::Handler<HttpStream> for ContentFetcherHandler<
fn on_request_readable(&mut self, decoder: &mut Decoder<HttpStream>) -> Next { fn on_request_readable(&mut self, decoder: &mut Decoder<HttpStream>) -> Next {
let (status, next) = match self.status { let (status, next) = match self.status {
// Request may time out // Request may time out
FetchState::InProgress { ref deadline, .. } if *deadline < Instant::now() => { FetchState::InProgress(_) if self.fetch_control.deadline < Instant::now() => {
trace!(target: "dapps", "Fetching dapp failed because of timeout."); trace!(target: "dapps", "Fetching dapp failed because of timeout.");
let timeout = ContentHandler::error( let timeout = ContentHandler::error(
StatusCode::GatewayTimeout, StatusCode::GatewayTimeout,
@ -160,7 +241,7 @@ impl<H: ContentValidator> server::Handler<HttpStream> for ContentFetcherHandler<
Self::close_client(&mut self.client); Self::close_client(&mut self.client);
(Some(FetchState::Error(timeout)), Next::write()) (Some(FetchState::Error(timeout)), Next::write())
}, },
FetchState::InProgress { ref receiver, .. } => { FetchState::InProgress(ref receiver) => {
// Check if there is an answer // Check if there is an answer
let rec = receiver.try_recv(); let rec = receiver.try_recv();
match rec { match rec {
@ -179,11 +260,13 @@ impl<H: ContentValidator> server::Handler<HttpStream> for ContentFetcherHandler<
Some(&format!("{:?}", e)) Some(&format!("{:?}", e))
)) ))
}, },
Ok(manifest) => FetchState::Done(manifest) Ok(manifest) => {
let address = redirection_address(self.using_dapps_domains, &manifest.id);
FetchState::Done(manifest, Redirection::new(&address))
},
}; };
// Remove temporary zip file // Remove temporary zip file
// TODO [todr] Uncomment me let _ = fs::remove_file(path);
// let _ = fs::remove_file(path);
(Some(state), Next::write()) (Some(state), Next::write())
}, },
Ok(Err(e)) => { Ok(Err(e)) => {
@ -205,6 +288,7 @@ impl<H: ContentValidator> server::Handler<HttpStream> for ContentFetcherHandler<
}; };
if let Some(status) = status { if let Some(status) = status {
self.fetch_control.set_status(&status);
self.status = status; self.status = status;
} }
@ -213,12 +297,7 @@ impl<H: ContentValidator> server::Handler<HttpStream> for ContentFetcherHandler<
fn on_response(&mut self, res: &mut server::Response) -> Next { fn on_response(&mut self, res: &mut server::Response) -> Next {
match self.status { match self.status {
FetchState::Done(ref manifest) => { FetchState::Done(_, ref mut handler) => handler.on_response(res),
trace!(target: "dapps", "Fetching dapp finished. Redirecting to {}", manifest.id);
res.set_status(StatusCode::Found);
res.headers_mut().set(header::Location(redirection_address(self.using_dapps_domains, &manifest.id)));
Next::write()
},
FetchState::Error(ref mut handler) => handler.on_response(res), FetchState::Error(ref mut handler) => handler.on_response(res),
_ => Next::end(), _ => Next::end(),
} }
@ -226,9 +305,9 @@ impl<H: ContentValidator> server::Handler<HttpStream> for ContentFetcherHandler<
fn on_response_writable(&mut self, encoder: &mut Encoder<HttpStream>) -> Next { fn on_response_writable(&mut self, encoder: &mut Encoder<HttpStream>) -> Next {
match self.status { match self.status {
FetchState::Done(_, ref mut handler) => handler.on_response_writable(encoder),
FetchState::Error(ref mut handler) => handler.on_response_writable(encoder), FetchState::Error(ref mut handler) => handler.on_response_writable(encoder),
_ => Next::end(), _ => Next::end(),
} }
} }
} }

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

@ -27,7 +27,7 @@ pub use self::auth::AuthRequiredHandler;
pub use self::echo::EchoHandler; pub use self::echo::EchoHandler;
pub use self::content::ContentHandler; pub use self::content::ContentHandler;
pub use self::redirect::Redirection; pub use self::redirect::Redirection;
pub use self::fetch::{ContentFetcherHandler, ContentValidator}; pub use self::fetch::{ContentFetcherHandler, ContentValidator, FetchControl};
use url::Url; use url::Url;
use hyper::{server, header, net, uri}; use hyper::{server, header, net, uri};

11
dapps/src/handlers/redirect.rs
View File

@ -20,15 +20,20 @@ use hyper::{header, server, Decoder, Encoder, Next};
use hyper::net::HttpStream; use hyper::net::HttpStream;
use hyper::status::StatusCode; use hyper::status::StatusCode;
#[derive(Clone)]
pub struct Redirection { pub struct Redirection {
to_url: String to_url: String
} }
impl Redirection { impl Redirection {
pub fn new(url: &str) -> Box<Self> { pub fn new(url: &str) -> Self {
Box::new(Redirection { Redirection {
to_url: url.to_owned() to_url: url.to_owned()
}) }
}
pub fn boxed(url: &str) -> Box<Self> {
Box::new(Self::new(url))
} }
} }

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

@ -104,7 +104,7 @@ impl<A: Authorization + 'static> server::Handler<HttpStream> for Router<A> {
// Redirect any GET request to home. // Redirect any GET request to home.
_ if *req.method() == hyper::method::Method::Get => { _ if *req.method() == hyper::method::Method::Get => {
let address = apps::redirection_address(false, self.main_page); let address = apps::redirection_address(false, self.main_page);
Redirection::new(address.as_str()) Redirection::boxed(address.as_str())
}, },
// RPC by default // RPC by default
_ => { _ => {