async fn subscribe_slow(
        url: &str,
        barrier: Arc<Barrier>,
        client_id: &str,
    ) -> Result<(), reqwest::Error> {
        barrier.wait().await;
        let response = reqwest::get(url).await.unwrap();
        barrier.wait().await;

        time::sleep(Duration::from_secs(5)).await;

        let mut stream = response.bytes_stream();
        let pause_between_events = Duration::from_secs(100) / MAX_EVENT_COUNT;
        while let Some(item) = stream.next().await {
            // The function is expected to exit here with an `UnexpectedEof` error.
            let bytes = item?;
            let chunk = str::from_utf8(bytes.as_ref()).unwrap();
            if chunk.lines().any(|line| line == ":") {
                debug!("{} received keepalive: exiting", client_id);
                break;
            }
            time::sleep(pause_between_events).await;
        }

        Ok(())
    }
// Similar to the `subscribe()` function, except this has a long pause at the start and short
// pauses after each read.
//
// The objective is to create backpressure by filling the client's receive buffer, then filling
// the server's send buffer, which in turn causes the server's internal broadcast channel to
// deem that client as lagging.
Similar to the `subscribe()` function, except this has a long pause at the start and short
pauses after each read.

The objective is to create backpressure by filling the client's receive buffer, then filling
the server's send buffer, which in turn causes the server's internal broadcast channel to
deem that client as lagging.
Similar to the ` subscribe () ` function except this has a long pause at the start and short pauses after each read . The objective is to create backpressure by filling the client ' s receive buffer then filling the server ' s send buffer which in turn causes the server ' s internal broadcast channel to deem that client as lagging .
async fn subscribe_slow(
        url: &str,
        barrier: Arc<Barrier>,
        client_id: &str,
    ) -> Result<(), reqwest::Error> {
        barrier.wait().await;
        let response = reqwest::get(url).await.unwrap();
        barrier.wait().await;
        time::sleep(Duration::from_secs(5)).await;
        let mut stream = response.bytes_stream();
        let pause_between_events = Duration::from_secs(100) / MAX_EVENT_COUNT;
        while let Some(item) = stream.next().await {
            // The function is expected to exit here with an `UnexpectedEof` error.
            let bytes = item?;
            let chunk = str::from_utf8(bytes.as_ref()).unwrap();
            if chunk.lines().any(|line| line == ":") {
                debug!("{} received keepalive: exiting", client_id);
                break;
            }
            time::sleep(pause_between_events).await;
        }
        Ok(())
    }
async fn subscribe_slow ( url : & str , barrier : Arc < Barrier > , client_id : & str , ) -> Result < ( ) , reqwest :: Error > { barrier . wait ( ) . await ; let response = reqwest :: get ( url ) . await . unwrap ( ) ; barrier . wait ( ) . await ; time :: sleep ( Duration :: from_secs ( 5 ) ) . await ; let mut stream = response . bytes_stream ( ) ; let pause_between_events = Duration :: from_secs ( 100 ) / MAX_EVENT_COUNT ; while let Some ( item ) = stream . next ( ) . await { // The function is expected to exit here with an `UnexpectedEof` error. let bytes = item ? ; let chunk = str :: from_utf8 ( bytes . as_ref ( ) ) . unwrap ( ) ; if chunk . lines ( ) . any ( | line | line == " " ) { debug ! ( " " client_id ) ; break ; } time :: sleep ( pause_between_events ) . await ; } Ok ( ( ) ) }
Similar to the `subscribe()` function, except this has a long pause at the start and short
pauses after each read.
Similar to the ` subscribe () ` function except this has a long pause at the start and short pauses after each read .
url
&str
barrier
Arc<Barrier>
client_id
&str
url
&str
barrier
Arc<Barrier>
client_id
&str