blockchainevents

client/api/src/client.rs

@@ -18,16 +18,16 @@
 
 //! A set of APIs supported by the client along with their primitives.
 
+use crate::{blockchain::Info, notifications::StorageEventStream, FinalizeSummary, ImportSummary};
+use futures::Stream;
 use sp_consensus::BlockOrigin;
 use sp_core::storage::StorageKey;
 use sp_runtime::{
-	generic::{BlockId, SignedBlock},
+	generic::{Block, BlockId, SignedBlock},
 	traits::{Block as BlockT, NumberFor},
 	Justifications,
 };
-use std::{collections::HashSet, convert::TryFrom, fmt, sync::Arc};
-
-use crate::{blockchain::Info, notifications::StorageEventStream, FinalizeSummary, ImportSummary};
+use std::{collections::HashSet, convert::TryFrom, fmt, pin::Pin, sync::Arc};
 
 use sc_transaction_pool_api::ChainEvent;
 use sc_utils::mpsc::TracingUnboundedReceiver;
@@ -57,6 +57,13 @@ pub trait BlockOf {
 	type Type: BlockT;
 }
 
+pub trait BlockchainRPCEvents<Block: BlockT> {
+	fn finality_notification_stream(&self) -> Pin<Box<dyn Stream<Item = Block::Header> + Send>>;
+
+	fn import_notification_stream(&self) -> Pin<Box<dyn Stream<Item = Block::Header> + Send>>;
+
+	fn best_head_stream(&self) -> Pin<Box<dyn Stream<Item = Block::Header> + Send>>;
+}
 /// A source of blockchain events.
 pub trait BlockchainEvents<Block: BlockT> {
 	/// Get block import event stream. Not guaranteed to be fired for every

client/service/src/builder.rs

@@ -17,7 +17,7 @@
 // along with this program. If not, see <https://www.gnu.org/licenses/>.
 
 use crate::{
-	build_network_future,
+	build_network_collator_future, build_network_future,
 	client::{Client, ClientConfig},
 	config::{Configuration, KeystoreConfig, PrometheusConfig, TransactionStorageMode},
 	error::Error,
@@ -31,7 +31,8 @@ use prometheus_endpoint::Registry;
 use sc_chain_spec::get_extension;
 use sc_client_api::{
 	execution_extensions::ExecutionExtensions, proof_provider::ProofProvider, BadBlocks,
-	BlockBackend, BlockchainEvents, ExecutorProvider, ForkBlocks, StorageProvider, UsageProvider,
+	BlockBackend, BlockchainEvents, BlockchainRPCEvents, ExecutorProvider, ForkBlocks,
+	StorageProvider, UsageProvider,
 };
 use sc_client_db::{Backend, DatabaseSettings};
 use sc_consensus::import_queue::ImportQueue;
@@ -745,7 +746,7 @@ pub struct BuildCollatorNetworkParams<'a, TImpQu, TCl> {
 }
 
 /// Build the network service, the network status sinks and an RPC sender.
-pub fn build_collator_network<TBl, TExPool, TImpQu, TCl>(
+pub fn build_collator_network<TBl, TImpQu, TCl>(
 	params: BuildCollatorNetworkParams<TImpQu, TCl>,
 ) -> Result<
 	(
@@ -763,9 +764,8 @@ where
 		+ BlockIdTo<TBl, Error = sp_blockchain::Error>
 		+ ProofProvider<TBl>
 		+ HeaderBackend<TBl>
-		+ BlockchainEvents<TBl>
+		+ BlockchainRPCEvents<TBl>
 		+ 'static,
-	TExPool: MaintainedTransactionPool<Block = TBl, Hash = <TBl as BlockT>::Hash> + 'static,
 	TImpQu: ImportQueue<TBl> + 'static,
 {
 	let BuildCollatorNetworkParams { config, client, spawn_handle, import_queue } = params;
@@ -873,7 +873,7 @@ where
 
 	let (system_rpc_tx, system_rpc_rx) = tracing_unbounded("mpsc_system_rpc");
 
-	let future = build_network_future(
+	let future = build_network_collator_future(
 		config.role.clone(),
 		network_mut,
 		client,

client/service/src/lib.rs

@@ -64,7 +64,7 @@ pub use sc_chain_spec::{
 	ChainSpec, ChainType, Extension as ChainSpecExtension, GenericChainSpec, NoExtension,
 	Properties, RuntimeGenesis,
 };
-use sc_client_api::{blockchain::HeaderBackend, BlockchainEvents};
+use sc_client_api::{blockchain::HeaderBackend, BlockchainEvents, BlockchainRPCEvents};
 pub use sc_consensus::ImportQueue;
 pub use sc_executor::NativeExecutionDispatch;
 #[doc(hidden)]
@@ -273,6 +273,157 @@ async fn build_network_future<
 	}
 }
 
+/// Builds a never-ending future that continuously polls the network.
+///
+/// The `status_sink` contain a list of senders to send a periodic network status to.
+async fn build_network_collator_future<
+	B: BlockT,
+	C: BlockchainRPCEvents<B> + HeaderBackend<B>,
+	H: sc_network::ExHashT,
+>(
+	role: Role,
+	mut network: sc_network::NetworkWorker<B, H>,
+	client: Arc<C>,
+	mut rpc_rx: TracingUnboundedReceiver<sc_rpc::system::Request<B>>,
+	should_have_peers: bool,
+	announce_imported_blocks: bool,
+) {
+	let mut imported_blocks_stream = client.import_notification_stream().fuse();
+
+	// Current best block at initialization, to report to the RPC layer.
+	let starting_block = client.info().best_number;
+
+	// Stream of finalized blocks reported by the client.
+	let mut finality_notification_stream = client.finality_notification_stream().fuse();
+	let mut new_best_notification_stream = client.best_head_stream().fuse();
+
+	loop {
+		futures::select! {
+			// List of blocks that the client has imported.
+			notification = imported_blocks_stream.next() => {
+				let notification = match notification {
+					Some(n) => n,
+					// If this stream is shut down, that means the client has shut down, and the
+					// most appropriate thing to do for the network future is to shut down too.
+					None => return,
+				};
+
+				// TODO skunert No need to announce anything
+				if announce_imported_blocks {
+					network.service().announce_block(notification.hash(), None);
+				}
+			}
+
+			notification = new_best_notification_stream.next() => {
+				let notification = match notification {
+					Some(n) => n,
+					// If this stream is shut down, that means the client has shut down, and the
+					// most appropriate thing to do for the network future is to shut down too.
+					None => return,
+				};
+
+					network.service().new_best_block_imported(
+						notification.hash(),
+						notification.number().clone(),
+					);
+			}
+
+			// List of blocks that the client has finalized.
+			notification = finality_notification_stream.select_next_some() => {
+				network.on_block_finalized(notification.hash(), notification);
+			}
+
+			// Answer incoming RPC requests.
+			request = rpc_rx.select_next_some() => {
+				match request {
+					sc_rpc::system::Request::Health(sender) => {
+						let _ = sender.send(sc_rpc::system::Health {
+							peers: network.peers_debug_info().len(),
+							is_syncing: network.service().is_major_syncing(),
+							should_have_peers,
+						});
+					},
+					sc_rpc::system::Request::LocalPeerId(sender) => {
+						let _ = sender.send(network.local_peer_id().to_base58());
+					},
+					sc_rpc::system::Request::LocalListenAddresses(sender) => {
+						let peer_id = network.local_peer_id().clone().into();
+						let p2p_proto_suffix = sc_network::multiaddr::Protocol::P2p(peer_id);
+						let addresses = network.listen_addresses()
+							.map(|addr| addr.clone().with(p2p_proto_suffix.clone()).to_string())
+							.collect();
+						let _ = sender.send(addresses);
+					},
+					sc_rpc::system::Request::Peers(sender) => {
+						let _ = sender.send(network.peers_debug_info().into_iter().map(|(peer_id, p)|
+							sc_rpc::system::PeerInfo {
+								peer_id: peer_id.to_base58(),
+								roles: format!("{:?}", p.roles),
+								best_hash: p.best_hash,
+								best_number: p.best_number,
+							}
+						).collect());
+					}
+					sc_rpc::system::Request::NetworkState(sender) => {
+						if let Some(network_state) = serde_json::to_value(&network.network_state()).ok() {
+							let _ = sender.send(network_state);
+						}
+					}
+					sc_rpc::system::Request::NetworkAddReservedPeer(peer_addr, sender) => {
+						let x = network.add_reserved_peer(peer_addr)
+							.map_err(sc_rpc::system::error::Error::MalformattedPeerArg);
+						let _ = sender.send(x);
+					}
+					sc_rpc::system::Request::NetworkRemoveReservedPeer(peer_id, sender) => {
+						let _ = match peer_id.parse::<PeerId>() {
+							Ok(peer_id) => {
+								network.remove_reserved_peer(peer_id);
+								sender.send(Ok(()))
+							}
+							Err(e) => sender.send(Err(sc_rpc::system::error::Error::MalformattedPeerArg(
+								e.to_string(),
+							))),
+						};
+					}
+					sc_rpc::system::Request::NetworkReservedPeers(sender) => {
+						let reserved_peers = network.reserved_peers();
+						let reserved_peers = reserved_peers
+							.map(|peer_id| peer_id.to_base58())
+							.collect();
+
+						let _ = sender.send(reserved_peers);
+					}
+					sc_rpc::system::Request::NodeRoles(sender) => {
+						use sc_rpc::system::NodeRole;
+
+						let node_role = match role {
+							Role::Authority { .. } => NodeRole::Authority,
+							Role::Light => NodeRole::LightClient,
+							Role::Full => NodeRole::Full,
+						};
+
+						let _ = sender.send(vec![node_role]);
+					}
+					sc_rpc::system::Request::SyncState(sender) => {
+						use sc_rpc::system::SyncState;
+
+						let _ = sender.send(SyncState {
+							starting_block: starting_block,
+							current_block: client.info().best_number,
+							highest_block: network.best_seen_block(),
+						});
+					}
+				}
+			}
+
+			// The network worker has done something. Nothing special to do, but could be
+			// used in the future to perform actions in response of things that happened on
+			// the network.
+			_ = (&mut network).fuse() => {}
+		}
+	}
+}
+
 // Wrapper for HTTP and WS servers that makes sure they are properly shut down.
 mod waiting {
 	pub struct HttpServer(pub Option<sc_rpc_server::HttpServer>);