Introduce RPC Worker (#1462)

* Extract json-rpc-client and introduce worker

* Initial rpc worker

* Add error handling

* Use bounded channels for listeners

* Improve naming and clean up

* Use tracing channels

* Improve code readability

* Decrease channel size limit

* Remove unused dependency

* Fix docs

* RPC -> Rpc

* Start worker in initialization method

* Print error in case a distribution channel is full

* Fix docs

* Make `RpcStreamWorker` private

Co-authored-by: Davide Galassi <davxy@datawok.net>

* Use tokio channels and add TODO item

* Remove `Option` from `to_worker_channel`

Co-authored-by: Davide Galassi <davxy@datawok.net>

test/service/src/lib.rs

@@ -37,7 +37,7 @@ use cumulus_client_service::{
 use cumulus_primitives_core::ParaId;
 use cumulus_relay_chain_inprocess_interface::RelayChainInProcessInterface;
 use cumulus_relay_chain_interface::{RelayChainError, RelayChainInterface, RelayChainResult};
-use cumulus_relay_chain_rpc_interface::RelayChainRPCInterface;
+use cumulus_relay_chain_rpc_interface::{create_client_and_start_worker, RelayChainRpcInterface};
 use cumulus_test_runtime::{Hash, Header, NodeBlock as Block, RuntimeApi};
 use parking_lot::Mutex;
 
@@ -182,7 +182,8 @@ async fn build_relay_chain_interface(
 	task_manager: &mut TaskManager,
 ) -> RelayChainResult<Arc<dyn RelayChainInterface + 'static>> {
 	if let Some(relay_chain_url) = collator_options.relay_chain_rpc_url {
-		return Ok(Arc::new(RelayChainRPCInterface::new(relay_chain_url).await?) as Arc<_>)
+		let client = create_client_and_start_worker(relay_chain_url, task_manager).await?;
+		return Ok(Arc::new(RelayChainRpcInterface::new(client)) as Arc<_>)
 	}
 
 	let relay_chain_full_node = polkadot_test_service::new_full(