fn start_peer(
        &self,
        client_type: TransportClientType,
        peer_id: &NodeId,
        peer_record: &NodeRecord,
        client_state: &mut ClientState,
    ) -> Result<(), TransportErrorCode> {
        if client_state.peer_map.get(&peer_id).is_some() {
            return Err(TransportErrorCode::PeerAlreadyRegistered);
        }

        let is_peer_server = Self::is_peer_server(&self.node_id, &peer_id);
        let mut peer_state = PeerState {
            flow_map: HashMap::new(),
            connect_cancelers: Vec::new(),
        };

        // TODO: P2P-514
        let mut queue_size_map = HashMap::new();
        let flow_ips = get_flow_ips(peer_record)?;
        for flow_config in &self.config.p2p_flows {
            let flow_tag = FlowTag::from(flow_config.flow_tag);
            queue_size_map.insert(flow_tag, QueueSize::from(flow_config.queue_size));
            if !is_peer_server {
                let peer_ip = flow_ips
                    .get(&flow_tag)
                    .map_or("Unknown Peer IP".to_string(), |x| x.to_string());
                let flow_label = get_flow_label(&peer_ip, peer_id);
                let flow_id = FlowId {
                    client_type,
                    peer_id: *peer_id,
                    flow_tag,
                };
                let flow_state = FlowState {
                    flow_id,
                    flow_tag_label: flow_config.flow_tag.to_string(),
                    flow_label: flow_label.clone(),
                    connection_state: ConnectionState::Listening,
                    abort_handles: Option::None,
                    send_queue: Box::new(SendQueueImpl::new(
                        flow_label,
                        &flow_tag,
                        QueueSize::from(flow_config.queue_size),
                        self.send_queue_metrics.clone(),
                    )),
                };
                self.report_connection_state(&flow_state);
                peer_state.flow_map.insert(flow_tag, flow_state);
            }
        }
        if !is_peer_server {
            client_state.peer_map.insert(*peer_id, peer_state);
            return Ok(());
        }

        for flow_endpoint in &peer_record.p2p_flow_endpoints {
            let connect_canceler = Arc::new(AtomicBool::new(false));
            let endpoint = match &flow_endpoint.endpoint {
                Some(x) => x,
                None => {
                    warn!(
                        self.log,
                        "ControlPlane::start_peer(): missing endpoint flow = {:?}",
                        flow_endpoint.flow_tag
                    );
                    continue;
                }
            };

            let flow_tag = FlowTag::from(flow_endpoint.flow_tag);
            let queue_size = match queue_size_map.get(&flow_tag) {
                Some(queue_size) => queue_size,
                None => {
                    error!(
                        self.log,
                        "ControlPlane::start_peer(): TransportConfig NodeRecord mismatch = {:?}",
                        flow_endpoint.flow_tag
                    );
                    continue;
                }
            };

            let peer_ip = IpAddr::from_str(endpoint.ip_addr.as_str())
                .unwrap_or_else(|_| panic!("Invalid node IP: {}", endpoint.ip_addr));
            let flow_label = get_flow_label(&endpoint.ip_addr.as_str(), peer_id);
            let server_port = endpoint.port as u16;
            self.spawn_connect_task(
                client_type,
                flow_endpoint.flow_tag.into(),
                *peer_id,
                peer_ip,
                ServerPort::from(server_port),
                connect_canceler.clone(),
            );
            let flow_id = FlowId {
                client_type,
                peer_id: *peer_id,
                flow_tag,
            };
            let flow_state = FlowState {
                flow_id,
                flow_tag_label: flow_endpoint.flow_tag.to_string(),
                flow_label: flow_label.clone(),
                connection_state: ConnectionState::Connecting(SocketAddr::new(
                    peer_ip,
                    server_port,
                )),
                abort_handles: Option::None,
                send_queue: Box::new(SendQueueImpl::new(
                    flow_label.clone(),
                    &flow_tag,
                    *queue_size,
                    self.send_queue_metrics.clone(),
                )),
            };
            self.report_connection_state(&flow_state);
            peer_state
                .flow_map
                .insert(flow_endpoint.flow_tag.into(), flow_state);
            peer_state.connect_cancelers.push(connect_canceler);
        }

        client_state.peer_map.insert(*peer_id, peer_state);
        Ok(())
    }
