Merge remote-tracking branch 'upstream/master' into hardcoded-sync

Cargo.toml

@@ -14,7 +14,6 @@ docopt = "0.8"
 clap = "2"
 term_size = "0.3"
 textwrap = "0.9"
-time = "0.1"
 num_cpus = "1.2"
 number_prefix = "0.2"
 rpassword = "1.0"
@@ -43,6 +42,7 @@ ethcore-logger = { path = "logger" }
 ethcore-migrations = { path = "ethcore/migrations" }
 ethcore-miner = { path = "miner" }
 ethcore-network = { path = "util/network" }
+ethcore-service = { path = "ethcore/service" }
 ethcore-stratum = { path = "stratum" }
 ethcore-transaction = { path = "ethcore/transaction" }
 ethereum-types = "0.2"
@@ -72,6 +72,8 @@ journaldb = { path = "util/journaldb" }
 parity-dapps = { path = "dapps", optional = true }
 ethcore-secretstore = { path = "secret_store", optional = true }
 
+registrar = { path = "registrar" }
+
 [build-dependencies]
 rustc_version = "0.2"
 
@@ -130,6 +132,5 @@ members = [
 	"evmbin",
 	"miner",
 	"transaction-pool",
-	"whisper",
-	"util/rlp_compress"
+	"whisper"
 ]

dapps/Cargo.toml

@@ -37,6 +37,7 @@ parity-reactor = { path = "../util/reactor" }
 parity-ui = { path = "./ui" }
 keccak-hash = { path = "../util/hash" }
 parity-version = { path = "../util/version" }
+registrar = { path = "../registrar" }
 
 [dev-dependencies]
 env_logger = "0.4"

dapps/src/apps/fetcher/installers.rs

@@ -19,9 +19,10 @@ use std::{fs, fmt};
 use std::io::{self, Read, Write};
 use std::path::PathBuf;
 use ethereum_types::H256;
-use fetch::{self, Mime};
+use fetch;
 use futures_cpupool::CpuPool;
 use hash::keccak_buffer;
+use mime_guess::Mime;
 
 use apps::manifest::{MANIFEST_FILENAME, deserialize_manifest, serialize_manifest, Manifest};
 use handlers::{ContentValidator, ValidatorResponse};
@@ -53,7 +54,7 @@ fn write_response_and_check_hash(
 
 	// Now write the response
 	let mut file = io::BufWriter::new(fs::File::create(&content_path)?);
-	let mut reader = io::BufReader::new(response);
+	let mut reader = io::BufReader::new(fetch::BodyReader::new(response));
 	io::copy(&mut reader, &mut file)?;
 	file.flush()?;
 

dapps/src/apps/fetcher/mod.rs

@@ -216,6 +216,7 @@ impl<R: URLHint + 'static, F: Fetch> Endpoint for ContentFetcher<F, R> {
 										),
 										self.embeddable_on.clone(),
 										self.fetch.clone(),
+										self.pool.clone(),
 									)
 								},
 							    URLHintResult::GithubDapp(content) => {
@@ -232,6 +233,7 @@ impl<R: URLHint + 'static, F: Fetch> Endpoint for ContentFetcher<F, R> {
 										),
 										self.embeddable_on.clone(),
 										self.fetch.clone(),
+										self.pool.clone(),
 									)
 							    },
 								URLHintResult::Content(content) => {
@@ -248,6 +250,7 @@ impl<R: URLHint + 'static, F: Fetch> Endpoint for ContentFetcher<F, R> {
 										),
 										self.embeddable_on.clone(),
 										self.fetch.clone(),
+										self.pool.clone(),
 									)
 								},
 							};
