// Copyright 2017-2018 Parity Technologies (UK) Ltd.
// This file is part of Substrate.
// Substrate is free software: you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
// Substrate is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU General Public License for more details.
// You should have received a copy of the GNU General Public License
// along with Substrate. If not, see .
//! Substrate service. Starts a thread that spins up the network, client, and extrinsic pool.
//! Manages communication between them.
#![warn(unused_extern_crates)]
extern crate futures;
extern crate exit_future;
extern crate serde;
extern crate serde_json;
extern crate parking_lot;
extern crate substrate_keystore as keystore;
extern crate substrate_primitives as primitives;
extern crate sr_primitives as runtime_primitives;
extern crate substrate_consensus_common as consensus_common;
extern crate substrate_network as network;
extern crate substrate_executor;
extern crate substrate_client as client;
extern crate substrate_client_db as client_db;
extern crate parity_codec as codec;
extern crate substrate_transaction_pool as transaction_pool;
extern crate substrate_rpc_servers as rpc;
extern crate target_info;
extern crate tokio;
#[macro_use]
extern crate substrate_telemetry as tel;
#[macro_use]
extern crate error_chain;
#[macro_use]
extern crate slog; // needed until we can reexport `slog_info` from `substrate_telemetry`
#[macro_use]
extern crate log;
#[macro_use]
extern crate serde_derive;
#[cfg(test)]
extern crate substrate_test_client;
mod components;
mod error;
mod chain_spec;
pub mod config;
pub mod chain_ops;
use std::io;
use std::net::SocketAddr;
use std::collections::HashMap;
#[doc(hidden)]
pub use std::{ops::Deref, result::Result, sync::Arc};
use futures::prelude::*;
use keystore::Store as Keystore;
use client::BlockchainEvents;
use runtime_primitives::generic::BlockId;
use runtime_primitives::traits::{Header, As};
use exit_future::Signal;
#[doc(hidden)]
pub use tokio::runtime::TaskExecutor;
use substrate_executor::NativeExecutor;
use codec::{Encode, Decode};
pub use self::error::{ErrorKind, Error};
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::ExecutionStrategy;
pub use components::{ServiceFactory, FullBackend, FullExecutor, LightBackend,
LightExecutor, Components, PoolApi, ComponentClient,
ComponentBlock, FullClient, LightClient, FullComponents, LightComponents,
CodeExecutor, NetworkService, FactoryChainSpec, FactoryBlock,
FactoryFullConfiguration, RuntimeGenesis, FactoryGenesis,
ComponentExHash, ComponentExtrinsic, FactoryExtrinsic
};
use components::{StartRPC, MaintainTransactionPool};
#[doc(hidden)]
pub use network::OnDemand;
const DEFAULT_PROTOCOL_ID: &'static str = "sup";
/// Substrate service.
pub struct Service {
client: Arc>,
network: Option>>,
transaction_pool: Arc>,
keystore: Keystore,
exit: ::exit_future::Exit,
signal: Option,
/// Configuration of this Service
pub config: FactoryFullConfiguration,
_rpc: Box<::std::any::Any + Send + Sync>,
_telemetry: Option>,
}
/// Creates bare client without any networking.
pub fn new_client(config: &FactoryFullConfiguration)
-> Result>>, error::Error>
{
let executor = NativeExecutor::new();
let (client, _) = components::FullComponents::::build_client(
config,
executor,
)?;
Ok(client)
}
impl Service {
/// Creates a new service.
pub fn new(
mut config: FactoryFullConfiguration,
task_executor: TaskExecutor,
)
-> Result
{
let (signal, exit) = ::exit_future::signal();
// Create client
let executor = NativeExecutor::new();
let mut keystore = Keystore::open(config.keystore_path.as_str().into())?;
// This is meant to be for testing only
// FIXME: remove this - https://github.com/paritytech/substrate/issues/1063
for seed in &config.keys {
keystore.generate_from_seed(seed)?;
}
// Keep the public key for telemetry
let public_key = match keystore.contents()?.get(0) {
Some(public_key) => public_key.clone(),
None => {
let key = keystore.generate("")?;
let public_key = key.public();
info!("Generated a new keypair: {:?}", public_key);
public_key
}
};
let (client, on_demand) = Components::build_client(&config, executor)?;
let import_queue = Arc::new(Components::build_import_queue(&mut config, client.clone())?);
let best_header = client.best_block_header()?;
let version = config.full_version();
info!("Best block: #{}", best_header.number());
telemetry!("node.start"; "height" => best_header.number().as_(), "best" => ?best_header.hash());
let network_protocol = ::build_network_protocol(&config)?;
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 == Roles::LIGHT),
pool: transaction_pool.clone(),
client: client.clone(),
});
let network_params = network::config::Params {
config: network::config::ProtocolConfig { roles: config.roles },
network_config: config.network.clone(),
chain: client.clone(),
on_demand: on_demand.as_ref().map(|d| d.clone() as _),
transaction_pool: transaction_pool_adapter.clone() as _,
specialization: network_protocol,
};
let protocol_id = {
let protocol_id_full = config.chain_spec.protocol_id().unwrap_or(DEFAULT_PROTOCOL_ID).as_bytes();
let mut protocol_id = network::ProtocolId::default();
if protocol_id_full.len() > protocol_id.len() {
warn!("Protocol ID truncated to {} chars", protocol_id.len());
}
let id_len = protocol_id_full.len().min(protocol_id.len());
&mut protocol_id[0..id_len].copy_from_slice(&protocol_id_full[0..id_len]);
protocol_id
};
let has_bootnodes = !network_params.network_config.boot_nodes.is_empty();
let network = network::Service::new(
network_params,
protocol_id,
import_queue
)?;
on_demand.map(|on_demand| on_demand.set_service_link(Arc::downgrade(&network)));
{
// block notifications
let network = Arc::downgrade(&network);
let txpool = Arc::downgrade(&transaction_pool);
let wclient = Arc::downgrade(&client);
let events = client.import_notification_stream()
.for_each(move |notification| {
if let Some(network) = network.upgrade() {
network.on_block_imported(notification.hash, ¬ification.header);
}
if let (Some(txpool), Some(client)) = (txpool.upgrade(), wclient.upgrade()) {
Components::TransactionPool::on_block_imported(
&BlockId::hash(notification.hash),
&*client,
&*txpool,
).map_err(|e| warn!("Pool error processing new block: {:?}", e))?;
}
Ok(())
})
.select(exit.clone())
.then(|_| Ok(()));
task_executor.spawn(events);
}
{
// extrinsic notifications
let network = Arc::downgrade(&network);
let events = transaction_pool.import_notification_stream()
// TODO [ToDr] Consider throttling?
.for_each(move |_| {
if let Some(network) = network.upgrade() {
network.trigger_repropagate();
}
Ok(())
})
.select(exit.clone())
.then(|_| Ok(()));
task_executor.spawn(events);
}
// RPC
let system_info = rpc::apis::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(),
};
let rpc = Components::RPC::start_rpc(
client.clone(), network.clone(), has_bootnodes, system_info, config.rpc_http, config.rpc_ws, task_executor.clone(), transaction_pool.clone(),
)?;
// Telemetry
let telemetry = config.telemetry_url.clone().map(|url| {
let is_authority = config.roles == Roles::AUTHORITY;
let pubkey = format!("{}", public_key);
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();
Arc::new(tel::init_telemetry(tel::TelemetryConfig {
url: url,
on_connect: Box::new(move || {
telemetry!("system.connected";
"name" => name.clone(),
"implementation" => impl_name.clone(),
"version" => version.clone(),
"config" => "",
"chain" => chain_name.clone(),
"pubkey" => &pubkey,
"authority" => is_authority
);
}),
}))
});
Ok(Service {
client,
network: Some(network),
transaction_pool,
signal: Some(signal),
keystore,
config,
exit,
_rpc: Box::new(rpc),
_telemetry: telemetry,
})
}
/// give the authority key, if we are an authority and have a key
pub fn authority_key(&self) -> Option {
if self.config.roles != Roles::AUTHORITY { return None }
let keystore = &self.keystore;
if let Ok(Some(Ok(key))) = keystore.contents().map(|keys| keys.get(0)
.map(|k| keystore.load(k, "")))
{
Some(key)
} else {
None
}
}
pub fn telemetry(&self) -> Option> {
self._telemetry.as_ref().map(|t| t.clone())
}
}
impl Service where Components: components::Components {
/// Get shared client instance.
pub fn client(&self) -> Arc> {
self.client.clone()
}
/// Get shared network instance.
pub fn network(&self) -> Arc> {
self.network.as_ref().expect("self.network always Some").clone()
}
/// Get shared extrinsic pool instance.
pub fn transaction_pool(&self) -> Arc> {
self.transaction_pool.clone()
}
/// Get shared keystore.
pub fn keystore(&self) -> &Keystore {
&self.keystore
}
/// Get a handle to a future that will resolve on exit.
pub fn on_exit(&self) -> ::exit_future::Exit {
self.exit.clone()
}
}
impl Drop for Service where Components: components::Components {
fn drop(&mut self) {
debug!(target: "service", "Substrate service shutdown");
drop(self.network.take());
if let Some(signal) = self.signal.take() {
signal.fire();
}
}
}
fn maybe_start_server(address: Option, start: F) -> Result