Extrinsics PubSub (#349)

* Extrinsic subscriptions.

* Handle RPC errors better.

* Add tests for extrinsics and unignored others.

* Handle client errors.

* Fix compilation.

polkadot/cli/src/lib.rs

@@ -221,7 +221,7 @@ pub fn run<I, T, W>(args: I, worker: W) -> error::Result<()> where
 			info!("Starting collator");
 			// TODO [rob]: collation node implementation
 			// This isn't a thing. Different parachains will have their own collator executables and
-			// maybe link to libpolkadot to get a light-client. 
+			// maybe link to libpolkadot to get a light-client.
 			service::Roles::LIGHT
 		} else if matches.is_present("light") {
 			info!("Starting (light)");
@@ -478,9 +478,9 @@ fn run_until_exit<C, W>(
 		let ws_address = parse_address("127.0.0.1:9944", "ws-port", matches)?;
 
 		let handler = || {
-			let client = (&service as &substrate_service::Service<C>).client();
+			let client = substrate_service::Service::client(&service);
 			let chain = rpc::apis::chain::Chain::new(client.clone(), executor.clone());
-			let author = rpc::apis::author::Author::new(client.clone(), service.extrinsic_pool());
+			let author = rpc::apis::author::Author::new(client.clone(), service.extrinsic_pool(), executor.clone());
 			rpc::rpc_handler::<service::ComponentBlock<C>, _, _, _, _>(
 				client,
 				chain,

polkadot/transaction-pool/src/lib.rs

@@ -44,8 +44,13 @@ use std::{
 };
 
 use codec::{Decode, Encode};
-use extrinsic_pool::{Pool, Listener, txpool::{self, Readiness, scoring::{Change, Choice}}};
+use extrinsic_pool::{
-use extrinsic_pool::api::{ExtrinsicPool, EventStream};
+	api::{ExtrinsicPool, EventStream},
+	txpool::{self, Readiness, scoring::{Change, Choice}},
+	watcher::Watcher,
+	Pool,
+	Listener,
+};
 use polkadot_api::PolkadotApi;
 use primitives::{AccountId, BlockId, Hash, Index, UncheckedExtrinsic as FutureProofUncheckedExtrinsic};
 use runtime::{Address, UncheckedExtrinsic};
@@ -385,6 +390,8 @@ impl<A> Deref for TransactionPool<A> {
 	}
 }
 
+// TODO: more general transaction pool, which can handle more kinds of vec-encoded transactions,
+// even when runtime is out of date.
 impl<A> ExtrinsicPool<FutureProofUncheckedExtrinsic, BlockId, Hash> for TransactionPool<A> where
 	A: Send + Sync + 'static,
 	A: PolkadotApi,
@@ -392,8 +399,6 @@ impl<A> ExtrinsicPool<FutureProofUncheckedExtrinsic, BlockId, Hash> for Transact
 	type Error = Error;
 
 	fn submit(&self, block: BlockId, xts: Vec<FutureProofUncheckedExtrinsic>) -> Result<Vec<Hash>> {
-		// TODO: more general transaction pool, which can handle more kinds of vec-encoded transactions,
-		// even when runtime is out of date.
 		xts.into_iter()
 			.map(|xt| xt.encode())
 			.map(|encoded| {
@@ -404,6 +409,18 @@ impl<A> ExtrinsicPool<FutureProofUncheckedExtrinsic, BlockId, Hash> for Transact
 			.collect()
 	}
 
+	fn submit_and_watch(&self, block: BlockId, xt: FutureProofUncheckedExtrinsic) -> Result<Watcher<Hash>> {
+		let encoded = xt.encode();
+		let decoded = UncheckedExtrinsic::decode(&mut &encoded[..]).ok_or(ErrorKind::InvalidExtrinsicFormat)?;
+
+		let verifier = Verifier {
+			api: &*self.api,
+			at_block: block,
+		};
+
+		self.inner.submit_and_watch(verifier, decoded)
+	}
+
 	fn light_status(&self) -> LightStatus {
 		self.inner.light_status()
 	}