pub(crate) async fn ipc(
    mut receiver: Receiver<Message>,
    init_data_sender: Sender<(HiveProbeData, HiveTargetData)>,
    notify_results_ready: Arc<Notify>,
) {
    let socket_path = Path::new(SOCKET_PATH);

    let mpsc_handler = tokio::spawn(async move {
        while let Some(msg) = receiver.recv().await {
            match msg {
                Message::InitError(_) => todo!(),
                Message::TestResult(result) => {
                    let mut results = TEST_RESULTS.lock().await;
                    results.push(result);
                }
            }
        }
    });

    let ipc_handler = tokio::spawn(async move {
        let connector = tower::service_fn(move |_: Uri| {
            let path = socket_path;
            Box::pin(async move {
                let stream = UnixStream::connect(path).await?;
                Ok::<_, io::Error>(IpcConnection { stream })
            })
        });

        let client: Client<_, Body> = hyper::Client::builder().build(connector);

        let client_copy = client.clone();

        let initialization = tokio::spawn(async move {
            let probes = retry::try_request(client_copy.clone(), requests::get_probes())
                .await
                .unwrap();

            let targets = retry::try_request(client_copy.clone(), requests::get_targets())
                .await
                .unwrap();

            let probe_data;
            if let IpcMessage::ProbeInitData(data) = probes {
                probe_data = data;
            } else {
                panic!("Received wrong IpcMessage enum variant from the monitor!")
            }

            let target_data;
            if let IpcMessage::TargetInitData(data) = targets {
                target_data = data;
            } else {
                panic!("Received wrong IpcMessage enum variant from the monitor!")
            }

            // Notify main thread with init data, so it can start with testing
            init_data_sender.send((probe_data, target_data)).expect("Failed to send init data to main thread. Is the receiver still in scope and the thread still running?");
        });

        let client_copy = client.clone();

        let result_waiter = tokio::spawn(async move {
            notify_results_ready.notified().await;
            let mut results = TEST_RESULTS.lock().await;
            retry::try_request(
                client_copy,
                requests::post_test_results(mem::take(&mut *results)),
            )
            .await
            .unwrap();
        });

        initialization
            .await
            .expect("Failed to get initialization data from monitor");

        result_waiter
            .await
            .expect("Failed to send test results to monitor");
    });

    ipc_handler.await.unwrap();
    mpsc_handler.await.unwrap();
}
/// This function is the async entrypoint of tokio. All ipc from and to the monitor application are done here
This function is the async entrypoint of tokio. All ipc from and to the monitor application are done here
This function is the async entrypoint of tokio . All ipc from and to the monitor application are done here
pub(crate) async fn ipc(
    mut receiver: Receiver<Message>,
    init_data_sender: Sender<(HiveProbeData, HiveTargetData)>,
    notify_results_ready: Arc<Notify>,
) {
    let socket_path = Path::new(SOCKET_PATH);
    let mpsc_handler = tokio::spawn(async move {
        while let Some(msg) = receiver.recv().await {
            match msg {
                Message::InitError(_) => todo!(),
                Message::TestResult(result) => {
                    let mut results = TEST_RESULTS.lock().await;
                    results.push(result);
                }
            }
        }
    });
    let ipc_handler = tokio::spawn(async move {
        let connector = tower::service_fn(move |_: Uri| {
            let path = socket_path;
            Box::pin(async move {
                let stream = UnixStream::connect(path).await?;
                Ok::<_, io::Error>(IpcConnection { stream })
            })
        });
        let client: Client<_, Body> = hyper::Client::builder().build(connector);
        let client_copy = client.clone();
        let initialization = tokio::spawn(async move {
            let probes = retry::try_request(client_copy.clone(), requests::get_probes())
                .await
                .unwrap();
            let targets = retry::try_request(client_copy.clone(), requests::get_targets())
                .await
                .unwrap();
            let probe_data;
            if let IpcMessage::ProbeInitData(data) = probes {
                probe_data = data;
            } else {
                panic!("Received wrong IpcMessage enum variant from the monitor!")
            }
            let target_data;
            if let IpcMessage::TargetInitData(data) = targets {
                target_data = data;
            } else {
                panic!("Received wrong IpcMessage enum variant from the monitor!")
            }
            // Notify main thread with init data, so it can start with testing
            init_data_sender.send((probe_data, target_data)).expect("Failed to send init data to main thread. Is the receiver still in scope and the thread still running?");
        });
        let client_copy = client.clone();
        let result_waiter = tokio::spawn(async move {
            notify_results_ready.notified().await;
            let mut results = TEST_RESULTS.lock().await;
            retry::try_request(
                client_copy,
                requests::post_test_results(mem::take(&mut *results)),
            )
            .await
            .unwrap();
        });
        initialization
            .await
            .expect("Failed to get initialization data from monitor");
        result_waiter
            .await
            .expect("Failed to send test results to monitor");
    });
    ipc_handler.await.unwrap();
    mpsc_handler.await.unwrap();
}
pub ( crate ) async fn ipc ( mut receiver : Receiver < Message > , init_data_sender : Sender < ( HiveProbeData , HiveTargetData ) > , notify_results_ready : Arc < Notify > , ) { let socket_path = Path :: new ( SOCKET_PATH ) ; let mpsc_handler = tokio :: spawn ( async move { while let Some ( msg ) = receiver . recv ( ) . await { match msg { Message :: InitError ( _ ) => todo ! ( ) , Message :: TestResult ( result ) => { let mut results = TEST_RESULTS . lock ( ) . await ; results . push ( result ) ; } } } } ) ; let ipc_handler = tokio :: spawn ( async move { let connector = tower :: service_fn ( move | _ : Uri | { let path = socket_path ; Box :: pin ( async move { let stream = UnixStream :: connect ( path ) . await ? ; Ok :: < _ , io :: Error > ( IpcConnection { stream } ) } ) } ) ; let client : Client < _ , Body > = hyper :: Client :: builder ( ) . build ( connector ) ; let client_copy = client . clone ( ) ; let initialization = tokio :: spawn ( async move { let probes = retry :: try_request ( client_copy . clone ( ) , requests :: get_probes ( ) ) . await . unwrap ( ) ; let targets = retry :: try_request ( client_copy . clone ( ) , requests :: get_targets ( ) ) . await . unwrap ( ) ; let probe_data ; if let IpcMessage :: ProbeInitData ( data ) = probes { probe_data = data ; } else { panic ! ( " " ) } let target_data ; if let IpcMessage :: TargetInitData ( data ) = targets { target_data = data ; } else { panic ! ( " " ) } // Notify main thread with init data, so it can start with testing init_data_sender . send ( ( probe_data , target_data ) ) . expect ( " " ) ; } ) ; let client_copy = client . clone ( ) ; let result_waiter = tokio :: spawn ( async move { notify_results_ready . notified ( ) . await ; let mut results = TEST_RESULTS . lock ( ) . await ; retry :: try_request ( client_copy , requests :: post_test_results ( mem :: take ( & mut * results ) ) , ) . await . unwrap ( ) ; } ) ; initialization . await . expect ( " " ) ; result_waiter . await . expect ( " " ) ; } ) ; ipc_handler . await . unwrap ( ) ; mpsc_handler . await . unwrap ( ) ; }
This function is the async entrypoint of tokio.
This function is the async entrypoint of tokio .
receiver
Receiver<Message>
init_data_sender
Sender<(HiveProbeData, HiveTargetData)>
notify_results_ready
Arc<Notify>