async fn spawn_inbound_listener_with_peer_limit<S>(
    peerset_initial_target_size: impl Into<Option<usize>>,
    listen_handshaker: S,
) -> (Config, mpsc::Receiver<DiscoveredPeer>)
where
    S: Service<peer::HandshakeRequest, Response = peer::Client, Error = BoxError>
        + Clone
        + Send
        + 'static,
    S::Future: Send + 'static,
{
    // Create a test config that listens on any unused port.
    let listen_addr = "127.0.0.1:0".parse().unwrap();
    let mut config = Config {
        listen_addr,
        ..Config::default()
    };

    if let Some(peerset_initial_target_size) = peerset_initial_target_size.into() {
        config.peerset_initial_target_size = peerset_initial_target_size;
    }

    // Open the listener port.
    let (tcp_listener, listen_addr) = open_listener(&config.clone()).await;

    // Make enough inbound connections to go over the limit, even if the limit is zero.
    // Make the channels large enough to hold all the connections.
    let over_limit_connections = config.peerset_inbound_connection_limit() * 2 + 1;
    let (peerset_tx, peerset_rx) = mpsc::channel::<DiscoveredPeer>(over_limit_connections);

    // Start listening for connections.
    let listen_fut = accept_inbound_connections(
        config.clone(),
        tcp_listener,
        listen_handshaker,
        peerset_tx.clone(),
    );
    let listen_task_handle = tokio::spawn(listen_fut);

    // Open inbound connections.
    let mut outbound_task_handles = Vec::new();
    for _ in 0..over_limit_connections {
        let outbound_fut = async move {
            let outbound_result = TcpStream::connect(listen_addr).await;
            // Let other tasks run before we block on reading.
            tokio::task::yield_now().await;

            if let Ok(outbound_stream) = outbound_result {
                // Wait until the listener closes the connection.
                // The handshaker is fake, so it never sends any data.
                let readable_result = outbound_stream.readable().await;
                info!(
                    ?readable_result,
                    "outbound connection became readable or errored: \
                     closing connection to test inbound listener"
                );
            } else {
                // If the connection is closed quickly, we might get errors here.
                debug!(
                    ?outbound_result,
                    "outbound connection error in inbound listener test"
                );
            }
        };

        let outbound_task_handle = tokio::spawn(outbound_fut);
        outbound_task_handles.push(outbound_task_handle);
    }

    // Let the listener run for a while.
    tokio::time::sleep(LISTENER_TEST_DURATION).await;

    // Stop the listener and outbound tasks, and let them finish.
    listen_task_handle.abort();
    for outbound_task_handle in outbound_task_handles {
        outbound_task_handle.abort();
    }
    tokio::task::yield_now().await;

    // Check for panics or errors in the listener.
    let listen_result = listen_task_handle.now_or_never();
    assert!(
        matches!(listen_result, None)
            || matches!(listen_result, Some(Err(ref e)) if e.is_cancelled()),
        "unexpected error or panic in inbound peer listener task: {:?}",
        listen_result,
    );

    (config, peerset_rx)
}
/// Run an inbound peer listener with `peerset_initial_target_size` and `handshaker`.
///
/// Binds the local listener to an unused localhost port.
/// Uses the default values for all other config fields.
///
/// Returns the generated [`Config`], and the peer set receiver.
Run an inbound peer listener with `peerset_initial_target_size` and `handshaker`.
Binds the local listener to an unused localhost port.
Uses the default values for all other config fields.

