Cherry pick upstream (#11)

* core: allow setting max ws rpc connections (#2632)

* core: allow setting max ws rpc connections

* style: break long lines

* core: fix service tests

* Bump structopt and fix compilation. (#2736)

core/cli/src/lib.rs

@@ -51,7 +51,7 @@ pub use structopt::clap::App;
 use params::{
 	RunCmd, PurgeChainCmd, RevertCmd, ImportBlocksCmd, ExportBlocksCmd, BuildSpecCmd,
 	NetworkConfigurationParams, SharedParams, MergeParameters, TransactionPoolParams,
-	NodeKeyParams, NodeKeyType
+	NodeKeyParams, NodeKeyType, Cors,
 };
 pub use params::{NoCustom, CoreParams};
 pub use traits::{GetLogFilter, AugmentClap};
@@ -480,11 +480,12 @@ where
 	config.rpc_ws = Some(
 		parse_address(&format!("{}:{}", ws_interface, 9944), cli.ws_port)?
 	);
+	config.rpc_ws_max_connections = cli.ws_max_connections;
 	config.rpc_cors = cli.rpc_cors.unwrap_or_else(|| if is_dev {
 		log::warn!("Running in --dev mode, RPC CORS has been disabled.");
-		None
+		Cors::All
 	} else {
-		Some(vec![
+		Cors::List(vec![
 			"http://localhost:*".into(),
 			"http://127.0.0.1:*".into(),
 			"https://localhost:*".into(),
@@ -494,7 +495,7 @@ where
 			"https://cennznet-ui.centrality.me".into(),
 			"https://cennznet-ui.centrality.cloud".into(),
 		])
-	});
+	}).into();
 
 	// Override telemetry
 	if cli.no_telemetry {

core/cli/src/params.rs

@@ -333,13 +333,17 @@ pub struct RunCmd {
 	#[structopt(long = "ws-port", value_name = "PORT")]
 	pub ws_port: Option<u16>,
 
+	/// Maximum number of WS RPC server connections.
+	#[structopt(long = "ws-max-connections", value_name = "COUNT")]
+	pub ws_max_connections: Option<usize>,
+
 	/// Specify browser Origins allowed to access the HTTP & WS RPC servers.
 	/// It's a comma-separated list of origins (protocol://domain or special `null` value).
 	/// Value of `all` will disable origin validation.
 	/// Default is to allow localhost, https://polkadot.js.org and https://substrate-ui.parity.io origins.
 	/// When running in --dev mode the default is to allow all origins.
 	#[structopt(long = "rpc-cors", value_name = "ORIGINS", parse(try_from_str = "parse_cors"))]
-	pub rpc_cors: Option<Option<Vec<String>>>,
+	pub rpc_cors: Option<Cors>,
 
 	/// Specify the pruning mode, a number of blocks to keep or 'archive'. Default is 256.
 	#[structopt(long = "pruning", value_name = "PRUNING_MODE")]
@@ -482,8 +486,29 @@ fn parse_telemetry_endpoints(s: &str) -> Result<(String, u8), Box<std::error::Er
 	}
 }
 
+/// CORS setting
+///
+/// The type is introduced to overcome `Option<Option<T>>`
+/// handling of `structopt`.
+#[derive(Clone, Debug)]
+pub enum Cors {
+	/// All hosts allowed
+	All,
+	/// Only hosts on the list are allowed.
+	List(Vec<String>),
+}
+
+impl From<Cors> for Option<Vec<String>> {
+	fn from(cors: Cors) -> Self {
+		match cors {
+			Cors::All => None,
+			Cors::List(list) => Some(list),
+		}
+	}
+}
+
 /// Parse cors origins
-fn parse_cors(s: &str) -> Result<Option<Vec<String>>, Box<std::error::Error>> {
+fn parse_cors(s: &str) -> Result<Cors, Box<std::error::Error>> {
 	let mut is_all = false;
 	let mut origins = Vec::new();
 	for part in s.split(',') {
@@ -496,7 +521,7 @@ fn parse_cors(s: &str) -> Result<Option<Vec<String>>, Box<std::error::Error>> {
 		}
 	}
 
-	Ok(if is_all { None } else { Some(origins) })
+	Ok(if is_all { Cors::All } else { Cors::List(origins) })
 }
 
 impl_augment_clap!(RunCmd);

core/rpc-servers/src/lib.rs

@@ -24,9 +24,12 @@ use std::io;
 use log::error;
 use sr_primitives::{traits::{Block as BlockT, NumberFor}, generic::SignedBlock};
 
-/// Maximal payload accepted by RPC servers
+/// Maximal payload accepted by RPC servers.
 const MAX_PAYLOAD: usize = 15 * 1024 * 1024;
 
+/// Default maximum number of connections for WS RPC servers.
+const WS_MAX_CONNECTIONS: usize = 100;
+
 type Metadata = apis::metadata::Metadata;
 type RpcHandler = pubsub::PubSubHandler<Metadata>;
 pub type HttpServer = http::Server;
@@ -76,11 +79,13 @@ pub fn start_http(
 /// Start WS server listening on given address.
 pub fn start_ws(
 	addr: &std::net::SocketAddr,
+	max_connections: Option<usize>,
 	cors: Option<&Vec<String>>,
 	io: RpcHandler,
 ) -> io::Result<ws::Server> {
 	ws::ServerBuilder::with_meta_extractor(io, |context: &ws::RequestContext| Metadata::new(context.sender()))
 		.max_payload(MAX_PAYLOAD)
+		.max_connections(max_connections.unwrap_or(WS_MAX_CONNECTIONS))
 		.allowed_origins(map_cors(cors))
 		.start(addr)
 		.map_err(|err| match err {

core/service/src/components.rs

@@ -143,6 +143,7 @@ pub trait StartRPC<C: Components> {
 		system_info: SystemInfo,
 		rpc_http: Option<SocketAddr>,
 		rpc_ws: Option<SocketAddr>,
+		rpc_ws_max_connections: Option<usize>,
 		rpc_cors: Option<Vec<String>>,
 		task_executor: TaskExecutor,
 		transaction_pool: Arc<TransactionPool<C::TransactionPoolApi>>,
@@ -162,6 +163,7 @@ impl<C: Components> StartRPC<Self> for C where
 		rpc_system_info: SystemInfo,
 		rpc_http: Option<SocketAddr>,
 		rpc_ws: Option<SocketAddr>,
+		rpc_ws_max_connections: Option<usize>,
 		rpc_cors: Option<Vec<String>>,
 		task_executor: TaskExecutor,
 		transaction_pool: Arc<TransactionPool<C::TransactionPoolApi>>,
@@ -186,8 +188,19 @@ impl<C: Components> StartRPC<Self> for C where
 		};
 
 		Ok((
-			maybe_start_server(rpc_http, |address| rpc::start_http(address, rpc_cors.as_ref(), handler()))?,
-			maybe_start_server(rpc_ws, |address| rpc::start_ws(address, rpc_cors.as_ref(), handler()))?.map(Mutex::new),
+			maybe_start_server(
+				rpc_http,
+				|address| rpc::start_http(address, rpc_cors.as_ref(), handler()),
+			)?,
+			maybe_start_server(
+				rpc_ws,
+				|address| rpc::start_ws(
+					address,
+					rpc_ws_max_connections,
+					rpc_cors.as_ref(),
+					handler(),
+				),
+			)?.map(Mutex::new),
 		))
 	}
 }

core/service/src/config.rs

@@ -66,6 +66,8 @@ pub struct Configuration<C, G: Serialize + DeserializeOwned + BuildStorage> {
 	pub rpc_http: Option<SocketAddr>,
 	/// RPC over Websockets binding address. `None` if disabled.
 	pub rpc_ws: Option<SocketAddr>,
+	/// Maximum number of connections for WebSockets RPC server. `None` if default.
+	pub rpc_ws_max_connections: Option<usize>,
 	/// CORS settings for HTTP & WS servers. `None` if all origins are allowed.
 	pub rpc_cors: Option<Vec<String>>,
 	/// Telemetry service URL. `None` if disabled.
@@ -102,6 +104,7 @@ impl<C: Default, G: Serialize + DeserializeOwned + BuildStorage> Configuration<C
 			execution_strategies: Default::default(),
 			rpc_http: None,
 			rpc_ws: None,
+			rpc_ws_max_connections: None,
 			rpc_cors: Some(vec![]),
 			telemetry_endpoints: None,
 			default_heap_pages: None,

core/service/src/lib.rs

@@ -317,8 +317,16 @@ impl<Components: components::Components> Service<Components> {
 			properties: config.chain_spec.properties(),
 		};
 		let rpc = Components::RuntimeServices::start_rpc(
-			client.clone(), network.clone(), has_bootnodes, system_info, config.rpc_http,
-			config.rpc_ws, config.rpc_cors.clone(), task_executor.clone(), transaction_pool.clone(),
+			client.clone(),
+			network.clone(),
+			has_bootnodes,
+			system_info,
+			config.rpc_http,
+			config.rpc_ws,
+			config.rpc_ws_max_connections,
+			config.rpc_cors.clone(),
+			task_executor.clone(),
+			transaction_pool.clone(),
 		)?;
 
 		let telemetry_connection_sinks: Arc<Mutex<Vec<mpsc::UnboundedSender<()>>>> = Default::default();

core/service/test/src/lib.rs

@@ -121,6 +121,7 @@ fn node_config<F: ServiceFactory> (
 		execution_strategies: Default::default(),
 		rpc_http: None,
 		rpc_ws: None,
+		rpc_ws_max_connections: None,
 		rpc_cors: None,
 		telemetry_endpoints: None,
 		default_heap_pages: None,