RPC: Pending Extrinsics (#563)

* Expose extrinsics in pool.

* Add test.

* Use latest transaction pool.

* Fix compilation.

substrate/substrate/extrinsic-pool/src/api.rs

@@ -16,6 +16,9 @@
 
 //! External API for extrinsic pool.
 
+use std::fmt::Debug;
+
+use serde::{Serialize, de::DeserializeOwned};
 use txpool;
 use futures::sync::mpsc;
 
@@ -43,6 +46,9 @@ pub trait ExtrinsicPool<Ex, BlockId, Hash>: Send + Sync + 'static {
 	/// Error type
 	type Error: Error;
 
+	/// Pooled extrinsics
+	type InPool: Debug + Serialize + DeserializeOwned + Send + Sync + 'static;
+
 	/// Submit a collection of extrinsics to the pool.
 	fn submit(&self, block: BlockId, xt: Vec<Ex>) -> Result<Vec<Hash>, Self::Error>;
 
@@ -54,4 +60,7 @@ pub trait ExtrinsicPool<Ex, BlockId, Hash>: Send + Sync + 'static {
 
 	/// Return an event stream of transactions imported to the pool.
 	fn import_notification_stream(&self) -> EventStream;
+
+	/// Return all extrinsics in the pool aggregated by the sender.
+	fn all(&self) -> Self::InPool;
 }

substrate/substrate/extrinsic-pool/src/pool.rs

@@ -149,4 +149,19 @@ impl<Hash, VEx, S, E> Pool<Hash, VEx, S, E> where
 	{
 		f(self.pool.read().pending(ready))
 	}
+
+	/// Retrieve all transactions in the pool. The transactions might be unordered.
+	pub fn all<F, T>(&self, f: F) -> T where
+		F: FnOnce(txpool::UnorderedIterator<VEx, AlwaysReady, S>) -> T,
+	{
+		f(self.pool.read().unordered_pending(AlwaysReady))
+	}
+}
+
+/// A Readiness implementation that returns `Ready` for all transactions.
+pub struct AlwaysReady;
+impl<VEx> txpool::Ready<VEx> for AlwaysReady {
+	fn is_ready(&mut self, _tx: &VEx) -> txpool::Readiness {
+		txpool::Readiness::Ready
+	}
 }

substrate/substrate/rpc-servers/Cargo.toml

@@ -9,5 +9,6 @@ jsonrpc-http-server = { git = "https://github.com/paritytech/jsonrpc.git" }
 jsonrpc-pubsub = { git = "https://github.com/paritytech/jsonrpc.git" }
 jsonrpc-ws-server = { git = "https://github.com/paritytech/jsonrpc.git" }
 log = "0.3"
+serde = "1.0"
 substrate-rpc = { path = "../rpc", version = "0.1" }
 substrate-runtime-primitives = { path = "../runtime/primitives" }

substrate/substrate/rpc-servers/src/lib.rs

@@ -24,6 +24,7 @@ extern crate jsonrpc_core as rpc;
 extern crate jsonrpc_http_server as http;
 extern crate jsonrpc_pubsub as pubsub;
 extern crate jsonrpc_ws_server as ws;
+extern crate serde;
 extern crate substrate_runtime_primitives;
 
 #[macro_use]
@@ -38,16 +39,17 @@ pub type HttpServer = http::Server;
 pub type WsServer = ws::Server;
 
 /// Construct rpc `IoHandler`