@@ -280,7 +283,7 @@ impl<R: URLHint + 'static, F: Fetch> Endpoint for ContentFetcher<F, R> {
 mod tests {
 	use std::env;
 	use std::sync::Arc;
-	use fetch::{Fetch, Client};
+	use fetch::Client;
 	use futures::{future, Future};
 	use hash_fetch::urlhint::{URLHint, URLHintResult};
 	use ethereum_types::H256;

dapps/src/apps/mod.rs

@@ -81,7 +81,7 @@ pub fn all_endpoints<F: Fetch>(
 	insert::<parity_ui::old::App>(&mut pages, "v1", Embeddable::Yes(embeddable.clone()), pool.clone());
 
 	pages.insert("proxy".into(), ProxyPac::boxed(embeddable.clone(), dapps_domain.to_owned()));
-	pages.insert(WEB_PATH.into(), Web::boxed(embeddable.clone(), web_proxy_tokens.clone(), fetch.clone()));
+	pages.insert(WEB_PATH.into(), Web::boxed(embeddable.clone(), web_proxy_tokens.clone(), fetch.clone(), pool.clone()));
 
 	(local_endpoints, pages)
 }

dapps/src/handlers/fetch.rs

@@ -23,6 +23,7 @@ use std::time::{Instant, Duration};
 use fetch::{self, Fetch};
 use futures::sync::oneshot;
 use futures::{self, Future};
+use futures_cpupool::CpuPool;
 use hyper::{self, Method, StatusCode};
 use parking_lot::Mutex;
 
@@ -35,7 +36,7 @@ const FETCH_TIMEOUT: u64 = 300;
 
 pub enum ValidatorResponse {
 	Local(local::Dapp),
-	Streaming(StreamingHandler<fetch::Response>),
+	Streaming(StreamingHandler<fetch::BodyReader>),
 }
 
 pub trait ContentValidator: Sized + Send + 'static {
@@ -252,6 +253,7 @@ impl ContentFetcherHandler {
 		installer: H,
 		embeddable_on: Embeddable,
 		fetch: F,
+		pool: CpuPool,
 	) -> Self {
 		let fetch_control = FetchControl::default();
 		let errors = Errors { embeddable_on };
@@ -262,6 +264,7 @@ impl ContentFetcherHandler {
 			Method::Get => {
 				trace!(target: "dapps", "Fetching content from: {:?}", url);
 				FetchState::InProgress(Self::fetch_content(
+						pool,
 						fetch,
 						url,
 						fetch_control.abort.clone(),
@@ -282,6 +285,7 @@ impl ContentFetcherHandler {
 	}
 
 	fn fetch_content<H: ContentValidator, F: Fetch>(
+		pool: CpuPool,
 		fetch: F,
 		url: &str,
 		abort: Arc<AtomicBool>,
@@ -290,8 +294,8 @@ impl ContentFetcherHandler {
 		installer: H,
 	) -> Box<Future<Item=FetchState, Error=()> + Send> {
 		// Start fetching the content
-		let fetch2 = fetch.clone();
-		let future = fetch.fetch_with_abort(url, abort.into()).then(move |result| {
+		let pool2 = pool.clone();
+		let future = fetch.fetch(url, abort.into()).then(move |result| {
 			trace!(target: "dapps", "Fetching content finished. Starting validation: {:?}", result);
 			Ok(match result {
 				Ok(response) => match installer.validate_and_install(response) {
@@ -303,7 +307,7 @@ impl ContentFetcherHandler {
 					Ok(ValidatorResponse::Streaming(stream)) => {
 						trace!(target: "dapps", "Validation OK. Streaming response.");
 						let (reading, response) = stream.into_response();
-						fetch2.process_and_forget(reading);
+						pool.spawn(reading).forget();
 						FetchState::Streaming(response)
 					},
 					Err(e) => {
@@ -319,7 +323,7 @@ impl ContentFetcherHandler {
 		});
 
 		// make sure to run within fetch thread pool.
-		fetch.process(future)
+		Box::new(pool2.spawn(future))
 	}
 }
 

dapps/src/lib.rs

@@ -41,6 +41,7 @@ extern crate parity_hash_fetch as hash_fetch;
 extern crate parity_ui;
 extern crate keccak_hash as hash;
 extern crate parity_version;
+extern crate registrar;
 
 #[macro_use]
 extern crate futures;
@@ -80,7 +81,7 @@ use parking_lot::RwLock;
 use fetch::Fetch;
 use node_health::NodeHealth;
 
-pub use hash_fetch::urlhint::ContractClient;
+pub use registrar::{RegistrarClient, Asynchronous};
 pub use node_health::SyncStatus;
 
 
@@ -155,7 +156,7 @@ impl Middleware {
 		pool: CpuPool,
 		health: NodeHealth,
 		dapps_domain: &str,
-		registrar: Arc<ContractClient>,
+		registrar: Arc<RegistrarClient<Call=Asynchronous>>,
 		sync_status: Arc<SyncStatus>,
 		fetch: F,
 	) -> Self {
@@ -198,7 +199,7 @@ impl Middleware {
 		dapps_path: PathBuf,
 		extra_dapps: Vec<PathBuf>,
 		dapps_domain: &str,
-		registrar: Arc<ContractClient>,
+		registrar: Arc<RegistrarClient<Call=Asynchronous>>,
 		sync_status: Arc<SyncStatus>,
 		web_proxy_tokens: Arc<WebProxyTokens>,
 		fetch: F,

dapps/src/tests/helpers/fetch.rs

@@ -14,12 +14,13 @@
 // You should have received a copy of the GNU General Public License
 // along with Parity.  If not, see <http://www.gnu.org/licenses/>.
 
-use std::{io, thread, time};
+use std::{thread, time};
 use std::sync::{atomic, mpsc, Arc};
 use parking_lot::Mutex;
+use hyper;
 
 use futures::{self, Future};
-use fetch::{self, Fetch};
+use fetch::{self, Fetch, Url};
 
 pub struct FetchControl {
 	sender: mpsc::Sender<()>,
@@ -96,11 +97,8 @@ impl FakeFetch {
 impl Fetch for FakeFetch {
 	type Result = Box<Future<Item = fetch::Response, Error = fetch::Error> + Send>;
 
-	fn new() -> Result<Self, fetch::Error> where Self: Sized {
-		Ok(FakeFetch::default())
-	}
-
-	fn fetch_with_abort(&self, url: &str, _abort: fetch::Abort) -> Self::Result {
+	fn fetch(&self, url: &str, abort: fetch::Abort) -> Self::Result {
+		let u = Url::parse(url).unwrap();
 		self.requested.lock().push(url.into());
 		let manual = self.manual.clone();
 		let response = self.response.clone();
@@ -111,23 +109,10 @@ impl Fetch for FakeFetch {
 				// wait for manual resume
 				let _ = rx.recv();
 			}
-
 			let data = response.lock().take().unwrap_or(b"Some content");
-			let cursor = io::Cursor::new(data);
-			tx.send(fetch::Response::from_reader(cursor)).unwrap();
+			tx.send(fetch::Response::new(u, hyper::Response::new().with_body(data), abort)).unwrap();
 		});
 
 		Box::new(rx.map_err(|_| fetch::Error::Aborted))
 	}
-
-	fn process_and_forget<F, I, E>(&self, f: F) where
-		F: Future<Item=I, Error=E> + Send + 'static,
-		I: Send + 'static,
-		E: Send + 'static,
-	{
-		// Spawn the task in a separate thread.
-		thread::spawn(|| {
-			let _ = f.wait();
-		});
-	}
 }

dapps/src/tests/helpers/mod.rs

@@ -24,7 +24,7 @@ use jsonrpc_http_server::{self as http, Host, DomainsValidation};
 use parity_reactor::Remote;
 
 use devtools::http_client;
-use hash_fetch::urlhint::ContractClient;
+use registrar::{RegistrarClient, Asynchronous};
 use fetch::{Fetch, Client as FetchClient};
 use node_health::{NodeHealth, TimeChecker, CpuPool};
 
@@ -63,7 +63,7 @@ pub fn init_server<F, B>(process: F, io: IoHandler) -> (Server, Arc<FakeRegistra
 	let mut dapps_path = env::temp_dir();
 	dapps_path.push("non-existent-dir-to-prevent-fs-files-from-loading");
 
-	let mut builder = ServerBuilder::new(&dapps_path, registrar.clone());
+	let mut builder = ServerBuilder::new(FetchClient::new().unwrap(), &dapps_path, registrar.clone());
 	builder.signer_address = Some(("127.0.0.1".into(), SIGNER_PORT));
 	let server = process(builder).start_unsecured_http(&"127.0.0.1:0".parse().unwrap(), io).unwrap();
 	(
@@ -144,26 +144,26 @@ pub fn assert_security_headers_for_embed(headers: &[String]) {
 /// Webapps HTTP+RPC server build.
 pub struct ServerBuilder<T: Fetch = FetchClient> {
 	dapps_path: PathBuf,
-	registrar: Arc<ContractClient>,
+	registrar: Arc<RegistrarClient<Call=Asynchronous>>,
 	sync_status: Arc<SyncStatus>,
 	web_proxy_tokens: Arc<WebProxyTokens>,
 	signer_address: Option<(String, u16)>,
 	allowed_hosts: DomainsValidation<Host>,
-	fetch: Option<T>,
+	fetch: T,
 	serve_ui: bool,
 }
 
 impl ServerBuilder {
 	/// Construct new dapps server
-	pub fn new<P: AsRef<Path>>(dapps_path: P, registrar: Arc<ContractClient>) -> Self {
+	pub fn new<P: AsRef<Path>>(fetch: FetchClient, dapps_path: P, registrar: Arc<RegistrarClient<Call=Asynchronous>>) -> Self {
 		ServerBuilder {
 			dapps_path: dapps_path.as_ref().to_owned(),
 			registrar: registrar,
 			sync_status: Arc::new(FakeSync(false)),
 			web_proxy_tokens: Arc::new(|_| None),
 			signer_address: None,
 			allowed_hosts: DomainsValidation::Disabled,
-			fetch: None,
+			fetch: fetch,
 			serve_ui: false,
 		}
 	}
@@ -179,15 +179,14 @@ impl<T: Fetch> ServerBuilder<T> {
 			web_proxy_tokens: self.web_proxy_tokens,
 			signer_address: self.signer_address,
 			allowed_hosts: self.allowed_hosts,
-			fetch: Some(fetch),
+			fetch: fetch,
 			serve_ui: self.serve_ui,
 		}
 	}
 
 	/// Asynchronously start server with no authentication,
 	/// returns result with `Server` handle on success or an error.
 	pub fn start_unsecured_http(self, addr: &SocketAddr, io: IoHandler) -> io::Result<Server> {
-		let fetch = self.fetch_client();
 		Server::start_http(
 			addr,
 			io,
@@ -199,17 +198,10 @@ impl<T: Fetch> ServerBuilder<T> {
 			self.sync_status,
 			self.web_proxy_tokens,
 			Remote::new_sync(),
-			fetch,
+			self.fetch,
 			self.serve_ui,
 		)
 	}
-
-	fn fetch_client(&self) -> T {
-		match self.fetch.clone() {
-			Some(fetch) => fetch,
-			None => T::new().unwrap(),
-		}
-	}
 }
 
 const DAPPS_DOMAIN: &'static str = "web3.site";
@@ -227,7 +219,7 @@ impl Server {
 		signer_address: Option<(String, u16)>,
 		dapps_path: PathBuf,
 		extra_dapps: Vec<PathBuf>,
-		registrar: Arc<ContractClient>,
+		registrar: Arc<RegistrarClient<Call=Asynchronous>>,
 		sync_status: Arc<SyncStatus>,
 		web_proxy_tokens: Arc<WebProxyTokens>,
 		remote: Remote,

dapps/src/tests/helpers/registrar.rs

@@ -18,10 +18,9 @@ use std::str;
 use std::sync::Arc;
 use std::collections::HashMap;
 
-use futures::Future;
 use ethereum_types::{H256, Address};
 use bytes::{Bytes, ToPretty};
-use hash_fetch::urlhint::ContractClient;
+use registrar::{RegistrarClient, Asynchronous};
 use parking_lot::Mutex;
 use rustc_hex::FromHex;
 
@@ -62,12 +61,14 @@ impl FakeRegistrar {
 	}
 }
 
-impl ContractClient for FakeRegistrar {
-	fn registrar(&self) -> Result<Address, String> {
+impl RegistrarClient for FakeRegistrar {
+	type Call = Asynchronous;
+
+	fn registrar_address(&self) -> Result<Address, String> {
 		Ok(REGISTRAR.parse().unwrap())
 	}
 
-	fn call(&self, address: Address, data: Bytes) -> Box<Future<Item = Bytes, Error = String> + Send> {
+	fn call_contract(&self, address: Address, data: Bytes) -> Self::Call {
 		let call = (address.to_hex(), data.to_hex());
 		self.calls.lock().push(call.clone());
 		let res = self.responses.lock().get(&call).cloned().expect(&format!("No response for call: {:?}", call));