fn use_keyspace(
        &mut self,
        keyspace_name: &VerifiedKeyspaceName,
        response_sender: tokio::sync::oneshot::Sender<Result<(), QueryError>>,
    ) {
        self.current_keyspace = Some(keyspace_name.clone());

        let mut conns = self.conns.clone();
        let keyspace_name = keyspace_name.clone();
        let address = self.address;

        let fut = async move {
            let mut use_keyspace_futures = Vec::new();

            for shard_conns in conns.iter_mut() {
                for conn in shard_conns.iter_mut() {
                    let fut = conn.use_keyspace(&keyspace_name);
                    use_keyspace_futures.push(fut);
                }
            }

            if use_keyspace_futures.is_empty() {
                return Ok(());
            }

            let use_keyspace_results: Vec<Result<(), QueryError>> =
                futures::future::join_all(use_keyspace_futures).await;

            // If there was at least one Ok and the rest were IoErrors we can return Ok
            // keyspace name is correct and will be used on broken connection on the next reconnect

            // If there were only IoErrors then return IoError
            // If there was an error different than IoError return this error - something is wrong

            let mut was_ok: bool = false;
            let mut io_error: Option<Arc<std::io::Error>> = None;

            for result in use_keyspace_results {
                match result {
                    Ok(()) => was_ok = true,
                    Err(err) => match err {
                        QueryError::IoError(io_err) => io_error = Some(io_err),
                        _ => return Err(err),
                    },
                }
            }

            if was_ok {
                return Ok(());
            }

            // We can unwrap io_error because use_keyspace_futures must be nonempty
            Err(QueryError::IoError(io_error.unwrap()))
        };

        tokio::task::spawn(async move {
            let res = fut.await;
            match &res {
                Ok(()) => debug!("[{}] Successfully changed current keyspace", address),
                Err(err) => warn!("[{}] Failed to change keyspace: {:?}", address, err),
            }
            let _ = response_sender.send(res);
        });
    }
// Sets current keyspace for available connections.
// Connections which are being currently opened and future connections
// will have this keyspace set when they appear on `ready_connections`.
// Sends response to the `response_sender` when all current connections
// have their keyspace set.
Sets current keyspace for available connections.
Connections which are being currently opened and future connections
will have this keyspace set when they appear on `ready_connections`.
Sends response to the `response_sender` when all current connections
have their keyspace set.
Sets current keyspace for available connections . Connections which are being currently opened and future connections will have this keyspace set when they appear on ` ready_connections ` . Sends response to the ` response_sender ` when all current connections have their keyspace set .
fn use_keyspace(
        &mut self,
        keyspace_name: &VerifiedKeyspaceName,
        response_sender: tokio::sync::oneshot::Sender<Result<(), QueryError>>,
    ) {
        self.current_keyspace = Some(keyspace_name.clone());
        let mut conns = self.conns.clone();
        let keyspace_name = keyspace_name.clone();
        let address = self.address;
        let fut = async move {
            let mut use_keyspace_futures = Vec::new();
            for shard_conns in conns.iter_mut() {
                for conn in shard_conns.iter_mut() {
                    let fut = conn.use_keyspace(&keyspace_name);
                    use_keyspace_futures.push(fut);
                }
            }
            if use_keyspace_futures.is_empty() {
                return Ok(());
            }
            let use_keyspace_results: Vec<Result<(), QueryError>> =
                futures::future::join_all(use_keyspace_futures).await;
            // If there was at least one Ok and the rest were IoErrors we can return Ok
            // keyspace name is correct and will be used on broken connection on the next reconnect
            // If there were only IoErrors then return IoError
            // If there was an error different than IoError return this error - something is wrong
            let mut was_ok: bool = false;
            let mut io_error: Option<Arc<std::io::Error>> = None;
            for result in use_keyspace_results {
                match result {
                    Ok(()) => was_ok = true,
                    Err(err) => match err {
                        QueryError::IoError(io_err) => io_error = Some(io_err),
                        _ => return Err(err),
                    },
                }
            }
            if was_ok {
                return Ok(());
            }
            // We can unwrap io_error because use_keyspace_futures must be nonempty
            Err(QueryError::IoError(io_error.unwrap()))
        };
        tokio::task::spawn(async move {
            let res = fut.await;
            match &res {
                Ok(()) => debug!("[{}] Successfully changed current keyspace", address),
                Err(err) => warn!("[{}] Failed to change keyspace: {:?}", address, err),
            }
            let _ = response_sender.send(res);
        });
    }
fn use_keyspace ( & mut self , keyspace_name : & VerifiedKeyspaceName , response_sender : tokio :: sync :: oneshot :: Sender < Result < ( ) , QueryError > > , ) { self . current_keyspace = Some ( keyspace_name . clone ( ) ) ; let mut conns = self . conns . clone ( ) ; let keyspace_name = keyspace_name . clone ( ) ; let address = self . address ; let fut = async move { let mut use_keyspace_futures = Vec :: new ( ) ; for shard_conns in conns . iter_mut ( ) { for conn in shard_conns . iter_mut ( ) { let fut = conn . use_keyspace ( & keyspace_name ) ; use_keyspace_futures . push ( fut ) ; } } if use_keyspace_futures . is_empty ( ) { return Ok ( ( ) ) ; } let use_keyspace_results : Vec < Result < ( ) , QueryError > > = futures :: future :: join_all ( use_keyspace_futures ) . await ; // If there was at least one Ok and the rest were IoErrors we can return Ok // keyspace name is correct and will be used on broken connection on the next reconnect // If there were only IoErrors then return IoError // If there was an error different than IoError return this error - something is wrong let mut was_ok : bool = false ; let mut io_error : Option < Arc < std :: io :: Error > > = None ; for result in use_keyspace_results { match result { Ok ( ( ) ) => was_ok = true , Err ( err ) => match err { QueryError :: IoError ( io_err ) => io_error = Some ( io_err ) , _ => return Err ( err ) , } , } } if was_ok { return Ok ( ( ) ) ; } // We can unwrap io_error because use_keyspace_futures must be nonempty Err ( QueryError :: IoError ( io_error . unwrap ( ) ) ) } ; tokio :: task :: spawn ( async move { let res = fut . await ; match & res { Ok ( ( ) ) => debug ! ( " " address ) , Err ( err ) => warn ! ( " " address err ) , } let _ = response_sender . send ( res ) ; } ) ; }
Sets current keyspace for available connections.
Sets current keyspace for available connections .
&mut self
keyspace_name
&VerifiedKeyspaceName
response_sender
tokio::sync::oneshot::Sender<Result<(), QueryError>>
&mut self
keyspace_name
&VerifiedKeyspaceName
response_sender
tokio::sync::oneshot::Sender<Result<(), QueryError>>