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

        let mut peer_state = PeerState {
            flow_map: HashMap::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 role == ConnectionRole::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::new(
                    flow_id,
                    flow_config.flow_tag.to_string(),
                    flow_label.clone(),
                    ConnectionState::Listening,
                    Box::new(SendQueueImpl::new(
                        flow_label,
                        &flow_tag,
                        QueueSize::from(flow_config.queue_size),
                        self.send_queue_metrics.clone(),
                    )),
                    self.control_plane_metrics.clone(),
                );
                peer_state.flow_map.insert(flow_tag, flow_state);
            }
        }
        if role == ConnectionRole::Server {
            client_state.peer_map.insert(*peer_id, peer_state);
            return Ok(());
        }

        for flow_endpoint in &peer_record.p2p_flow_endpoints {
            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;
            let connecting_task = self.spawn_connect_task(
                client_type,
                flow_endpoint.flow_tag.into(),
                *peer_id,
                peer_ip,
                ServerPort::from(server_port),
            );
            let connecting_state = Connecting {
                peer_addr: SocketAddr::new(peer_ip, server_port),
                connecting_task,
            };
            let flow_id = FlowId {
                client_type,
                peer_id: *peer_id,
                flow_tag,
            };
            let flow_state = FlowState::new(
                flow_id,
                flow_endpoint.flow_tag.to_string(),
                flow_label.clone(),
                ConnectionState::Connecting(connecting_state),
                Box::new(SendQueueImpl::new(
                    flow_label.clone(),
                    &flow_tag,
                    *queue_size,
                    self.send_queue_metrics.clone(),
                )),
                self.control_plane_metrics.clone(),
            );
            peer_state
                .flow_map
                .insert(flow_endpoint.flow_tag.into(), flow_state);
        }

        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,
        role: ConnectionRole,
    ) -> Result<(), TransportErrorCode> {
        if client_state.peer_map.get(peer_id).is_some() {
            return Err(TransportErrorCode::PeerAlreadyRegistered);
        }
        let mut peer_state = PeerState {
            flow_map: HashMap::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 role == ConnectionRole::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::new(
                    flow_id,
                    flow_config.flow_tag.to_string(),
                    flow_label.clone(),
                    ConnectionState::Listening,
                    Box::new(SendQueueImpl::new(
                        flow_label,
                        &flow_tag,
                        QueueSize::from(flow_config.queue_size),
                        self.send_queue_metrics.clone(),
                    )),
                    self.control_plane_metrics.clone(),
                );
                peer_state.flow_map.insert(flow_tag, flow_state);
            }
        }
        if role == ConnectionRole::Server {
            client_state.peer_map.insert(*peer_id, peer_state);
            return Ok(());
        }
        for flow_endpoint in &peer_record.p2p_flow_endpoints {
            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;
            let connecting_task = self.spawn_connect_task(
                client_type,
                flow_endpoint.flow_tag.into(),
                *peer_id,
                peer_ip,
                ServerPort::from(server_port),
            );
            let connecting_state = Connecting {
                peer_addr: SocketAddr::new(peer_ip, server_port),
                connecting_task,
            };
            let flow_id = FlowId {
                client_type,
                peer_id: *peer_id,
                flow_tag,
            };
            let flow_state = FlowState::new(
                flow_id,
                flow_endpoint.flow_tag.to_string(),
                flow_label.clone(),
                ConnectionState::Connecting(connecting_state),
                Box::new(SendQueueImpl::new(
                    flow_label.clone(),
                    &flow_tag,
                    *queue_size,
                    self.send_queue_metrics.clone(),
                )),
                self.control_plane_metrics.clone(),
            );
            peer_state
                .flow_map
                .insert(flow_endpoint.flow_tag.into(), flow_state);
        }
        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 , role : ConnectionRole , ) -> Result < ( ) , TransportErrorCode > { if client_state . peer_map . get ( peer_id ) . is_some ( ) { return Err ( TransportErrorCode :: PeerAlreadyRegistered ) ; } let mut peer_state = PeerState { flow_map : HashMap :: 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 role == ConnectionRole :: 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 :: new ( flow_id , flow_config . flow_tag . to_string ( ) , flow_label . clone ( ) , ConnectionState :: Listening , Box :: new ( SendQueueImpl :: new ( flow_label , & flow_tag , QueueSize :: from ( flow_config . queue_size ) , self . send_queue_metrics . clone ( ) , ) ) , self . control_plane_metrics . clone ( ) , ) ; peer_state . flow_map . insert ( flow_tag , flow_state ) ; } } if role == ConnectionRole :: Server { client_state . peer_map . insert ( * peer_id , peer_state ) ; return Ok ( ( ) ) ; } for flow_endpoint in & peer_record . p2p_flow_endpoints { 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 ; let connecting_task = self . spawn_connect_task ( client_type , flow_endpoint . flow_tag . into ( ) , * peer_id , peer_ip , ServerPort :: from ( server_port ) , ) ; let connecting_state = Connecting { peer_addr : SocketAddr :: new ( peer_ip , server_port ) , connecting_task , } ; let flow_id = FlowId { client_type , peer_id : * peer_id , flow_tag , } ; let flow_state = FlowState :: new ( flow_id , flow_endpoint . flow_tag . to_string ( ) , flow_label . clone ( ) , ConnectionState :: Connecting ( connecting_state ) , Box :: new ( SendQueueImpl :: new ( flow_label . clone ( ) , & flow_tag , * queue_size , self . send_queue_metrics . clone ( ) , ) ) , self . control_plane_metrics . clone ( ) , ) ; peer_state . flow_map . insert ( flow_endpoint . flow_tag . into ( ) , flow_state ) ; } 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
role
ConnectionRole
&self
client_type
TransportClientType
peer_id
&NodeId
peer_record
&NodeRecord
client_state
&mut ClientState
role
ConnectionRole