pub(in crate::nats_client) async fn try_connect(opts: NatsClientOptions, cluster_uris: &Vec<String>, keep_retrying: bool) -> Result<TcpStream, RatsioError> {
        let valid_addresses = cluster_uris.iter().flat_map(|raw_uri| {
            let uri = if raw_uri.starts_with("nats://") {
                (&raw_uri[7..]).to_string()
            } else {
                raw_uri.clone()
            };
            match uri.parse::<SocketAddr>() {
                Ok(addr) => Some((uri, addr)).into_iter().collect::<Vec<_>>(),
                Err(_err) => {
                    match uri.to_socket_addrs() {
                        Ok(ips_iter) => ips_iter.map(|x| {
                            (uri.clone(), x)
                        }).collect::<Vec<_>>(),
                        Err(err) => {
                            error!("Unable resolve url => {} to ip address => {}", &uri, err);
                            Default::default()
                        }
                    }
                }
            }
        }).collect::<Vec<_>>();
        if valid_addresses.len() == 0 {
            return Err(RatsioError::GenericError("No valid NATS uris".into()));
        }
        loop {
            for uri_and_addr in valid_addresses.clone() {
                let (uri, addr) = uri_and_addr;
                match tokio::net::TcpStream::connect(&addr).await {
                    Ok(tcp_stream) => return Ok(tcp_stream),
                    Err(err) => {
                        error!("Error connecting to {} - {:?}", uri, err);
                    }
                }
            }
            error!("Unable to connect to any of the Nats servers, will retry again.");
            if keep_retrying {
                let _ = Delay::new(Duration::from_millis(opts.reconnect_timeout)).await;
            }else{
                return Err(RatsioError::NoRouteToHostError);
            }
        }
    }
//Establish tcp connection with one of the Nats servers
// Process - go through uris in opts.cluster_uris, trying one at a time.
//     If none of attempted connections succeed, wait and try again.
Establish tcp connection with one of the Nats servers
Process - go through uris in opts.cluster_uris, trying one at a time.
If none of attempted connections succeed, wait and try again.
Establish tcp connection with one of the Nats servers Process - go through uris in opts . cluster_uris trying one at a time . If none of attempted connections succeed wait and try again .
pub(in crate::nats_client) async fn try_connect(opts: NatsClientOptions, cluster_uris: &Vec<String>, keep_retrying: bool) -> Result<TcpStream, RatsioError> {
        let valid_addresses = cluster_uris.iter().flat_map(|raw_uri| {
            let uri = if raw_uri.starts_with("nats://") {
                (&raw_uri[7..]).to_string()
            } else {
                raw_uri.clone()
            };
            match uri.parse::<SocketAddr>() {
                Ok(addr) => Some((uri, addr)).into_iter().collect::<Vec<_>>(),
                Err(_err) => {
                    match uri.to_socket_addrs() {
                        Ok(ips_iter) => ips_iter.map(|x| {
                            (uri.clone(), x)
                        }).collect::<Vec<_>>(),
                        Err(err) => {
                            error!("Unable resolve url => {} to ip address => {}", &uri, err);
                            Default::default()
                        }
                    }
                }
            }
        }).collect::<Vec<_>>();
        if valid_addresses.len() == 0 {
            return Err(RatsioError::GenericError("No valid NATS uris".into()));
        }
        loop {
            for uri_and_addr in valid_addresses.clone() {
                let (uri, addr) = uri_and_addr;
                match tokio::net::TcpStream::connect(&addr).await {
                    Ok(tcp_stream) => return Ok(tcp_stream),
                    Err(err) => {
                        error!("Error connecting to {} - {:?}", uri, err);
                    }
                }
            }
            error!("Unable to connect to any of the Nats servers, will retry again.");
            if keep_retrying {
                let _ = Delay::new(Duration::from_millis(opts.reconnect_timeout)).await;
            }else{
                return Err(RatsioError::NoRouteToHostError);
            }
        }
    }
pub ( in crate :: nats_client ) async fn try_connect ( opts : NatsClientOptions , cluster_uris : & Vec < String > , keep_retrying : bool ) -> Result < TcpStream , RatsioError > { let valid_addresses = cluster_uris . iter ( ) . flat_map ( | raw_uri | { let uri = if raw_uri . starts_with ( " " ) { ( & raw_uri [ 7 .. ] ) . to_string ( ) } else { raw_uri . clone ( ) } ; match uri . parse :: < SocketAddr > ( ) { Ok ( addr ) => Some ( ( uri , addr ) ) . into_iter ( ) . collect :: < Vec < _ > > ( ) , Err ( _err ) => { match uri . to_socket_addrs ( ) { Ok ( ips_iter ) => ips_iter . map ( | x | { ( uri . clone ( ) , x ) } ) . collect :: < Vec < _ > > ( ) , Err ( err ) => { error ! ( " " uri err ) ; Default :: default ( ) } } } } } ) . collect :: < Vec < _ > > ( ) ; if valid_addresses . len ( ) == 0 { return Err ( RatsioError :: GenericError ( " " . into ( ) ) ) ; } loop { for uri_and_addr in valid_addresses . clone ( ) { let ( uri , addr ) = uri_and_addr ; match tokio :: net :: TcpStream :: connect ( & addr ) . await { Ok ( tcp_stream ) => return Ok ( tcp_stream ) , Err ( err ) => { error ! ( " " uri err ) ; } } } error ! ( " " ) ; if keep_retrying { let _ = Delay :: new ( Duration :: from_millis ( opts . reconnect_timeout ) ) . await ; } else { return Err ( RatsioError :: NoRouteToHostError ) ; } } }
Establish tcp connection with one of the Nats servers
Process - go through uris in opts.cluster_uris, trying one at a time.
Establish tcp connection with one of the Nats servers Process - go through uris in opts . cluster_uris trying one at a time .
opts
NatsClientOptions
cluster_uris
&Vec<String>
keep_retrying
bool
opts
NatsClientOptions
cluster_uris
&Vec<String>
keep_retrying
bool