fn process_ingress(
    ingress_sender_mu: &Mutex<Option<IngressIngestionService>>,
    ingress_hist_reader: &dyn IngressHistoryReader,
    msg: SignedIngress,
    time_limit: Duration,
) -> Result<WasmResult, UserError> {
    let msg_id = msg.id();

    tokio::task::block_in_place(move || {
        let rt = tokio::runtime::Runtime::new().unwrap();
        rt.block_on(async move {
            let mut ingress_sender_guard = ingress_sender_mu.lock().unwrap();
            ingress_sender_guard
                .as_mut()
                .unwrap()
                .ready()
                .await
                .expect("The service must always be able to process requests.")
                .call(msg)
                .await
                .unwrap()
                .unwrap();
        });
    });

    let start = Instant::now();
    loop {
        std::thread::sleep(Duration::from_millis(5));
        let ingress_result = (ingress_hist_reader.get_latest_status())(&msg_id);
        match ingress_result {
            IngressStatus::Completed { result, .. } => {
                // Don't forget! Signal the runtime to stop.
                return Ok(result);
            }
            IngressStatus::Failed { error, .. } => {
                // Don't forget! Signal the runtime to stop.
                return Err(error);
            }
            IngressStatus::Received { .. }
            | IngressStatus::Processing { .. }
            | IngressStatus::Unknown => (),
        }
        if Instant::now().duration_since(start) > time_limit {
            panic!(
                "Ingress message did not finish executing in {} seconds, panicking",
                time_limit.as_secs()
            );
        }

        // Either this requires only few iterations, so waiting a bit is not an
        // issue. Or, it requires a lot of iterations and then we don't want to burn
        // CPU cycles on CI where such loops run in parallel and potentially steal
        // each others CPU time for no good reason.
        sleep(Duration::from_millis(100));
    }
}
/// Executes an ingress message and blocks till execution finishes.
///
/// Note: To ensure that this function does not block forever (in case of bugs),
/// this function will panic if the process is not finished in some amount of
/// time.
Executes an ingress message and blocks till execution finishes.
Note: To ensure that this function does not block forever (in case of bugs),
this function will panic if the process is not finished in some amount of
time.
Executes an ingress message and blocks till execution finishes . Note : To ensure that this function does not block forever ( in case of bugs ) this function will panic if the process is not finished in some amount of time .
fn process_ingress(
    ingress_sender_mu: &Mutex<Option<IngressIngestionService>>,
    ingress_hist_reader: &dyn IngressHistoryReader,
    msg: SignedIngress,
    time_limit: Duration,
) -> Result<WasmResult, UserError> {
    let msg_id = msg.id();
    tokio::task::block_in_place(move || {
        let rt = tokio::runtime::Runtime::new().unwrap();
        rt.block_on(async move {
            let mut ingress_sender_guard = ingress_sender_mu.lock().unwrap();
            ingress_sender_guard
                .as_mut()
                .unwrap()
                .ready()
                .await
                .expect("The service must always be able to process requests.")
                .call(msg)
                .await
                .unwrap()
                .unwrap();
        });
    });
    let start = Instant::now();
    loop {
        std::thread::sleep(Duration::from_millis(5));
        let ingress_result = (ingress_hist_reader.get_latest_status())(&msg_id);
        match ingress_result {
            IngressStatus::Completed { result, .. } => {
                // Don't forget! Signal the runtime to stop.
                return Ok(result);
            }
            IngressStatus::Failed { error, .. } => {
                // Don't forget! Signal the runtime to stop.
                return Err(error);
            }
            IngressStatus::Received { .. }
            | IngressStatus::Processing { .. }
            | IngressStatus::Unknown => (),
        }
        if Instant::now().duration_since(start) > time_limit {
            panic!(
                "Ingress message did not finish executing in {} seconds, panicking",
                time_limit.as_secs()
            );
        }
        // Either this requires only few iterations, so waiting a bit is not an
        // issue. Or, it requires a lot of iterations and then we don't want to burn
        // CPU cycles on CI where such loops run in parallel and potentially steal
        // each others CPU time for no good reason.
        sleep(Duration::from_millis(100));
    }
}
fn process_ingress ( ingress_sender_mu : & Mutex < Option < IngressIngestionService > > , ingress_hist_reader : & dyn IngressHistoryReader , msg : SignedIngress , time_limit : Duration , ) -> Result < WasmResult , UserError > { let msg_id = msg . id ( ) ; tokio :: task :: block_in_place ( move | | { let rt = tokio :: runtime :: Runtime :: new ( ) . unwrap ( ) ; rt . block_on ( async move { let mut ingress_sender_guard = ingress_sender_mu . lock ( ) . unwrap ( ) ; ingress_sender_guard . as_mut ( ) . unwrap ( ) . ready ( ) . await . expect ( " " ) . call ( msg ) . await . unwrap ( ) . unwrap ( ) ; } ) ; } ) ; let start = Instant :: now ( ) ; loop { std :: thread :: sleep ( Duration :: from_millis ( 5 ) ) ; let ingress_result = ( ingress_hist_reader . get_latest_status ( ) ) ( & msg_id ) ; match ingress_result { IngressStatus :: Completed { result , .. } => { // Don't forget! Signal the runtime to stop. return Ok ( result ) ; } IngressStatus :: Failed { error , .. } => { // Don't forget! Signal the runtime to stop. return Err ( error ) ; } IngressStatus :: Received { .. } | IngressStatus :: Processing { .. } | IngressStatus :: Unknown => ( ) , } if Instant :: now ( ) . duration_since ( start ) > time_limit { panic ! ( " " time_limit as_secs ( ) ) ; } // Either this requires only few iterations, so waiting a bit is not an // issue. Or, it requires a lot of iterations and then we don't want to burn // CPU cycles on CI where such loops run in parallel and potentially steal // each others CPU time for no good reason. sleep ( Duration :: from_millis ( 100 ) ) ; } }
Executes an ingress message and blocks till execution finishes.
Executes an ingress message and blocks till execution finishes .
ingress_sender_mu
&Mutex<Option<IngressIngestionService>>
ingress_hist_reader
&dyn IngressHistoryReader
msg
SignedIngress
time_limit
Duration
ingress_sender_mu
&Mutex<Option<IngressIngestionService>>
ingress_hist_reader
&dyn IngressHistoryReader
msg
SignedIngress
time_limit
Duration