mirror of
https://github.com/pezkuwichain/pezkuwi-subxt.git
synced 2026-06-29 03:17:25 +00:00
Service factory refactor (#3382)
* Move Service::new to a macro * Move function calls to macros * Extract offchain_workers and start_rpc in separate function In follow-up commits, we want to be able to directly call maintain_transaction_pool, offchain_workers, and start_rpc, without having to implement the Components trait. This commit is a preliminary step: we extract the code to freestanding functions. * Introduce an AbstractService trait * Introduce NewService as an implementation detail of Service * Implement traits on NewService instead Instead of implementing AbstractService, Future, and Executor on Service, we implement them on NewService instead. The implementations of AbstractService, Future, and Executor on Service still exist, but they just wrap to the respective implementations for NewService. * Move components creation back to macro invocation Instead of having multiple $build_ parameters passed to the macro, let's group them all into one. This change is necessary for the follow-up commits, because we are going to call new_impl! only after all the components have already been built. * Add a $block parameter to new_impl This makes it possible to be explicit as what the generic parameter of the NewServiceis, without relying on type inference. * Introduce the ServiceBuilder struct Introduces a new builder-like ServiceBuilder struct that creates a NewService. * Macro-ify import_blocks, export_blocks and revert_chain Similar to the introduction of new_impl!, we extract the actual code into a macro, letting us get rid of the Components and Factory traits * Add export_blocks, import_blocks and revert_chain methods on ServiceBuilder Can be used as a replacement for the chain_ops::* methods * Add run_with_builder Instead of just run, adds run_with_builder to ParseAndPrepareExport/Import/Revert. This lets you run these operations with a ServiceBuilder instead of a ServiceFactory. * Transition node and node-template to ServiceBuilder * Transition transaction-factory to the new service factory This is technically a breaking change, but the transaction-factory crate is only ever used from within substrate-node, which this commit updates as well. * Remove old service factory * Adjust the AbstractService trait to be more usable We slightly change the trait bounds in order to make all the methods usable. * Make substrate-service-test compile * Fix the node-cli tests * Remove the old API * Remove the components module * Fix indentation on chain_ops * Line widths * Fix bad line widths commit * Line widths again 🤦 * Fix the sync test * Apply suggestions from code review Co-Authored-By: Gavin Wood <i@gavwood.com> * Address some concerns * Remove TelemetryOnConnect * Remove informant::start * Update jsonrpc * Rename factory to builder * Line widths 😩
This commit is contained in:
Pierre Krieger
committed by
Bastian Köcher
Bastian Köcher
parent
144bd228af
commit
5b8ebf7baf
+218
-397
@@ -19,51 +19,43 @@
|
||||
|
||||
#![warn(missing_docs)]
|
||||
|
||||
mod components;
|
||||
mod chain_spec;
|
||||
pub mod config;
|
||||
#[macro_use]
|
||||
pub mod chain_ops;
|
||||
pub mod error;
|
||||
|
||||
use std::io;
|
||||
use std::marker::PhantomData;
|
||||
use std::net::SocketAddr;
|
||||
use std::collections::HashMap;
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
use std::time::{Duration, Instant};
|
||||
use serde::{Serialize, de::DeserializeOwned};
|
||||
use futures::sync::mpsc;
|
||||
use parking_lot::Mutex;
|
||||
|
||||
use client::{BlockchainEvents, backend::Backend, runtime_api::BlockT};
|
||||
use client::{runtime_api::BlockT, Client};
|
||||
use exit_future::Signal;
|
||||
use futures::prelude::*;
|
||||
use futures03::stream::{StreamExt as _, TryStreamExt as _};
|
||||
use keystore::Store as Keystore;
|
||||
use network::{NetworkState, NetworkStateInfo};
|
||||
use log::{log, info, warn, debug, error, Level};
|
||||
use network::{NetworkService, NetworkState, specialization::NetworkSpecialization};
|
||||
use log::{log, warn, debug, error, Level};
|
||||
use codec::{Encode, Decode};
|
||||
use primitives::{Blake2Hasher, H256};
|
||||
use sr_primitives::BuildStorage;
|
||||
use sr_primitives::generic::BlockId;
|
||||
use sr_primitives::traits::{Header, NumberFor, SaturatedConversion};
|
||||
use substrate_executor::NativeExecutor;
|
||||
use sysinfo::{get_current_pid, ProcessExt, System, SystemExt};
|
||||
use tel::{telemetry, SUBSTRATE_INFO};
|
||||
use sr_primitives::traits::NumberFor;
|
||||
|
||||
pub use self::error::Error;
|
||||
pub use self::builder::{ServiceBuilder, ServiceBuilderExport, ServiceBuilderImport, ServiceBuilderRevert};
|
||||
pub use config::{Configuration, Roles, PruningMode};
|
||||
pub use chain_spec::{ChainSpec, Properties};
|
||||
pub use transaction_pool::txpool::{
|
||||
self, Pool as TransactionPool, Options as TransactionPoolOptions, ChainApi, IntoPoolError
|
||||
};
|
||||
pub use client::FinalityNotifications;
|
||||
|
||||
pub use components::{
|
||||
ServiceFactory, FullBackend, FullExecutor, LightBackend,
|
||||
LightExecutor, Components, PoolApi, ComponentClient, ComponentOffchainStorage,
|
||||
ComponentBlock, FullClient, LightClient, FullComponents, LightComponents,
|
||||
CodeExecutor, NetworkService, FactoryChainSpec, FactoryBlock,
|
||||
FactoryFullConfiguration, RuntimeGenesis, FactoryGenesis,
|
||||
ComponentExHash, ComponentExtrinsic, FactoryExtrinsic, InitialSessionKeys,
|
||||
};
|
||||
use components::{StartRpc, MaintainTransactionPool, OffchainWorker};
|
||||
pub use rpc::Metadata as RpcMetadata;
|
||||
#[doc(hidden)]
|
||||
pub use std::{ops::Deref, result::Result, sync::Arc};
|
||||
#[doc(hidden)]
|
||||
@@ -74,15 +66,15 @@ pub use futures::future::Executor;
|
||||
const DEFAULT_PROTOCOL_ID: &str = "sup";
|
||||
|
||||
/// Substrate service.
|
||||
pub struct Service<Components: components::Components> {
|
||||
client: Arc<ComponentClient<Components>>,
|
||||
select_chain: Option<Components::SelectChain>,
|
||||
network: Arc<components::NetworkService<Components>>,
|
||||
pub struct NewService<TCfg, TBl, TCl, TSc, TNetStatus, TNet, TTxPool, TOc> {
|
||||
client: Arc<TCl>,
|
||||
select_chain: Option<TSc>,
|
||||
network: Arc<TNet>,
|
||||
/// Sinks to propagate network status updates.
|
||||
network_status_sinks: Arc<Mutex<Vec<mpsc::UnboundedSender<(
|
||||
NetworkStatus<ComponentBlock<Components>>, NetworkState
|
||||
TNetStatus, NetworkState
|
||||
)>>>>,
|
||||
transaction_pool: Arc<TransactionPool<Components::TransactionPoolApi>>,
|
||||
transaction_pool: Arc<TTxPool>,
|
||||
/// A future that resolves when the service has exited, this is useful to
|
||||
/// make sure any internally spawned futures stop when the service does.
|
||||
exit: exit_future::Exit,
|
||||
@@ -100,31 +92,22 @@ pub struct Service<Components: components::Components> {
|
||||
/// The elements must then be polled manually.
|
||||
to_poll: Vec<Box<dyn Future<Item = (), Error = ()> + Send>>,
|
||||
/// Configuration of this Service
|
||||
config: FactoryFullConfiguration<Components::Factory>,
|
||||
rpc_handlers: components::RpcHandler,
|
||||
config: TCfg,
|
||||
rpc_handlers: rpc_servers::RpcHandler<rpc::Metadata>,
|
||||
_rpc: Box<dyn std::any::Any + Send + Sync>,
|
||||
_telemetry: Option<tel::Telemetry>,
|
||||
_telemetry_on_connect_sinks: Arc<Mutex<Vec<mpsc::UnboundedSender<()>>>>,
|
||||
_offchain_workers: Option<Arc<offchain::OffchainWorkers<
|
||||
ComponentClient<Components>,
|
||||
ComponentOffchainStorage<Components>,
|
||||
ComponentBlock<Components>>
|
||||
>>,
|
||||
_offchain_workers: Option<Arc<TOc>>,
|
||||
keystore: keystore::KeyStorePtr,
|
||||
marker: PhantomData<TBl>,
|
||||
}
|
||||
|
||||
/// Creates bare client without any networking.
|
||||
pub fn new_client<Factory: components::ServiceFactory>(
|
||||
config: &FactoryFullConfiguration<Factory>,
|
||||
) -> Result<Arc<ComponentClient<components::FullComponents<Factory>>>, error::Error> {
|
||||
let executor = NativeExecutor::new(config.default_heap_pages);
|
||||
/// A set of traits for the runtime genesis config.
|
||||
pub trait RuntimeGenesis: Serialize + DeserializeOwned + BuildStorage {}
|
||||
impl<T: Serialize + DeserializeOwned + BuildStorage> RuntimeGenesis for T {}
|
||||
|
||||
components::FullComponents::<Factory>::build_client(
|
||||
config,
|
||||
executor,
|
||||
None,
|
||||
).map(|r| r.0)
|
||||
}
|
||||
/// Alias for a an implementation of `futures::future::Executor`.
|
||||
pub type TaskExecutor = Arc<dyn Executor<Box<dyn Future<Item = (), Error = ()> + Send>> + Send + Sync>;
|
||||
|
||||
/// An handle for spawning tasks in the service.
|
||||
#[derive(Clone)]
|
||||
@@ -146,59 +129,38 @@ impl Executor<Box<dyn Future<Item = (), Error = ()> + Send>> for SpawnTaskHandle
|
||||
}
|
||||
}
|
||||
|
||||
/// Stream of events for connection established to a telemetry server.
|
||||
pub type TelemetryOnConnectNotifications = mpsc::UnboundedReceiver<()>;
|
||||
|
||||
/// Used to hook on telemetry connection established events.
|
||||
pub struct TelemetryOnConnect {
|
||||
/// Event stream.
|
||||
pub telemetry_connection_sinks: TelemetryOnConnectNotifications,
|
||||
}
|
||||
|
||||
impl<Components: components::Components> Service<Components> {
|
||||
/// Creates a new service.
|
||||
pub fn new(
|
||||
mut config: FactoryFullConfiguration<Components::Factory>,
|
||||
) -> Result<Self, error::Error> {
|
||||
macro_rules! new_impl {
|
||||
(
|
||||
$block:ty,
|
||||
$config:ident,
|
||||
$build_components:expr,
|
||||
$maintain_transaction_pool:expr,
|
||||
$offchain_workers:expr,
|
||||
$start_rpc:expr,
|
||||
) => {{
|
||||
let (signal, exit) = exit_future::signal();
|
||||
|
||||
// List of asynchronous tasks to spawn. We collect them, then spawn them all at once.
|
||||
let (to_spawn_tx, to_spawn_rx) =
|
||||
mpsc::unbounded::<Box<dyn Future<Item = (), Error = ()> + Send>>();
|
||||
|
||||
// Create client
|
||||
let executor = NativeExecutor::new(config.default_heap_pages);
|
||||
|
||||
let keystore = Keystore::open(config.keystore_path.clone(), config.keystore_password.clone())?;
|
||||
|
||||
let (client, on_demand) = Components::build_client(&config, executor, Some(keystore.clone()))?;
|
||||
let select_chain = Components::build_select_chain(&mut config, client.clone())?;
|
||||
|
||||
let transaction_pool = Arc::new(
|
||||
Components::build_transaction_pool(config.transaction_pool.clone(), client.clone())?
|
||||
);
|
||||
let transaction_pool_adapter = Arc::new(TransactionPoolAdapter {
|
||||
imports_external_transactions: !config.roles.is_light(),
|
||||
pool: transaction_pool.clone(),
|
||||
client: client.clone(),
|
||||
});
|
||||
|
||||
let (import_queue, finality_proof_request_builder) = Components::build_import_queue(
|
||||
&mut config,
|
||||
client.clone(),
|
||||
select_chain.clone(),
|
||||
Some(transaction_pool.clone()),
|
||||
)?;
|
||||
// Create all the components.
|
||||
let (
|
||||
client,
|
||||
on_demand,
|
||||
keystore,
|
||||
select_chain,
|
||||
import_queue,
|
||||
finality_proof_request_builder,
|
||||
finality_proof_provider,
|
||||
network_protocol,
|
||||
transaction_pool,
|
||||
rpc_extensions
|
||||
) = $build_components(&mut $config)?;
|
||||
let import_queue = Box::new(import_queue);
|
||||
let finality_proof_provider = Components::build_finality_proof_provider(client.clone())?;
|
||||
let chain_info = client.info().chain;
|
||||
|
||||
Components::RuntimeServices::generate_initial_session_keys(
|
||||
client.clone(),
|
||||
config.dev_key_seed.clone().map(|s| vec![s]).unwrap_or_default(),
|
||||
)?;
|
||||
|
||||
let version = config.full_version();
|
||||
let version = $config.full_version();
|
||||
info!("Highest known block at #{}", chain_info.best_number);
|
||||
telemetry!(
|
||||
SUBSTRATE_INFO;
|
||||
@@ -207,10 +169,14 @@ impl<Components: components::Components> Service<Components> {
|
||||
"best" => ?chain_info.best_hash
|
||||
);
|
||||
|
||||
let network_protocol = <Components::Factory>::build_network_protocol(&config)?;
|
||||
let transaction_pool_adapter = Arc::new(TransactionPoolAdapter {
|
||||
imports_external_transactions: !$config.roles.is_light(),
|
||||
pool: transaction_pool.clone(),
|
||||
client: client.clone(),
|
||||
});
|
||||
|
||||
let protocol_id = {
|
||||
let protocol_id_full = match config.chain_spec.protocol_id() {
|
||||
let protocol_id_full = match $config.chain_spec.protocol_id() {
|
||||
Some(pid) => pid,
|
||||
None => {
|
||||
warn!("Using default protocol ID {:?} because none is configured in the \
|
||||
@@ -223,8 +189,8 @@ impl<Components: components::Components> Service<Components> {
|
||||
};
|
||||
|
||||
let network_params = network::config::Params {
|
||||
roles: config.roles,
|
||||
network_config: config.network.clone(),
|
||||
roles: $config.roles,
|
||||
network_config: $config.network.clone(),
|
||||
chain: client.clone(),
|
||||
finality_proof_provider,
|
||||
finality_proof_request_builder,
|
||||
@@ -242,7 +208,7 @@ impl<Components: components::Components> Service<Components> {
|
||||
|
||||
#[allow(deprecated)]
|
||||
let offchain_storage = client.backend().offchain_storage();
|
||||
let offchain_workers = match (config.offchain_worker, offchain_storage) {
|
||||
let offchain_workers = match ($config.offchain_worker, offchain_storage) {
|
||||
(true, Some(db)) => {
|
||||
Some(Arc::new(offchain::OffchainWorkers::new(client.clone(), db)))
|
||||
},
|
||||
@@ -260,23 +226,25 @@ impl<Components: components::Components> Service<Components> {
|
||||
let offchain = offchain_workers.as_ref().map(Arc::downgrade);
|
||||
let to_spawn_tx_ = to_spawn_tx.clone();
|
||||
let network_state_info: Arc<dyn NetworkStateInfo + Send + Sync> = network.clone();
|
||||
let is_validator = config.roles.is_authority();
|
||||
let is_validator = $config.roles.is_authority();
|
||||
|
||||
let events = client.import_notification_stream()
|
||||
.map(|v| Ok::<_, ()>(v)).compat()
|
||||
.for_each(move |notification| {
|
||||
let number = *notification.header.number();
|
||||
let txpool = txpool.upgrade();
|
||||
|
||||
if let (Some(txpool), Some(client)) = (txpool.upgrade(), wclient.upgrade()) {
|
||||
Components::RuntimeServices::maintain_transaction_pool(
|
||||
if let (Some(txpool), Some(client)) = (txpool.as_ref(), wclient.upgrade()) {
|
||||
$maintain_transaction_pool(
|
||||
&BlockId::hash(notification.hash),
|
||||
&*client,
|
||||
&*txpool,
|
||||
).map_err(|e| warn!("Pool error processing new block: {:?}", e))?;
|
||||
}
|
||||
|
||||
if let (Some(txpool), Some(offchain)) = (txpool.upgrade(), offchain.as_ref().and_then(|o| o.upgrade())) {
|
||||
let future = Components::RuntimeServices::offchain_workers(
|
||||
let offchain = offchain.as_ref().and_then(|o| o.upgrade());
|
||||
if let (Some(txpool), Some(offchain)) = (txpool, offchain) {
|
||||
let future = $offchain_workers(
|
||||
&number,
|
||||
&offchain,
|
||||
&txpool,
|
||||
@@ -321,7 +289,7 @@ impl<Components: components::Components> Service<Components> {
|
||||
let client_ = client.clone();
|
||||
let mut sys = System::new();
|
||||
let self_pid = get_current_pid().ok();
|
||||
let (netstat_tx, netstat_rx) = mpsc::unbounded::<(NetworkStatus<ComponentBlock<Components>>, NetworkState)>();
|
||||
let (netstat_tx, netstat_rx) = mpsc::unbounded::<(NetworkStatus<_>, NetworkState)>();
|
||||
network_status_sinks.lock().push(netstat_tx);
|
||||
let tel_task = netstat_rx.for_each(move |(net_status, network_state)| {
|
||||
let info = client_.info();
|
||||
@@ -374,23 +342,23 @@ impl<Components: components::Components> Service<Components> {
|
||||
let (system_rpc_tx, system_rpc_rx) = futures03::channel::mpsc::unbounded();
|
||||
let gen_handler = || {
|
||||
let system_info = rpc::system::SystemInfo {
|
||||
chain_name: config.chain_spec.name().into(),
|
||||
impl_name: config.impl_name.into(),
|
||||
impl_version: config.impl_version.into(),
|
||||
properties: config.chain_spec.properties(),
|
||||
chain_name: $config.chain_spec.name().into(),
|
||||
impl_name: $config.impl_name.into(),
|
||||
impl_version: $config.impl_version.into(),
|
||||
properties: $config.chain_spec.properties(),
|
||||
};
|
||||
Components::RuntimeServices::start_rpc(
|
||||
$start_rpc(
|
||||
client.clone(),
|
||||
system_rpc_tx.clone(),
|
||||
system_info.clone(),
|
||||
Arc::new(SpawnTaskHandle { sender: to_spawn_tx.clone() }),
|
||||
transaction_pool.clone(),
|
||||
Components::build_rpc_extensions(client.clone(), transaction_pool.clone()),
|
||||
rpc_extensions.clone(),
|
||||
keystore.clone(),
|
||||
)
|
||||
};
|
||||
let rpc_handlers = gen_handler();
|
||||
let rpc = start_rpc_servers(&config, gen_handler)?;
|
||||
let rpc = start_rpc_servers(&$config, gen_handler)?;
|
||||
|
||||
let _ = to_spawn_tx.unbounded_send(Box::new(build_network_future(
|
||||
network_mut,
|
||||
@@ -406,17 +374,17 @@ impl<Components: components::Components> Service<Components> {
|
||||
let telemetry_connection_sinks: Arc<Mutex<Vec<mpsc::UnboundedSender<()>>>> = Default::default();
|
||||
|
||||
// Telemetry
|
||||
let telemetry = config.telemetry_endpoints.clone().map(|endpoints| {
|
||||
let is_authority = config.roles.is_authority();
|
||||
let telemetry = $config.telemetry_endpoints.clone().map(|endpoints| {
|
||||
let is_authority = $config.roles.is_authority();
|
||||
let network_id = network.local_peer_id().to_base58();
|
||||
let name = config.name.clone();
|
||||
let impl_name = config.impl_name.to_owned();
|
||||
let name = $config.name.clone();
|
||||
let impl_name = $config.impl_name.to_owned();
|
||||
let version = version.clone();
|
||||
let chain_name = config.chain_spec.name().to_owned();
|
||||
let chain_name = $config.chain_spec.name().to_owned();
|
||||
let telemetry_connection_sinks_ = telemetry_connection_sinks.clone();
|
||||
let telemetry = tel::init_telemetry(tel::TelemetryConfig {
|
||||
endpoints,
|
||||
wasm_external_transport: config.telemetry_external_transport.take(),
|
||||
wasm_external_transport: $config.telemetry_external_transport.take(),
|
||||
});
|
||||
let future = telemetry.clone()
|
||||
.map(|ev| Ok::<_, ()>(ev))
|
||||
@@ -446,7 +414,7 @@ impl<Components: components::Components> Service<Components> {
|
||||
telemetry
|
||||
});
|
||||
|
||||
Ok(Service {
|
||||
Ok(NewService {
|
||||
client,
|
||||
network,
|
||||
network_status_sinks,
|
||||
@@ -458,71 +426,65 @@ impl<Components: components::Components> Service<Components> {
|
||||
to_spawn_tx,
|
||||
to_spawn_rx,
|
||||
to_poll: Vec::new(),
|
||||
config,
|
||||
$config,
|
||||
rpc_handlers,
|
||||
_rpc: rpc,
|
||||
_telemetry: telemetry,
|
||||
_offchain_workers: offchain_workers,
|
||||
_telemetry_on_connect_sinks: telemetry_connection_sinks.clone(),
|
||||
keystore,
|
||||
marker: PhantomData::<$block>,
|
||||
})
|
||||
}
|
||||
}}
|
||||
}
|
||||
|
||||
/// Returns a reference to the config passed at initialization.
|
||||
pub fn config(&self) -> &FactoryFullConfiguration<Components::Factory> {
|
||||
&self.config
|
||||
}
|
||||
mod builder;
|
||||
|
||||
/// Returns a reference to the config passed at initialization.
|
||||
///
|
||||
/// > **Note**: This method is currently necessary because we extract some elements from the
|
||||
/// > configuration at the end of the service initialization. It is intended to be
|
||||
/// > removed.
|
||||
pub fn config_mut(&mut self) -> &mut FactoryFullConfiguration<Components::Factory> {
|
||||
&mut self.config
|
||||
}
|
||||
/// Abstraction over a Substrate service.
|
||||
pub trait AbstractService: 'static + Future<Item = (), Error = Error> +
|
||||
Executor<Box<dyn Future<Item = (), Error = ()> + Send>> + Send {
|
||||
/// Type of block of this chain.
|
||||
type Block: BlockT<Hash = H256>;
|
||||
/// Backend storage for the client.
|
||||
type Backend: 'static + client::backend::Backend<Self::Block, Blake2Hasher>;
|
||||
/// How to execute calls towards the runtime.
|
||||
type CallExecutor: 'static + client::CallExecutor<Self::Block, Blake2Hasher> + Send + Sync + Clone;
|
||||
/// API that the runtime provides.
|
||||
type RuntimeApi: Send + Sync;
|
||||
/// Configuration struct of the service.
|
||||
type Config;
|
||||
/// Chain selection algorithm.
|
||||
type SelectChain;
|
||||
/// API of the transaction pool.
|
||||
type TransactionPoolApi: ChainApi<Block = Self::Block>;
|
||||
/// Network specialization.
|
||||
type NetworkSpecialization: NetworkSpecialization<Self::Block>;
|
||||
|
||||
/// Get event stream for telemetry connection established events.
|
||||
pub fn telemetry_on_connect_stream(&self) -> TelemetryOnConnectNotifications {
|
||||
let (sink, stream) = mpsc::unbounded();
|
||||
self._telemetry_on_connect_sinks.lock().push(sink);
|
||||
stream
|
||||
}
|
||||
fn telemetry_on_connect_stream(&self) -> mpsc::UnboundedReceiver<()>;
|
||||
|
||||
/// Return a shared instance of Telemetry (if enabled)
|
||||
pub fn telemetry(&self) -> Option<tel::Telemetry> {
|
||||
self._telemetry.as_ref().map(|t| t.clone())
|
||||
}
|
||||
/// Returns the configuration passed on construction.
|
||||
fn config(&self) -> &Self::Config;
|
||||
|
||||
/// Returns the keystore instance.
|
||||
pub fn keystore(&self) -> keystore::KeyStorePtr {
|
||||
self.keystore.clone()
|
||||
}
|
||||
/// Returns the configuration passed on construction.
|
||||
fn config_mut(&mut self) -> &mut Self::Config;
|
||||
|
||||
/// return a shared instance of Telemetry (if enabled)
|
||||
fn telemetry(&self) -> Option<tel::Telemetry>;
|
||||
|
||||
/// Spawns a task in the background that runs the future passed as parameter.
|
||||
pub fn spawn_task(&self, task: impl Future<Item = (), Error = ()> + Send + 'static) {
|
||||
let _ = self.to_spawn_tx.unbounded_send(Box::new(task));
|
||||
}
|
||||
fn spawn_task(&self, task: impl Future<Item = (), Error = ()> + Send + 'static);
|
||||
|
||||
/// Spawns a task in the background that runs the future passed as
|
||||
/// parameter. The given task is considered essential, i.e. if it errors we
|
||||
/// trigger a service exit.
|
||||
pub fn spawn_essential_task(&self, task: impl Future<Item = (), Error = ()> + Send + 'static) {
|
||||
let essential_failed = self.essential_failed.clone();
|
||||
let essential_task = Box::new(task.map_err(move |_| {
|
||||
error!("Essential task failed. Shutting down service.");
|
||||
essential_failed.store(true, Ordering::Relaxed);
|
||||
}));
|
||||
|
||||
let _ = self.to_spawn_tx.unbounded_send(essential_task);
|
||||
}
|
||||
fn spawn_essential_task(&self, task: impl Future<Item = (), Error = ()> + Send + 'static);
|
||||
|
||||
/// Returns a handle for spawning tasks.
|
||||
pub fn spawn_task_handle(&self) -> SpawnTaskHandle {
|
||||
SpawnTaskHandle {
|
||||
sender: self.to_spawn_tx.clone(),
|
||||
}
|
||||
}
|
||||
fn spawn_task_handle(&self) -> SpawnTaskHandle;
|
||||
|
||||
/// Returns the keystore that stores keys.
|
||||
fn keystore(&self) -> keystore::KeyStorePtr;
|
||||
|
||||
/// Starts an RPC query.
|
||||
///
|
||||
@@ -533,46 +495,124 @@ impl<Components: components::Components> Service<Components> {
|
||||
///
|
||||
/// If the request subscribes you to events, the `Sender` in the `RpcSession` object is used to
|
||||
/// send back spontaneous events.
|
||||
pub fn rpc_query(&self, mem: &RpcSession, request: &str)
|
||||
-> impl Future<Item = Option<String>, Error = ()>
|
||||
{
|
||||
self.rpc_handlers.handle_request(request, mem.metadata.clone())
|
||||
}
|
||||
fn rpc_query(&self, mem: &RpcSession, request: &str) -> Box<dyn Future<Item = Option<String>, Error = ()> + Send>;
|
||||
|
||||
/// Get shared client instance.
|
||||
pub fn client(&self) -> Arc<ComponentClient<Components>> {
|
||||
fn client(&self) -> Arc<client::Client<Self::Backend, Self::CallExecutor, Self::Block, Self::RuntimeApi>>;
|
||||
|
||||
/// Get clone of select chain.
|
||||
fn select_chain(&self) -> Option<Self::SelectChain>;
|
||||
|
||||
/// Get shared network instance.
|
||||
fn network(&self) -> Arc<NetworkService<Self::Block, Self::NetworkSpecialization, H256>>;
|
||||
|
||||
/// Returns a receiver that periodically receives a status of the network.
|
||||
fn network_status(&self) -> mpsc::UnboundedReceiver<(NetworkStatus<Self::Block>, NetworkState)>;
|
||||
|
||||
/// Get shared transaction pool instance.
|
||||
fn transaction_pool(&self) -> Arc<TransactionPool<Self::TransactionPoolApi>>;
|
||||
|
||||
/// Get a handle to a future that will resolve on exit.
|
||||
fn on_exit(&self) -> ::exit_future::Exit;
|
||||
}
|
||||
|
||||
impl<TCfg, TBl, TBackend, TExec, TRtApi, TSc, TNetSpec, TExPoolApi, TOc> AbstractService for
|
||||
NewService<TCfg, TBl, Client<TBackend, TExec, TBl, TRtApi>, TSc, NetworkStatus<TBl>,
|
||||
NetworkService<TBl, TNetSpec, H256>, TransactionPool<TExPoolApi>, TOc>
|
||||
where TCfg: 'static + Send,
|
||||
TBl: BlockT<Hash = H256>,
|
||||
TBackend: 'static + client::backend::Backend<TBl, Blake2Hasher>,
|
||||
TExec: 'static + client::CallExecutor<TBl, Blake2Hasher> + Send + Sync + Clone,
|
||||
TRtApi: 'static + Send + Sync,
|
||||
TSc: 'static + Clone + Send,
|
||||
TExPoolApi: 'static + ChainApi<Block = TBl>,
|
||||
TOc: 'static + Send + Sync,
|
||||
TNetSpec: NetworkSpecialization<TBl>,
|
||||
{
|
||||
type Block = TBl;
|
||||
type Backend = TBackend;
|
||||
type CallExecutor = TExec;
|
||||
type RuntimeApi = TRtApi;
|
||||
type Config = TCfg;
|
||||
type SelectChain = TSc;
|
||||
type TransactionPoolApi = TExPoolApi;
|
||||
type NetworkSpecialization = TNetSpec;
|
||||
|
||||
fn config(&self) -> &Self::Config {
|
||||
&self.config
|
||||
}
|
||||
|
||||
fn config_mut(&mut self) -> &mut Self::Config {
|
||||
&mut self.config
|
||||
}
|
||||
|
||||
fn telemetry_on_connect_stream(&self) -> mpsc::UnboundedReceiver<()> {
|
||||
let (sink, stream) = mpsc::unbounded();
|
||||
self._telemetry_on_connect_sinks.lock().push(sink);
|
||||
stream
|
||||
}
|
||||
|
||||
fn telemetry(&self) -> Option<tel::Telemetry> {
|
||||
self._telemetry.as_ref().map(|t| t.clone())
|
||||
}
|
||||
|
||||
fn keystore(&self) -> keystore::KeyStorePtr {
|
||||
self.keystore.clone()
|
||||
}
|
||||
|
||||
fn spawn_task(&self, task: impl Future<Item = (), Error = ()> + Send + 'static) {
|
||||
let _ = self.to_spawn_tx.unbounded_send(Box::new(task));
|
||||
}
|
||||
|
||||
fn spawn_essential_task(&self, task: impl Future<Item = (), Error = ()> + Send + 'static) {
|
||||
let essential_failed = self.essential_failed.clone();
|
||||
let essential_task = Box::new(task.map_err(move |_| {
|
||||
error!("Essential task failed. Shutting down service.");
|
||||
essential_failed.store(true, Ordering::Relaxed);
|
||||
}));
|
||||
|
||||
let _ = self.to_spawn_tx.unbounded_send(essential_task);
|
||||
}
|
||||
|
||||
fn spawn_task_handle(&self) -> SpawnTaskHandle {
|
||||
SpawnTaskHandle {
|
||||
sender: self.to_spawn_tx.clone(),
|
||||
}
|
||||
}
|
||||
|
||||
fn rpc_query(&self, mem: &RpcSession, request: &str) -> Box<dyn Future<Item = Option<String>, Error = ()> + Send> {
|
||||
Box::new(self.rpc_handlers.handle_request(request, mem.metadata.clone()))
|
||||
}
|
||||
|
||||
fn client(&self) -> Arc<client::Client<Self::Backend, Self::CallExecutor, Self::Block, Self::RuntimeApi>> {
|
||||
self.client.clone()
|
||||
}
|
||||
|
||||
/// Get clone of select chain.
|
||||
pub fn select_chain(&self) -> Option<<Components as components::Components>::SelectChain> {
|
||||
fn select_chain(&self) -> Option<Self::SelectChain> {
|
||||
self.select_chain.clone()
|
||||
}
|
||||
|
||||
/// Get shared network instance.
|
||||
pub fn network(&self) -> Arc<components::NetworkService<Components>> {
|
||||
fn network(&self) -> Arc<NetworkService<Self::Block, Self::NetworkSpecialization, H256>> {
|
||||
self.network.clone()
|
||||
}
|
||||
|
||||
/// Returns a receiver that periodically receives a status of the network.
|
||||
pub fn network_status(&self) -> mpsc::UnboundedReceiver<(NetworkStatus<ComponentBlock<Components>>, NetworkState)> {
|
||||
fn network_status(&self) -> mpsc::UnboundedReceiver<(NetworkStatus<Self::Block>, NetworkState)> {
|
||||
let (sink, stream) = mpsc::unbounded();
|
||||
self.network_status_sinks.lock().push(sink);
|
||||
stream
|
||||
}
|
||||
|
||||
/// Get shared transaction pool instance.
|
||||
pub fn transaction_pool(&self) -> Arc<TransactionPool<Components::TransactionPoolApi>> {
|
||||
fn transaction_pool(&self) -> Arc<TransactionPool<Self::TransactionPoolApi>> {
|
||||
self.transaction_pool.clone()
|
||||
}
|
||||
|
||||
/// Get a handle to a future that will resolve on exit.
|
||||
pub fn on_exit(&self) -> ::exit_future::Exit {
|
||||
fn on_exit(&self) -> ::exit_future::Exit {
|
||||
self.exit.clone()
|
||||
}
|
||||
}
|
||||
|
||||
impl<Components> Future for Service<Components> where Components: components::Components {
|
||||
impl<TCfg, TBl, TCl, TSc, TNetStatus, TNet, TTxPool, TOc> Future for
|
||||
NewService<TCfg, TBl, TCl, TSc, TNetStatus, TNet, TTxPool, TOc> {
|
||||
type Item = ();
|
||||
type Error = Error;
|
||||
|
||||
@@ -603,9 +643,8 @@ impl<Components> Future for Service<Components> where Components: components::Co
|
||||
}
|
||||
}
|
||||
|
||||
impl<Components> Executor<Box<dyn Future<Item = (), Error = ()> + Send>>
|
||||
for Service<Components> where Components: components::Components
|
||||
{
|
||||
impl<TCfg, TBl, TCl, TSc, TNetStatus, TNet, TTxPool, TOc> Executor<Box<dyn Future<Item = (), Error = ()> + Send>> for
|
||||
NewService<TCfg, TBl, TCl, TSc, TNetStatus, TNet, TTxPool, TOc> {
|
||||
fn execute(
|
||||
&self,
|
||||
future: Box<dyn Future<Item = (), Error = ()> + Send>
|
||||
@@ -746,7 +785,8 @@ pub struct NetworkStatus<B: BlockT> {
|
||||
pub average_upload_per_sec: u64,
|
||||
}
|
||||
|
||||
impl<Components> Drop for Service<Components> where Components: components::Components {
|
||||
impl<TCfg, TBl, TCl, TSc, TNetStatus, TNet, TTxPool, TOc> Drop for
|
||||
NewService<TCfg, TBl, TCl, TSc, TNetStatus, TNet, TTxPool, TOc> {
|
||||
fn drop(&mut self) {
|
||||
debug!(target: "service", "Substrate service shutdown");
|
||||
if let Some(signal) = self.signal.take() {
|
||||
@@ -757,7 +797,7 @@ impl<Components> Drop for Service<Components> where Components: components::Comp
|
||||
|
||||
/// Starts RPC servers that run in their own thread, and returns an opaque object that keeps them alive.
|
||||
#[cfg(not(target_os = "unknown"))]
|
||||
fn start_rpc_servers<C, G, H: FnMut() -> components::RpcHandler>(
|
||||
fn start_rpc_servers<C, G, H: FnMut() -> rpc_servers::RpcHandler<rpc::Metadata>>(
|
||||
config: &Configuration<C, G>,
|
||||
mut gen_handler: H
|
||||
) -> Result<Box<dyn std::any::Any + Send + Sync>, error::Error> {
|
||||
@@ -906,225 +946,6 @@ where
|
||||
}
|
||||
}
|
||||
|
||||
/// Constructs a service factory with the given name that implements the `ServiceFactory` trait.
|
||||
/// The required parameters are required to be given in the exact order. Some parameters are followed
|
||||
/// by `{}` blocks. These blocks are required and used to initialize the given parameter.
|
||||
/// In these block it is required to write a closure that takes the same number of arguments,
|
||||
/// the corresponding function in the `ServiceFactory` trait provides.
|
||||
///
|
||||
/// # Example
|
||||
///
|
||||
/// ```
|
||||
/// # use substrate_service::{
|
||||
/// # construct_service_factory, Service, FullBackend, FullExecutor, LightBackend, LightExecutor,
|
||||
/// # FullComponents, LightComponents, FactoryFullConfiguration, FullClient
|
||||
/// # };
|
||||
/// # use transaction_pool::{self, txpool::{Pool as TransactionPool}};
|
||||
/// # use network::{config::DummyFinalityProofRequestBuilder, construct_simple_protocol};
|
||||
/// # use client::{self, LongestChain};
|
||||
/// # use consensus_common::import_queue::{BasicQueue, Verifier};
|
||||
/// # use consensus_common::{BlockOrigin, BlockImportParams, well_known_cache_keys::Id as CacheKeyId};
|
||||
/// # use node_runtime::{GenesisConfig, RuntimeApi};
|
||||
/// # use std::sync::Arc;
|
||||
/// # use node_primitives::Block;
|
||||
/// # use babe_primitives::AuthorityPair as BabePair;
|
||||
/// # use grandpa_primitives::AuthorityPair as GrandpaPair;
|
||||
/// # use sr_primitives::Justification;
|
||||
/// # use sr_primitives::traits::Block as BlockT;
|
||||
/// # use grandpa;
|
||||
/// # construct_simple_protocol! {
|
||||
/// # pub struct NodeProtocol where Block = Block { }
|
||||
/// # }
|
||||
/// # struct MyVerifier;
|
||||
/// # impl<B: BlockT> Verifier<B> for MyVerifier {
|
||||
/// # fn verify(
|
||||
/// # &mut self,
|
||||
/// # origin: BlockOrigin,
|
||||
/// # header: B::Header,
|
||||
/// # justification: Option<Justification>,
|
||||
/// # body: Option<Vec<B::Extrinsic>>,
|
||||
/// # ) -> Result<(BlockImportParams<B>, Option<Vec<(CacheKeyId, Vec<u8>)>>), String> {
|
||||
/// # unimplemented!();
|
||||
/// # }
|
||||
/// # }
|
||||
/// type FullChainApi<T> = transaction_pool::ChainApi<
|
||||
/// client::Client<FullBackend<T>, FullExecutor<T>, Block, RuntimeApi>, Block>;
|
||||
/// type LightChainApi<T> = transaction_pool::ChainApi<
|
||||
/// client::Client<LightBackend<T>, LightExecutor<T>, Block, RuntimeApi>, Block>;
|
||||
///
|
||||
/// construct_service_factory! {
|
||||
/// struct Factory {
|
||||
/// // Declare the block type
|
||||
/// Block = Block,
|
||||
/// RuntimeApi = RuntimeApi,
|
||||
/// // Declare the network protocol and give an initializer.
|
||||
/// NetworkProtocol = NodeProtocol { |config| Ok(NodeProtocol::new()) },
|
||||
/// RuntimeDispatch = node_executor::Executor,
|
||||
/// FullTransactionPoolApi = FullChainApi<Self>
|
||||
/// { |config, client| Ok(TransactionPool::new(config, transaction_pool::ChainApi::new(client))) },
|
||||
/// LightTransactionPoolApi = LightChainApi<Self>
|
||||
/// { |config, client| Ok(TransactionPool::new(config, transaction_pool::ChainApi::new(client))) },
|
||||
/// Genesis = GenesisConfig,
|
||||
/// Configuration = (),
|
||||
/// FullService = FullComponents<Self>
|
||||
/// { |config| <FullComponents<Factory>>::new(config) },
|
||||
/// // Setup as Consensus Authority (if the role and key are given)
|
||||
/// AuthoritySetup = {
|
||||
/// |service: Self::FullService| {
|
||||
/// Ok(service)
|
||||
/// }},
|
||||
/// LightService = LightComponents<Self>
|
||||
/// { |config| <LightComponents<Factory>>::new(config) },
|
||||
/// FullImportQueue = BasicQueue<Block>
|
||||
/// { |_, client, _, _| Ok(BasicQueue::new(MyVerifier, Box::new(client), None, None)) },
|
||||
/// LightImportQueue = BasicQueue<Block>
|
||||
/// { |_, client| {
|
||||
/// let fprb = Box::new(DummyFinalityProofRequestBuilder::default()) as Box<_>;
|
||||
/// Ok((BasicQueue::new(MyVerifier, Box::new(client), None, None), fprb))
|
||||
/// }},
|
||||
/// SelectChain = LongestChain<FullBackend<Self>, Self::Block>
|
||||
/// { |config: &FactoryFullConfiguration<Self>, client: Arc<FullClient<Self>>| {
|
||||
/// #[allow(deprecated)]
|
||||
/// Ok(LongestChain::new(client.backend().clone()))
|
||||
/// }},
|
||||
/// FinalityProofProvider = { |client: Arc<FullClient<Self>>| {
|
||||
/// Ok(Some(Arc::new(grandpa::FinalityProofProvider::new(client.clone(), client)) as _))
|
||||
/// }},
|
||||
/// RpcExtensions = (),
|
||||
/// }
|
||||
/// }
|
||||
/// ```
|
||||
#[macro_export]
|
||||
macro_rules! construct_service_factory {
|
||||
(
|
||||
$(#[$attr:meta])*
|
||||
struct $name:ident {
|
||||
Block = $block:ty,
|
||||
RuntimeApi = $runtime_api:ty,
|
||||
NetworkProtocol = $protocol:ty { $( $protocol_init:tt )* },
|
||||
RuntimeDispatch = $dispatch:ty,
|
||||
FullTransactionPoolApi = $full_transaction:ty { $( $full_transaction_init:tt )* },
|
||||
LightTransactionPoolApi = $light_transaction:ty { $( $light_transaction_init:tt )* },
|
||||
Genesis = $genesis:ty,
|
||||
Configuration = $config:ty,
|
||||
FullService = $full_service:ty { $( $full_service_init:tt )* },
|
||||
AuthoritySetup = { $( $authority_setup:tt )* },
|
||||
LightService = $light_service:ty { $( $light_service_init:tt )* },
|
||||
FullImportQueue = $full_import_queue:ty
|
||||
{ $( $full_import_queue_init:tt )* },
|
||||
LightImportQueue = $light_import_queue:ty
|
||||
{ $( $light_import_queue_init:tt )* },
|
||||
SelectChain = $select_chain:ty
|
||||
{ $( $select_chain_init:tt )* },
|
||||
FinalityProofProvider = { $( $finality_proof_provider_init:tt )* },
|
||||
RpcExtensions = $rpc_extensions_ty:ty
|
||||
$( { $( $rpc_extensions:tt )* } )?,
|
||||
}
|
||||
) => {
|
||||
$( #[$attr] )*
|
||||
pub struct $name {}
|
||||
|
||||
#[allow(unused_variables)]
|
||||
impl $crate::ServiceFactory for $name {
|
||||
type Block = $block;
|
||||
type RuntimeApi = $runtime_api;
|
||||
type NetworkProtocol = $protocol;
|
||||
type RuntimeDispatch = $dispatch;
|
||||
type FullTransactionPoolApi = $full_transaction;
|
||||
type LightTransactionPoolApi = $light_transaction;
|
||||
type Genesis = $genesis;
|
||||
type Configuration = $config;
|
||||
type FullService = $full_service;
|
||||
type LightService = $light_service;
|
||||
type FullImportQueue = $full_import_queue;
|
||||
type LightImportQueue = $light_import_queue;
|
||||
type SelectChain = $select_chain;
|
||||
type RpcExtensions = $rpc_extensions_ty;
|
||||
|
||||
fn build_full_transaction_pool(
|
||||
config: $crate::TransactionPoolOptions,
|
||||
client: $crate::Arc<$crate::FullClient<Self>>
|
||||
) -> $crate::Result<$crate::TransactionPool<Self::FullTransactionPoolApi>, $crate::Error>
|
||||
{
|
||||
( $( $full_transaction_init )* ) (config, client)
|
||||
}
|
||||
|
||||
fn build_light_transaction_pool(
|
||||
config: $crate::TransactionPoolOptions,
|
||||
client: $crate::Arc<$crate::LightClient<Self>>
|
||||
) -> $crate::Result<$crate::TransactionPool<Self::LightTransactionPoolApi>, $crate::Error>
|
||||
{
|
||||
( $( $light_transaction_init )* ) (config, client)
|
||||
}
|
||||
|
||||
fn build_network_protocol(config: &$crate::FactoryFullConfiguration<Self>)
|
||||
-> $crate::Result<Self::NetworkProtocol, $crate::Error>
|
||||
{
|
||||
( $( $protocol_init )* ) (config)
|
||||
}
|
||||
|
||||
fn build_select_chain(
|
||||
config: &mut $crate::FactoryFullConfiguration<Self>,
|
||||
client: Arc<$crate::FullClient<Self>>
|
||||
) -> $crate::Result<Self::SelectChain, $crate::Error> {
|
||||
( $( $select_chain_init )* ) (config, client)
|
||||
}
|
||||
|
||||
fn build_full_import_queue(
|
||||
config: &mut $crate::FactoryFullConfiguration<Self>,
|
||||
client: $crate::Arc<$crate::FullClient<Self>>,
|
||||
select_chain: Self::SelectChain,
|
||||
transaction_pool: Option<Arc<$crate::TransactionPool<Self::FullTransactionPoolApi>>>,
|
||||
) -> $crate::Result<Self::FullImportQueue, $crate::Error> {
|
||||
( $( $full_import_queue_init )* ) (config, client, select_chain, transaction_pool)
|
||||
}
|
||||
|
||||
fn build_light_import_queue(
|
||||
config: &mut FactoryFullConfiguration<Self>,
|
||||
client: Arc<$crate::LightClient<Self>>,
|
||||
) -> Result<(Self::LightImportQueue, $crate::BoxFinalityProofRequestBuilder<$block>), $crate::Error> {
|
||||
( $( $light_import_queue_init )* ) (config, client)
|
||||
}
|
||||
|
||||
fn build_finality_proof_provider(
|
||||
client: Arc<$crate::FullClient<Self>>
|
||||
) -> Result<Option<Arc<$crate::FinalityProofProvider<Self::Block>>>, $crate::Error> {
|
||||
( $( $finality_proof_provider_init )* ) (client)
|
||||
}
|
||||
|
||||
fn new_light(
|
||||
config: $crate::FactoryFullConfiguration<Self>
|
||||
) -> $crate::Result<Self::LightService, $crate::Error>
|
||||
{
|
||||
( $( $light_service_init )* ) (config)
|
||||
}
|
||||
|
||||
fn new_full(
|
||||
config: $crate::FactoryFullConfiguration<Self>
|
||||
) -> Result<Self::FullService, $crate::Error>
|
||||
{
|
||||
( $( $full_service_init )* ) (config).and_then(|service| {
|
||||
($( $authority_setup )*)(service)
|
||||
})
|
||||
}
|
||||
|
||||
fn build_full_rpc_extensions(
|
||||
client: Arc<$crate::FullClient<Self>>,
|
||||
transaction_pool: Arc<$crate::TransactionPool<Self::FullTransactionPoolApi>>,
|
||||
) -> Self::RpcExtensions {
|
||||
$( ( $( $rpc_extensions )* ) (client, transaction_pool) )?
|
||||
}
|
||||
|
||||
fn build_light_rpc_extensions(
|
||||
client: Arc<$crate::LightClient<Self>>,
|
||||
transaction_pool: Arc<$crate::TransactionPool<Self::LightTransactionPoolApi>>,
|
||||
) -> Self::RpcExtensions {
|
||||
$( ( $( $rpc_extensions )* ) (client, transaction_pool) )?
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
Reference in New Issue
Block a user