async fn connect_next_available_rpc_server(

    urls: &[String],

    starting_position: usize,

) -> Result<(usize, Arc<JsonRpcClient>), ()> {

    tracing::debug!(target: LOG_TARGET, starting_position, "Connecting to RPC server.");

    for (counter, url) in urls

        .iter()

        .cycle()

        .skip(starting_position)

        .take(urls.len())

        .enumerate()

        let index = (starting_position + counter) % urls.len();

        tracing::info!(

            "Trying to connect to next external orchestrator node.",

        match WsClientBuilder::default().build(&url).await {

            Ok(ws_client) => return Ok((index, Arc::new(ws_client))),

            Err(err) => tracing::debug!(target: LOG_TARGET, url, ?err, "Unable to connect."),

    Err(())

}

    pub async fn new(urls: Vec<String>) -> Result<(Self, mpsc::Sender<WsClientRequest>), ()> {

        if urls.is_empty() {

            return Err(());

        }

        let (active_index, active_client) = connect_next_available_rpc_server(&urls, 0).await?;

        let (request_sender, request_receiver) = mpsc::channel(100);

        Ok((

            Self {

                urls,

                active_client,

                active_index,

                request_receiver,

                best_header_listeners: vec![],

                imported_header_listeners: vec![],

                finalized_header_listeners: vec![],

            },

            request_sender,

        ))

    }

    async fn connect_to_new_rpc_server(&mut self) -> Result<(), ()> {

        let (active_index, active_client) =

            connect_next_available_rpc_server(&self.urls, self.active_index + 1).await?;

        self.active_index = active_index;

        self.active_client = active_client;

        Ok(())

    }

    fn send_request(

        &self,

        JsonRpcRequest {

            method,

            params,

            response_sender,

        }: JsonRpcRequest,

    ) -> RpcRequestFuture {

        let client = self.active_client.clone();

        async move {

            let response = client.request(&method, params.clone()).await;

            if let Err(JsonRpseeError::RestartNeeded(_)) = response {

                return Err(JsonRpcRequest {

                    method,

                    params,

                    response_sender,

                });

            }

            if let Err(err) = response_sender.send(response) {

                tracing::debug!(

                    "Recipient no longer interested in request result"

            }

            Ok(())

        }

        .boxed()

    }

    async fn get_subscriptions(&self) -> Result<OrchestratorSubscription, JsonRpseeError> {

        let import_subscription = <JsonRpcClient as ChainApiClient<

            dp_core::BlockNumber,

            dp_core::Hash,

            dp_core::Header,

            dp_core::SignedBlock,

        >>::subscribe_all_heads(&self.active_client)

        .await

        .map_err(|e| {

            tracing::error!(

                "Unable to open `chain_subscribeAllHeads` subscription."

            e

        })?;

        let best_subscription = <JsonRpcClient as ChainApiClient<

            dp_core::BlockNumber,

            dp_core::Hash,

            dp_core::Header,

            dp_core::SignedBlock,

        >>::subscribe_new_heads(&self.active_client)

        .await

        .map_err(|e| {

            tracing::error!(

                "Unable to open `chain_subscribeNewHeads` subscription."

            e

        })?;

        let finalized_subscription = <JsonRpcClient as ChainApiClient<

            dp_core::BlockNumber,

            dp_core::Hash,

            dp_core::Header,

            dp_core::SignedBlock,

        >>::subscribe_finalized_heads(&self.active_client)

        .await

        .map_err(|e| {

            tracing::error!(

                "Unable to open `chain_subscribeFinalizedHeads` subscription."

            e

        })?;

        Ok(OrchestratorSubscription {

            import_subscription,

            best_subscription,

            finalized_subscription,

        })

    }

    async fn handle_reconnect(

        &mut self,

        pending_requests: &mut FuturesUnordered<RpcRequestFuture>,

        first_failed_request: Option<JsonRpcRequest>,

    ) -> Result<(), String> {

        let mut requests_to_retry = Vec::new();

        if let Some(req) = first_failed_request {

            requests_to_retry.push(req)

        }

        while !pending_requests.is_empty() {

            if let Some(Err(req)) = pending_requests.next().await {

                requests_to_retry.push(req);

            }

        if self.connect_to_new_rpc_server().await.is_err() {

            return Err("Unable to find valid external RPC server, shutting down.".to_string());

        }

        for req in requests_to_retry.into_iter() {

            pending_requests.push(self.send_request(req));

        }

        self.get_subscriptions().await.map_err(|e| {

			format!("Not able to create streams from newly connected RPC server, shutting down. err: {:?}", e)

		})?;

        Ok(())

    }

    pub async fn run(mut self) {

        let mut pending_requests = FuturesUnordered::new();

        let mut connection_status = ConnectionStatus::Connected;

        let Ok(mut subscriptions) = self.get_subscriptions().await else {

            tracing::error!(target: LOG_TARGET, "Unable to fetch subscriptions on initial connection.");

            return;

        let mut imported_blocks_cache = LruMap::new(ByLength::new(40));

        let mut last_seen_finalized_num: dp_core::BlockNumber = 0;

            if let ConnectionStatus::Disconnected { failed_request } = connection_status {

                if let Err(message) = self

                    .handle_reconnect(&mut pending_requests, failed_request)

                    .await

                    tracing::error!(

                        "Unable to reconnect, stopping worker."

                    return;

                }

                connection_status = ConnectionStatus::Connected;

            }

            tokio::select! {

                req = self.request_receiver.recv() => match req {

                    Some(WsClientRequest::JsonRpcRequest(req)) => {

                        pending_requests.push(self.send_request(req));

                    },

                    Some(WsClientRequest::RegisterBestHeadListener(tx)) => {

                        self.best_header_listeners.push(tx);

                    },

                    Some(WsClientRequest::RegisterImportListener(tx)) => {

                        self.imported_header_listeners.push(tx);

                    },

                    Some(WsClientRequest::RegisterFinalizationListener(tx)) => {

                        self.finalized_header_listeners.push(tx);

                    },

                        tracing::error!(target: LOG_TARGET, "RPC client receiver closed. Stopping RPC Worker.");

                        return;

                pending = pending_requests.next(), if !pending_requests.is_empty() => {

                    if let Some(Err(req)) = pending {

                        connection_status = ConnectionStatus::Disconnected { failed_request: Some(req) };

                    }

                import_event = subscriptions.import_subscription.next() => {

                    match import_event {

                        Some(Ok(header)) => {

                            let hash = header.hash();

                            if imported_blocks_cache.peek(&hash).is_some() {

                                tracing::debug!(

                                    "Duplicate imported block header. This might happen after switching to a new RPC node. Skipping distribution."

                                continue;

                            }

                            imported_blocks_cache.insert(hash, ());

                            distribute(header, &mut self.imported_header_listeners);

                            tracing::error!(target: LOG_TARGET, "Subscription closed.");

                            connection_status = ConnectionStatus::Disconnected { failed_request: None};

                        Some(Err(error)) => {

                            tracing::error!(target: LOG_TARGET, ?error, "Error in RPC subscription.");

                            connection_status = ConnectionStatus::Disconnected { failed_request: None};

                best_header_event = subscriptions.best_subscription.next() => {

                    match best_header_event {

                        Some(Ok(header)) => distribute(header, &mut self.best_header_listeners),

                            tracing::error!(target: LOG_TARGET, "Subscription closed.");

                            connection_status = ConnectionStatus::Disconnected { failed_request: None};

                        Some(Err(error)) => {

                            tracing::error!(target: LOG_TARGET, ?error, "Error in RPC subscription.");

                            connection_status = ConnectionStatus::Disconnected { failed_request: None};

                finalized_event = subscriptions.finalized_subscription.next() => {

                    match finalized_event {

                        Some(Ok(header)) if header.number > last_seen_finalized_num => {

                            last_seen_finalized_num = header.number;

                            distribute(header, &mut self.finalized_header_listeners);

                        },

                        Some(Ok(header)) => {

                            tracing::debug!(

                                "Duplicate finalized block header. This might happen after switching to a new RPC node. Skipping distribution."

                            tracing::error!(target: LOG_TARGET, "Subscription closed.");

                            connection_status = ConnectionStatus::Disconnected { failed_request: None};

                        Some(Err(error)) => {

                            tracing::error!(target: LOG_TARGET, ?error, "Error in RPC subscription.");

                            connection_status = ConnectionStatus::Disconnected { failed_request: None};

    }

pub fn distribute<T: Clone + Send>(value: T, senders: &mut Vec<mpsc::Sender<T>>) {

    senders.retain_mut(|e| {

        match e.try_send(value.clone()) {

            Err(mpsc::error::TrySendError::Closed(_)) => false,

            Err(error) => {

                tracing::error!(target: LOG_TARGET, ?error, "Event distribution channel has reached its limit. This can lead to missed notifications.");

                true

            _ => true,

    });

}