Returns the generated [`Config`], and the peer set receiver.
Run an inbound peer listener with ` peerset_initial_target_size ` and ` handshaker ` . Binds the local listener to an unused localhost port . Uses the default values for all other config fields . Returns the generated [ ` Config ` ] and the peer set receiver .
async fn spawn_inbound_listener_with_peer_limit<S>(
    peerset_initial_target_size: impl Into<Option<usize>>,
    listen_handshaker: S,
) -> (Config, mpsc::Receiver<DiscoveredPeer>)
where
    S: Service<peer::HandshakeRequest, Response = peer::Client, Error = BoxError>
        + Clone
        + Send
        + 'static,
    S::Future: Send + 'static,
{
    // Create a test config that listens on any unused port.
    let listen_addr = "127.0.0.1:0".parse().unwrap();
    let mut config = Config {
        listen_addr,
        ..Config::default()
    };
    if let Some(peerset_initial_target_size) = peerset_initial_target_size.into() {
        config.peerset_initial_target_size = peerset_initial_target_size;
    }
    // Open the listener port.
    let (tcp_listener, listen_addr) = open_listener(&config.clone()).await;
    // Make enough inbound connections to go over the limit, even if the limit is zero.
    // Make the channels large enough to hold all the connections.
    let over_limit_connections = config.peerset_inbound_connection_limit() * 2 + 1;
    let (peerset_tx, peerset_rx) = mpsc::channel::<DiscoveredPeer>(over_limit_connections);
    // Start listening for connections.
    let listen_fut = accept_inbound_connections(
        config.clone(),
        tcp_listener,
        listen_handshaker,
        peerset_tx.clone(),
    );
    let listen_task_handle = tokio::spawn(listen_fut);
    // Open inbound connections.
    let mut outbound_task_handles = Vec::new();
    for _ in 0..over_limit_connections {
        let outbound_fut = async move {
            let outbound_result = TcpStream::connect(listen_addr).await;
            // Let other tasks run before we block on reading.
            tokio::task::yield_now().await;
            if let Ok(outbound_stream) = outbound_result {
                // Wait until the listener closes the connection.
                // The handshaker is fake, so it never sends any data.
                let readable_result = outbound_stream.readable().await;
                info!(
                    ?readable_result,
                    "outbound connection became readable or errored: \
                     closing connection to test inbound listener"
                );
            } else {
                // If the connection is closed quickly, we might get errors here.
                debug!(
                    ?outbound_result,
                    "outbound connection error in inbound listener test"
                );
            }
        };
        let outbound_task_handle = tokio::spawn(outbound_fut);
        outbound_task_handles.push(outbound_task_handle);
    }
    // Let the listener run for a while.
    tokio::time::sleep(LISTENER_TEST_DURATION).await;
    // Stop the listener and outbound tasks, and let them finish.
    listen_task_handle.abort();
    for outbound_task_handle in outbound_task_handles {
        outbound_task_handle.abort();
    }
    tokio::task::yield_now().await;
    // Check for panics or errors in the listener.
    let listen_result = listen_task_handle.now_or_never();
    assert!(
        matches!(listen_result, None)
            || matches!(listen_result, Some(Err(ref e)) if e.is_cancelled()),
        "unexpected error or panic in inbound peer listener task: {:?}",
        listen_result,
    );
    (config, peerset_rx)
}
async fn spawn_inbound_listener_with_peer_limit < S > ( peerset_initial_target_size : impl Into < Option < usize > > , listen_handshaker : S , ) -> ( Config , mpsc :: Receiver < DiscoveredPeer > ) where S : Service < peer :: HandshakeRequest , Response = peer :: Client , Error = BoxError > + Clone + Send + ' static , S :: Future : Send + ' static , { // Create a test config that listens on any unused port. let listen_addr = " " . parse ( ) . unwrap ( ) ; let mut config = Config { listen_addr , .. Config :: default ( ) } ; if let Some ( peerset_initial_target_size ) = peerset_initial_target_size . into ( ) { config . peerset_initial_target_size = peerset_initial_target_size ; } // Open the listener port. let ( tcp_listener , listen_addr ) = open_listener ( & config . clone ( ) ) . await ; // Make enough inbound connections to go over the limit, even if the limit is zero. // Make the channels large enough to hold all the connections. let over_limit_connections = config . peerset_inbound_connection_limit ( ) * 2 + 1 ; let ( peerset_tx , peerset_rx ) = mpsc :: channel :: < DiscoveredPeer > ( over_limit_connections ) ; // Start listening for connections. let listen_fut = accept_inbound_connections ( config . clone ( ) , tcp_listener , listen_handshaker , peerset_tx . clone ( ) , ) ; let listen_task_handle = tokio :: spawn ( listen_fut ) ; // Open inbound connections. let mut outbound_task_handles = Vec :: new ( ) ; for _ in 0 .. over_limit_connections { let outbound_fut = async move { let outbound_result = TcpStream :: connect ( listen_addr ) . await ; // Let other tasks run before we block on reading. tokio :: task :: yield_now ( ) . await ; if let Ok ( outbound_stream ) = outbound_result { // Wait until the listener closes the connection. // The handshaker is fake, so it never sends any data. let readable_result = outbound_stream . readable ( ) . await ; info ! ( readable_result " \
 " ) ; } else { // If the connection is closed quickly, we might get errors here. debug ! ( outbound_result " " ) ; } } ; let outbound_task_handle = tokio :: spawn ( outbound_fut ) ; outbound_task_handles . push ( outbound_task_handle ) ; } // Let the listener run for a while. tokio :: time :: sleep ( LISTENER_TEST_DURATION ) . await ; // Stop the listener and outbound tasks, and let them finish. listen_task_handle . abort ( ) ; for outbound_task_handle in outbound_task_handles { outbound_task_handle . abort ( ) ; } tokio :: task :: yield_now ( ) . await ; // Check for panics or errors in the listener. let listen_result = listen_task_handle . now_or_never ( ) ; assert ! ( matches ( listen_result None ) matches ( listen_result Some ( Err ( ref e ) ) if e is_cancelled ( ) ) " " listen_result ) ; ( config , peerset_rx ) }
Run an inbound peer listener with `peerset_initial_target_size` and `handshaker`.
Run an inbound peer listener with ` peerset_initial_target_size ` and ` handshaker ` .
peerset_initial_target_size
impl Into<Option<usize>>
listen_handshaker
S