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

* core: allow setting max ws rpc connections

* style: break long lines

* core: fix service tests

substrate/core/cli/src/lib.rs

@@ -487,6 +487,7 @@ 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

substrate/core/cli/src/params.rs

@@ -333,6 +333,10 @@ 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.

substrate/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 {

substrate/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),
 		))
 	}
 }
@@ -330,7 +343,7 @@ pub trait ServiceFactory: 'static + Sized {
 	/// Build the Fork Choice algorithm for full client
 	fn build_select_chain(
 		config: &mut FactoryFullConfiguration<Self>,
-		client: Arc<FullClient<Self>>, 
+		client: Arc<FullClient<Self>>,
 	) -> Result<Self::SelectChain, error::Error>;
 
 	/// Build full service.
@@ -497,7 +510,7 @@ impl<Factory: ServiceFactory> Components for FullComponents<Factory> {
 	}
 
 	fn build_transaction_pool(
-		config: TransactionPoolOptions, 
+		config: TransactionPoolOptions,
 		client: Arc<ComponentClient<Self>>
 	) -> Result<TransactionPool<Self::TransactionPoolApi>, error::Error> {
 		Factory::build_full_transaction_pool(config, client)

substrate/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.
@@ -104,6 +106,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,

substrate/core/service/src/lib.rs

@@ -326,8 +326,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();

substrate/core/service/test/src/lib.rs

@@ -120,6 +120,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,