pub(crate) async fn direct_stable(
    tid: u64,
    delivery: Vec<u8>,
    to: PeerId,
    addr: SocketAddr,
    global: Arc<Global>,
    is_recv_data: bool,
) -> Result<()> {
    debug!("Session want to connect directly.");
    let (endpoint_sender, endpoint_receiver) = new_endpoint_channel(); // transpot's use.
    let (stream_sender, mut stream_receiver) = new_endpoint_channel(); // session's use.
    let (mut session_key, remote_pk) = global.generate_remote();

    // 1. send stable connect.
    global
        .trans_send(TransportSendMessage::StableConnect(
            stream_sender.clone(),
            endpoint_receiver,
            addr,
            remote_pk,
        ))
        .await?;

    // 2. waiting remote send remote info.
    if let Some(EndpointMessage::Handshake(RemotePublic(remote_key, remote_peer, dh_key))) =
        stream_receiver.recv().await
    {
        // 3.1.1 if ok connected. keep it and update to stable.
        let remote_id = remote_key.peer_id();
        if remote_id != to {
            warn!("CHAMOMILE: STABLE CONNECT FAILURE UNKNOWN PEER.");
            return Err(new_io_error("session stable unknown peer."));
        }

        if &remote_id == global.peer_id() {
            warn!("CHAMOMILE: STABLE CONNECT NERVER TO SELF.");
            let _ = endpoint_sender.send(EndpointMessage::Close).await;
            if tid != 0 {
                global
                    .out_send(ReceiveMessage::Delivery(
                        DeliveryType::StableConnect,
                        tid,
                        false,
                        delivery,
                    ))
                    .await?;
            }
            return Err(new_io_error("session stable self failure."));
        }

        // 3.1.2 check & update session key.
        if !session_key.complete(&remote_key.pk, dh_key) {
            global.buffer.write().await.remove_connect(&to);
            return Err(new_io_error("session stable key failure."));
        }

        let remote_peer = nat(addr, remote_peer);
        let (session_sender, session_receiver) = new_session_channel(); // server's use.

        // 3.1.3 save to tmp buffer.
        let buffers = global
            .add_tmp(
                remote_id,
                KadValue(session_sender.clone(), stream_sender, remote_peer),
                true,
            )
            .await;

        let mut session = Session::new(
            remote_peer,
            session_sender,
            stream_receiver,
            ConnectType::Direct(endpoint_sender),
            session_key,
            global,
            is_recv_data,
        );

        // 3.1.4 send all connect info to remote.
        for buffer in buffers {
            session
                .send_core_data(CoreData::StableConnect(buffer.0, buffer.1))
                .await?;
        }

        // 3.1.5 upgrade to stable.
        if !session.is_stable {
            session.upgrade().await?;
        }

        // 3.1.6 session listen.
        session.listen(session_receiver).await
    } else {
        drop(stream_sender);
        drop(stream_receiver);
        drop(endpoint_sender);

        // 3.2.1 try start relay stable.
        let ss = if let Some((s, _, _)) = global.peer_list.read().await.get(&to) {
            Some(s.clone())
        } else {
            None
        };

        if let Some(ss) = ss {
            relay_stable(tid, delivery, to, ss, global, is_recv_data).await
        } else {
            if tid != 0 {
                global
                    .out_send(ReceiveMessage::Delivery(
                        DeliveryType::StableConnect,
                        tid,
                        false,
                        delivery,
                    ))
                    .await?;
            }
            global.buffer.write().await.remove_connect(&to);
            Err(new_io_error("no closest peer."))
        }
    }
}
/// direct start stable connection, if had IP.
direct start stable connection, if had IP.
direct start stable connection if had IP .
pub(crate) async fn direct_stable(
    tid: u64,
    delivery: Vec<u8>,
    to: PeerId,
    addr: SocketAddr,
    global: Arc<Global>,
    is_recv_data: bool,
) -> Result<()> {
    debug!("Session want to connect directly.");
    let (endpoint_sender, endpoint_receiver) = new_endpoint_channel(); // transpot's use.
    let (stream_sender, mut stream_receiver) = new_endpoint_channel(); // session's use.
    let (mut session_key, remote_pk) = global.generate_remote();
    // 1. send stable connect.
    global
        .trans_send(TransportSendMessage::StableConnect(
            stream_sender.clone(),
            endpoint_receiver,
            addr,
            remote_pk,
        ))
        .await?;
    // 2. waiting remote send remote info.
    if let Some(EndpointMessage::Handshake(RemotePublic(remote_key, remote_peer, dh_key))) =
        stream_receiver.recv().await
    {
        // 3.1.1 if ok connected. keep it and update to stable.
        let remote_id = remote_key.peer_id();
        if remote_id != to {
            warn!("CHAMOMILE: STABLE CONNECT FAILURE UNKNOWN PEER.");
            return Err(new_io_error("session stable unknown peer."));
        }
        if &remote_id == global.peer_id() {
            warn!("CHAMOMILE: STABLE CONNECT NERVER TO SELF.");
            let _ = endpoint_sender.send(EndpointMessage::Close).await;
            if tid != 0 {
                global
                    .out_send(ReceiveMessage::Delivery(
                        DeliveryType::StableConnect,
                        tid,
                        false,
                        delivery,
                    ))
                    .await?;
            }
            return Err(new_io_error("session stable self failure."));
        }
        // 3.1.2 check & update session key.
        if !session_key.complete(&remote_key.pk, dh_key) {
            global.buffer.write().await.remove_connect(&to);
            return Err(new_io_error("session stable key failure."));
        }
        let remote_peer = nat(addr, remote_peer);
        let (session_sender, session_receiver) = new_session_channel(); // server's use.
        // 3.1.3 save to tmp buffer.
        let buffers = global
            .add_tmp(
                remote_id,
                KadValue(session_sender.clone(), stream_sender, remote_peer),
                true,
            )
            .await;
        let mut session = Session::new(
            remote_peer,
            session_sender,
            stream_receiver,
            ConnectType::Direct(endpoint_sender),
            session_key,
            global,
            is_recv_data,
        );
        // 3.1.4 send all connect info to remote.
        for buffer in buffers {
            session
                .send_core_data(CoreData::StableConnect(buffer.0, buffer.1))
                .await?;
        }
        // 3.1.5 upgrade to stable.
        if !session.is_stable {
            session.upgrade().await?;
        }
        // 3.1.6 session listen.
        session.listen(session_receiver).await
    } else {
        drop(stream_sender);
        drop(stream_receiver);
        drop(endpoint_sender);
        // 3.2.1 try start relay stable.
        let ss = if let Some((s, _, _)) = global.peer_list.read().await.get(&to) {
            Some(s.clone())
        } else {
            None
        };
        if let Some(ss) = ss {
            relay_stable(tid, delivery, to, ss, global, is_recv_data).await
        } else {
            if tid != 0 {
                global
                    .out_send(ReceiveMessage::Delivery(
                        DeliveryType::StableConnect,
                        tid,
                        false,
                        delivery,
                    ))
                    .await?;
            }
            global.buffer.write().await.remove_connect(&to);
            Err(new_io_error("no closest peer."))
        }
    }
}
pub ( crate ) async fn direct_stable ( tid : u64 , delivery : Vec < u8 > , to : PeerId , addr : SocketAddr , global : Arc < Global > , is_recv_data : bool , ) -> Result < ( ) > { debug ! ( " " ) ; let ( endpoint_sender , endpoint_receiver ) = new_endpoint_channel ( ) ; // transpot's use. let ( stream_sender , mut stream_receiver ) = new_endpoint_channel ( ) ; // session's use. let ( mut session_key , remote_pk ) = global . generate_remote ( ) ; // 1. send stable connect. global . trans_send ( TransportSendMessage :: StableConnect ( stream_sender . clone ( ) , endpoint_receiver , addr , remote_pk , ) ) . await ? ; // 2. waiting remote send remote info. if let Some ( EndpointMessage :: Handshake ( RemotePublic ( remote_key , remote_peer , dh_key ) ) ) = stream_receiver . recv ( ) . await { // 3.1.1 if ok connected. keep it and update to stable. let remote_id = remote_key . peer_id ( ) ; if remote_id != to { warn ! ( " " ) ; return Err ( new_io_error ( " " ) ) ; } if & remote_id == global . peer_id ( ) { warn ! ( " " ) ; let _ = endpoint_sender . send ( EndpointMessage :: Close ) . await ; if tid != 0 { global . out_send ( ReceiveMessage :: Delivery ( DeliveryType :: StableConnect , tid , false , delivery , ) ) . await ? ; } return Err ( new_io_error ( " " ) ) ; } // 3.1.2 check & update session key. if ! session_key . complete ( & remote_key . pk , dh_key ) { global . buffer . write ( ) . await . remove_connect ( & to ) ; return Err ( new_io_error ( " " ) ) ; } let remote_peer = nat ( addr , remote_peer ) ; let ( session_sender , session_receiver ) = new_session_channel ( ) ; // server's use. // 3.1.3 save to tmp buffer. let buffers = global . add_tmp ( remote_id , KadValue ( session_sender . clone ( ) , stream_sender , remote_peer ) , true , ) . await ; let mut session = Session :: new ( remote_peer , session_sender , stream_receiver , ConnectType :: Direct ( endpoint_sender ) , session_key , global , is_recv_data , ) ; // 3.1.4 send all connect info to remote. for buffer in buffers { session . send_core_data ( CoreData :: StableConnect ( buffer . 0 , buffer . 1 ) ) . await ? ; } // 3.1.5 upgrade to stable. if ! session . is_stable { session . upgrade ( ) . await ? ; } // 3.1.6 session listen. session . listen ( session_receiver ) . await } else { drop ( stream_sender ) ; drop ( stream_receiver ) ; drop ( endpoint_sender ) ; // 3.2.1 try start relay stable. let ss = if let Some ( ( s , _ , _ ) ) = global . peer_list . read ( ) . await . get ( & to ) { Some ( s . clone ( ) ) } else { None } ; if let Some ( ss ) = ss { relay_stable ( tid , delivery , to , ss , global , is_recv_data ) . await } else { if tid != 0 { global . out_send ( ReceiveMessage :: Delivery ( DeliveryType :: StableConnect , tid , false , delivery , ) ) . await ? ; } global . buffer . write ( ) . await . remove_connect ( & to ) ; Err ( new_io_error ( " " ) ) } } }
direct start stable connection, if had IP.
direct start stable connection if had IP .
tid
u64
delivery
Vec<u8>
to
PeerId
addr
SocketAddr
global
Arc<Global>
is_recv_data
bool
tid
u64
delivery
Vec<u8>
to
PeerId
addr
SocketAddr
global
Arc<Global>
is_recv_data
bool