fn collect_logs(
    mut body: hyper::Body,
) -> (JoinHandle<Vec<serde_json::Value>>, oneshot::Sender<()>) {
    let (done_tx, done_rx) = oneshot::channel();
    let result = tokio::spawn(async move {
        let mut result = Vec::new();
        let logs = &mut result;
        let fut = async move {
            while let Some(res) = body.data().await {
                let chunk = match res {
                    Ok(chunk) => chunk,
                    Err(e) => {
                        println!("body failed: {}", e);
                        break;
                    }
                };
                let deserialized = serde_json::from_slice(&chunk[..]);
                tracing::info!(?deserialized);
                match deserialized {
                    Ok(json) => logs.push(json),
                    Err(error) => panic!(
                        "parsing logs as JSON failed\n  error: {error}\n  chunk: {:?}",
                        String::from_utf8_lossy(&chunk[..])
                    ),
                }
            }
        };
        future::select(Box::pin(fut), done_rx.map(|_| ())).await;

        result
    });

    (result, done_tx)
}
/// Spawns a task to collect all the logs in a streaming body and parse them as
/// JSON.
Spawns a task to collect all the logs in a streaming body and parse them as
JSON.
Spawns a task to collect all the logs in a streaming body and parse them as JSON .
fn collect_logs(
    mut body: hyper::Body,
) -> (JoinHandle<Vec<serde_json::Value>>, oneshot::Sender<()>) {
    let (done_tx, done_rx) = oneshot::channel();
    let result = tokio::spawn(async move {
        let mut result = Vec::new();
        let logs = &mut result;
        let fut = async move {
            while let Some(res) = body.data().await {
                let chunk = match res {
                    Ok(chunk) => chunk,
                    Err(e) => {
                        println!("body failed: {}", e);
                        break;
                    }
                };
                let deserialized = serde_json::from_slice(&chunk[..]);
                tracing::info!(?deserialized);
                match deserialized {
                    Ok(json) => logs.push(json),
                    Err(error) => panic!(
                        "parsing logs as JSON failed\n  error: {error}\n  chunk: {:?}",
                        String::from_utf8_lossy(&chunk[..])
                    ),
                }
            }
        };
        future::select(Box::pin(fut), done_rx.map(|_| ())).await;
        result
    });
    (result, done_tx)
}
fn collect_logs ( mut body : hyper :: Body , ) -> ( JoinHandle < Vec < serde_json :: Value > > , oneshot :: Sender < ( ) > ) { let ( done_tx , done_rx ) = oneshot :: channel ( ) ; let result = tokio :: spawn ( async move { let mut result = Vec :: new ( ) ; let logs = & mut result ; let fut = async move { while let Some ( res ) = body . data ( ) . await { let chunk = match res { Ok ( chunk ) => chunk , Err ( e ) => { println ! ( " " e ) ; break ; } } ; let deserialized = serde_json :: from_slice ( & chunk [ .. ] ) ; tracing :: info ! ( deserialized ) ; match deserialized { Ok ( json ) => logs . push ( json ) , Err ( error ) => panic ! ( " \n \n " String from_utf8_lossy ( chunk [ ] ) ) , } } } ; future :: select ( Box :: pin ( fut ) , done_rx . map ( | _ | ( ) ) ) . await ; result } ) ; ( result , done_tx ) }
Spawns a task to collect all the logs in a streaming body and parse them as
JSON.
Spawns a task to collect all the logs in a streaming body and parse them as JSON .
body
hyper::Body