-pub fn rpc_handler<Block: BlockT, S, C, A, Y>(
+pub fn rpc_handler<Block, PendingExtrinsics, S, C, A, Y>(
 	state: S,
 	chain: C,
 	author: A,
 	system: Y,
 ) -> RpcHandler where
-	Block: 'static,
+	Block: BlockT + 'static,
+	PendingExtrinsics: serde::Serialize + serde::de::DeserializeOwned + Send + Sync + 'static,
 	S: apis::state::StateApi<Block::Hash, Metadata=Metadata>,
 	C: apis::chain::ChainApi<Block::Hash, Block::Header, Metadata=Metadata>,
-	A: apis::author::AuthorApi<Block::Hash, Block::Extrinsic, Metadata=Metadata>,
+	A: apis::author::AuthorApi<Block::Hash, Block::Extrinsic, PendingExtrinsics, Metadata=Metadata>,
 	Y: apis::system::SystemApi,
 {
 	let mut io = pubsub::PubSubHandler::default();

substrate/substrate/rpc/src/author/mod.rs

@@ -41,7 +41,7 @@ use self::error::Result;
 
 build_rpc_trait! {
 	/// Substrate authoring RPC API
-	pub trait AuthorApi<Hash, Extrinsic> {
+	pub trait AuthorApi<Hash, Extrinsic, PendingExtrinsics> {
 		type Metadata;
 
 		/// Submit extrinsic for inclusion in block.
@@ -51,6 +51,10 @@ build_rpc_trait! {
 		#[rpc(name = "author_submitExtrinsic")]
 		fn submit_extrinsic(&self, Bytes) -> Result<Hash>;
 
+		/// Returns all pending extrinsics, potentially grouped by sender.
+		#[rpc(name = "author_pendingExtrinsics")]
+		fn pending_extrinsics(&self) -> Result<PendingExtrinsics>;
+
 		#[pubsub(name = "author_extrinsicUpdate")] {
 			/// Submit an extrinsic to watch.
 			#[rpc(name = "author_submitAndWatchExtrinsic")]
@@ -60,7 +64,6 @@ build_rpc_trait! {
 			#[rpc(name = "author_unwatchExtrinsic")]
 			fn unwatch_extrinsic(&self, SubscriptionId) -> Result<bool>;
 		}
-
 	}
 }
 
@@ -85,12 +88,13 @@ impl<B, E, Block: traits::Block, P> Author<B, E, Block, P> {
 	}
 }
 
-impl<B, E, Block, P, Ex, Hash> AuthorApi<Hash, Ex> for Author<B, E, Block, P> where
+impl<B, E, Block, P, Ex, Hash, InPool> AuthorApi<Hash, Ex, InPool> for Author<B, E, Block, P> where
 	B: client::backend::Backend<Block> + Send + Sync + 'static,
 	E: client::CallExecutor<Block> + Send + Sync + 'static,
 	Block: traits::Block + 'static,
-	Hash: traits::MaybeSerializeDebug + Sync + Send + 'static,
-	P: ExtrinsicPool<Ex, generic::BlockId<Block>, Hash>,
+	Hash: traits::MaybeSerializeDebug + Send + Sync + 'static,
+	InPool: traits::MaybeSerializeDebug + Send + Sync + 'static,
+	P: ExtrinsicPool<Ex, generic::BlockId<Block>, Hash, InPool=InPool>,
 	P::Error: 'static,
 	Ex: Codec,
 {
@@ -112,6 +116,10 @@ impl<B, E, Block, P, Ex, Hash> AuthorApi<Hash, Ex> for Author<B, E, Block, P> wh
 			)
 	}
 
+	fn pending_extrinsics(&self) -> Result<InPool> {
+		Ok(self.pool.all())
+	}
+
 	fn watch_extrinsic(&self, _metadata: Self::Metadata, subscriber: pubsub::Subscriber<Status<Hash>>, xt: Bytes) {
 
 		let submit = || -> Result<_> {
@@ -146,3 +154,4 @@ impl<B, E, Block, P, Ex, Hash> AuthorApi<Hash, Ex> for Author<B, E, Block, P> wh
 		Ok(self.subscriptions.cancel(id))
 	}
 }
+

substrate/substrate/rpc/src/author/tests.rs

@@ -46,6 +46,7 @@ impl fmt::Display for Error {
 
 impl<BlockHash> api::ExtrinsicPool<Extrinsic, BlockHash, u64> for DummyTxPool {
 	type Error = Error;
+	type InPool = Vec<u8>;
 
 	/// Submit extrinsic for inclusion in block.
 	fn submit(&self, _block: BlockHash, xt: Vec<Extrinsic>) -> Result<Vec<Hash>, Self::Error> {
@@ -79,6 +80,10 @@ impl<BlockHash> api::ExtrinsicPool<Extrinsic, BlockHash, u64> for DummyTxPool {
 	fn import_notification_stream(&self) -> api::EventStream {
 		unreachable!()
 	}
+
+	fn all(&self) -> Self::InPool {
+		vec![1, 2, 3, 4, 5]
+	}
 }
 
 #[test]
@@ -143,3 +148,18 @@ fn should_watch_extrinsic() {
 		Some(r#"{"jsonrpc":"2.0","method":"test","params":{"result":{"usurped":5},"subscription":0}}"#.into())
 	);
 }
+
+#[test]
+fn should_return_pending_extrinsics() {
+	let runtime = runtime::Runtime::new().unwrap();
+	let p = Author {
+		client: Arc::new(test_client::new()),
+		pool: Arc::new(DummyTxPool::default()),
+		subscriptions: Subscriptions::new(runtime.executor()),
+	};
+
+	assert_matches!(
+		p.pending_extrinsics(),
+		Ok(ref expected) if expected == &[1u8, 2, 3, 4, 5]
+	);
+}

substrate/substrate/service/src/lib.rs

@@ -210,7 +210,8 @@ impl<Components> Service<Components>
 				let chain = rpc::apis::chain::Chain::new(client.clone(), task_executor.clone());
 				let state = rpc::apis::state::State::new(client.clone(), task_executor.clone());
 				let author = rpc::apis::author::Author::new(client.clone(), extrinsic_pool.api(), task_executor.clone());
-				rpc::rpc_handler::<ComponentBlock<Components>, _, _, _, _>(
+
+				rpc::rpc_handler::<ComponentBlock<Components>, _, _, _, _, _>(
 					state,
 					chain,
 					author,