async fn connect_to_server(
        local_addr: &SockAddr,
        peer_addr: &SockAddr,
        canceler: &AtomicBool,
        log: &ReplicaLogger,
    ) -> Result<TcpStream, TransportErrorCode> {
        let mut retries: i64 = 0;
        let client_socket = match Self::init_client_socket(local_addr, log) {
            Ok(client_socket) => client_socket,
            Err(e) => return Err(e),
        };
        let addr = Self::get_socket_addr(peer_addr);
        let tcp_stream = client_socket.into_tcp_stream();
        let connect = TcpStream::connect_std(tcp_stream, &addr);
        let mut connect = connect.boxed();
        loop {
            retries += 1;
            if canceler.load(Ordering::SeqCst) {
                return Err(TransportErrorCode::ConnectOsError);
            }
            let timeout = delay_for(Duration::from_millis(CANCEL_CHECK_PERIOD_MILLISECONDS));
            match future::select(connect, timeout).await {
                // Notify the async completion handler
                Either::Left((result, _)) => {
                    match Self::connect_status(result) {
                        ConnectStatus::Success(stream) => {
                            return Self::set_send_sockopts(stream, log);
                        }
                        ConnectStatus::ServerDown => return Err(TransportErrorCode::ServerDown),
                        ConnectStatus::Error(err_code) => {
                            warn!(
                                log,
                                "ControlPlane::connect_to_server(): local_addr = {:?} peer_addr = {:?}, error {:?}, retries = {}",
                                local_addr,
                                peer_addr,
                                err_code,
                                retries,
                            );
                            return Err(TransportErrorCode::ConnectOsError);
                        }
                    };
                }
                // Cancel check period timeout
                Either::Right((_, next_connect)) => {
                    if canceler.load(Ordering::SeqCst) {
                        return Err(TransportErrorCode::ConnectOsError);
                    }
                    connect = next_connect;
                }
            };
        }
    }
/// Set up the client socket, and connect to the specified server peer
Set up the client socket, and connect to the specified server peer
Set up the client socket and connect to the specified server peer
async fn connect_to_server(
        local_addr: &SockAddr,
        peer_addr: &SockAddr,
        canceler: &AtomicBool,
        log: &ReplicaLogger,
    ) -> Result<TcpStream, TransportErrorCode> {
        let mut retries: i64 = 0;
        let client_socket = match Self::init_client_socket(local_addr, log) {
            Ok(client_socket) => client_socket,
            Err(e) => return Err(e),
        };
        let addr = Self::get_socket_addr(peer_addr);
        let tcp_stream = client_socket.into_tcp_stream();
        let connect = TcpStream::connect_std(tcp_stream, &addr);
        let mut connect = connect.boxed();
        loop {
            retries += 1;
            if canceler.load(Ordering::SeqCst) {
                return Err(TransportErrorCode::ConnectOsError);
            }
            let timeout = delay_for(Duration::from_millis(CANCEL_CHECK_PERIOD_MILLISECONDS));
            match future::select(connect, timeout).await {
                // Notify the async completion handler
                Either::Left((result, _)) => {
                    match Self::connect_status(result) {
                        ConnectStatus::Success(stream) => {
                            return Self::set_send_sockopts(stream, log);
                        }
                        ConnectStatus::ServerDown => return Err(TransportErrorCode::ServerDown),
                        ConnectStatus::Error(err_code) => {
                            warn!(
                                log,
                                "ControlPlane::connect_to_server(): local_addr = {:?} peer_addr = {:?}, error {:?}, retries = {}",
                                local_addr,
                                peer_addr,
                                err_code,
                                retries,
                            );
                            return Err(TransportErrorCode::ConnectOsError);
                        }
                    };
                }
                // Cancel check period timeout
                Either::Right((_, next_connect)) => {
                    if canceler.load(Ordering::SeqCst) {
                        return Err(TransportErrorCode::ConnectOsError);
                    }
                    connect = next_connect;
                }
            };
        }
    }
async fn connect_to_server ( local_addr : & SockAddr , peer_addr : & SockAddr , canceler : & AtomicBool , log : & ReplicaLogger , ) -> Result < TcpStream , TransportErrorCode > { let mut retries : i64 = 0 ; let client_socket = match Self :: init_client_socket ( local_addr , log ) { Ok ( client_socket ) => client_socket , Err ( e ) => return Err ( e ) , } ; let addr = Self :: get_socket_addr ( peer_addr ) ; let tcp_stream = client_socket . into_tcp_stream ( ) ; let connect = TcpStream :: connect_std ( tcp_stream , & addr ) ; let mut connect = connect . boxed ( ) ; loop { retries += 1 ; if canceler . load ( Ordering :: SeqCst ) { return Err ( TransportErrorCode :: ConnectOsError ) ; } let timeout = delay_for ( Duration :: from_millis ( CANCEL_CHECK_PERIOD_MILLISECONDS ) ) ; match future :: select ( connect , timeout ) . await { // Notify the async completion handler Either :: Left ( ( result , _ ) ) => { match Self :: connect_status ( result ) { ConnectStatus :: Success ( stream ) => { return Self :: set_send_sockopts ( stream , log ) ; } ConnectStatus :: ServerDown => return Err ( TransportErrorCode :: ServerDown ) , ConnectStatus :: Error ( err_code ) => { warn ! ( log " " local_addr peer_addr err_code retries ) ; return Err ( TransportErrorCode :: ConnectOsError ) ; } } ; } // Cancel check period timeout Either :: Right ( ( _ , next_connect ) ) => { if canceler . load ( Ordering :: SeqCst ) { return Err ( TransportErrorCode :: ConnectOsError ) ; } connect = next_connect ; } } ; } }
Set up the client socket, and connect to the specified server peer
Set up the client socket and connect to the specified server peer
local_addr
&SockAddr
peer_addr
&SockAddr
canceler
&AtomicBool
log
&ReplicaLogger
local_addr
&SockAddr
peer_addr
&SockAddr
canceler
&AtomicBool
log
&ReplicaLogger