/// Starts all connections to a peer and initializes the corresponding data
/// structures and tasks
Starts all connections to a peer and initializes the corresponding data
structures and tasks
Starts all connections to a peer and initializes the corresponding data structures and tasks
fn start_peer(
        &self,
        client_type: TransportClientType,
        peer_id: &NodeId,
        peer_record: &NodeRecord,
        client_state: &mut ClientState,
    ) -> Result<(), TransportErrorCode> {
        if client_state.peer_map.get(&peer_id).is_some() {
            return Err(TransportErrorCode::PeerAlreadyRegistered);
        }
        let is_peer_server = Self::is_peer_server(&self.node_id, &peer_id);
        let mut peer_state = PeerState {
            flow_map: HashMap::new(),
            connect_cancelers: Vec::new(),
        };
        // TODO: P2P-514
        let mut queue_size_map = HashMap::new();
        let flow_ips = get_flow_ips(peer_record)?;
        for flow_config in &self.config.p2p_flows {
            let flow_tag = FlowTag::from(flow_config.flow_tag);
            queue_size_map.insert(flow_tag, QueueSize::from(flow_config.queue_size));
            if !is_peer_server {
                let peer_ip = flow_ips
                    .get(&flow_tag)
                    .map_or("Unknown Peer IP".to_string(), |x| x.to_string());
                let flow_label = get_flow_label(&peer_ip, peer_id);
                let flow_id = FlowId {
                    client_type,
                    peer_id: *peer_id,
                    flow_tag,
                };
                let flow_state = FlowState {
                    flow_id,
                    flow_tag_label: flow_config.flow_tag.to_string(),
                    flow_label: flow_label.clone(),
                    connection_state: ConnectionState::Listening,
                    abort_handles: Option::None,
                    send_queue: Box::new(SendQueueImpl::new(
                        flow_label,
                        &flow_tag,
                        QueueSize::from(flow_config.queue_size),
                        self.send_queue_metrics.clone(),
                    )),
                };
                self.report_connection_state(&flow_state);
                peer_state.flow_map.insert(flow_tag, flow_state);
            }
        }
        if !is_peer_server {
            client_state.peer_map.insert(*peer_id, peer_state);
            return Ok(());
        }
        for flow_endpoint in &peer_record.p2p_flow_endpoints {
            let connect_canceler = Arc::new(AtomicBool::new(false));
            let endpoint = match &flow_endpoint.endpoint {
                Some(x) => x,
                None => {
                    warn!(
                        self.log,
                        "ControlPlane::start_peer(): missing endpoint flow = {:?}",
                        flow_endpoint.flow_tag
                    );
                    continue;
                }
            };
            let flow_tag = FlowTag::from(flow_endpoint.flow_tag);
            let queue_size = match queue_size_map.get(&flow_tag) {
                Some(queue_size) => queue_size,
                None => {
                    error!(
                        self.log,
                        "ControlPlane::start_peer(): TransportConfig NodeRecord mismatch = {:?}",
                        flow_endpoint.flow_tag
                    );
                    continue;
                }
            };
            let peer_ip = IpAddr::from_str(endpoint.ip_addr.as_str())
                .unwrap_or_else(|_| panic!("Invalid node IP: {}", endpoint.ip_addr));
            let flow_label = get_flow_label(&endpoint.ip_addr.as_str(), peer_id);
            let server_port = endpoint.port as u16;
            self.spawn_connect_task(
                client_type,
                flow_endpoint.flow_tag.into(),
                *peer_id,
                peer_ip,
                ServerPort::from(server_port),
                connect_canceler.clone(),
            );
            let flow_id = FlowId {
                client_type,
                peer_id: *peer_id,
                flow_tag,
            };
            let flow_state = FlowState {
                flow_id,
                flow_tag_label: flow_endpoint.flow_tag.to_string(),
                flow_label: flow_label.clone(),
                connection_state: ConnectionState::Connecting(SocketAddr::new(
                    peer_ip,
                    server_port,
                )),
                abort_handles: Option::None,
                send_queue: Box::new(SendQueueImpl::new(
                    flow_label.clone(),
                    &flow_tag,
                    *queue_size,
                    self.send_queue_metrics.clone(),
                )),
            };
            self.report_connection_state(&flow_state);
            peer_state
                .flow_map
                .insert(flow_endpoint.flow_tag.into(), flow_state);
            peer_state.connect_cancelers.push(connect_canceler);
        }
        client_state.peer_map.insert(*peer_id, peer_state);
        Ok(())
    }
fn start_peer ( & self , client_type : TransportClientType , peer_id : & NodeId , peer_record : & NodeRecord , client_state : & mut ClientState , ) -> Result < ( ) , TransportErrorCode > { if client_state . peer_map . get ( & peer_id ) . is_some ( ) { return Err ( TransportErrorCode :: PeerAlreadyRegistered ) ; } let is_peer_server = Self :: is_peer_server ( & self . node_id , & peer_id ) ; let mut peer_state = PeerState { flow_map : HashMap :: new ( ) , connect_cancelers : Vec :: new ( ) , } ; // TODO: P2P-514 let mut queue_size_map = HashMap :: new ( ) ; let flow_ips = get_flow_ips ( peer_record ) ? ; for flow_config in & self . config . p2p_flows { let flow_tag = FlowTag :: from ( flow_config . flow_tag ) ; queue_size_map . insert ( flow_tag , QueueSize :: from ( flow_config . queue_size ) ) ; if ! is_peer_server { let peer_ip = flow_ips . get ( & flow_tag ) . map_or ( " " . to_string ( ) , | x | x . to_string ( ) ) ; let flow_label = get_flow_label ( & peer_ip , peer_id ) ; let flow_id = FlowId { client_type , peer_id : * peer_id , flow_tag , } ; let flow_state = FlowState { flow_id , flow_tag_label : flow_config . flow_tag . to_string ( ) , flow_label : flow_label . clone ( ) , connection_state : ConnectionState :: Listening , abort_handles : Option :: None , send_queue : Box :: new ( SendQueueImpl :: new ( flow_label , & flow_tag , QueueSize :: from ( flow_config . queue_size ) , self . send_queue_metrics . clone ( ) , ) ) , } ; self . report_connection_state ( & flow_state ) ; peer_state . flow_map . insert ( flow_tag , flow_state ) ; } } if ! is_peer_server { client_state . peer_map . insert ( * peer_id , peer_state ) ; return Ok ( ( ) ) ; } for flow_endpoint in & peer_record . p2p_flow_endpoints { let connect_canceler = Arc :: new ( AtomicBool :: new ( false ) ) ; let endpoint = match & flow_endpoint . endpoint { Some ( x ) => x , None => { warn ! ( self log " " flow_endpoint flow_tag ) ; continue ; } } ; let flow_tag = FlowTag :: from ( flow_endpoint . flow_tag ) ; let queue_size = match queue_size_map . get ( & flow_tag ) { Some ( queue_size ) => queue_size , None => { error ! ( self log " " flow_endpoint flow_tag ) ; continue ; } } ; let peer_ip = IpAddr :: from_str ( endpoint . ip_addr . as_str ( ) ) . unwrap_or_else ( | _ | panic ! ( " " endpoint ip_addr ) ) ; let flow_label = get_flow_label ( & endpoint . ip_addr . as_str ( ) , peer_id ) ; let server_port = endpoint . port as u16 ; self . spawn_connect_task ( client_type , flow_endpoint . flow_tag . into ( ) , * peer_id , peer_ip , ServerPort :: from ( server_port ) , connect_canceler . clone ( ) , ) ; let flow_id = FlowId { client_type , peer_id : * peer_id , flow_tag , } ; let flow_state = FlowState { flow_id , flow_tag_label : flow_endpoint . flow_tag . to_string ( ) , flow_label : flow_label . clone ( ) , connection_state : ConnectionState :: Connecting ( SocketAddr :: new ( peer_ip , server_port , ) ) , abort_handles : Option :: None , send_queue : Box :: new ( SendQueueImpl :: new ( flow_label . clone ( ) , & flow_tag , * queue_size , self . send_queue_metrics . clone ( ) , ) ) , } ; self . report_connection_state ( & flow_state ) ; peer_state . flow_map . insert ( flow_endpoint . flow_tag . into ( ) , flow_state ) ; peer_state . connect_cancelers . push ( connect_canceler ) ; } client_state . peer_map . insert ( * peer_id , peer_state ) ; Ok ( ( ) ) }
Starts all connections to a peer and initializes the corresponding data
structures and tasks
Starts all connections to a peer and initializes the corresponding data structures and tasks
&self
client_type
TransportClientType
peer_id
&NodeId
peer_record
&NodeRecord
client_state
&mut ClientState
&self
client_type
TransportClientType
peer_id
&NodeId
peer_record
&NodeRecord
client_state
&mut ClientState