fn activity_task_handler(
        &mut self,
        worker: Arc<dyn CoreWorker>,
        activity: ActivityTask,
    ) -> Result<(), anyhow::Error> {
        match activity.variant {
            Some(activity_task::Variant::Start(start)) => {
                let act_fn = self
                    .activity_fns
                    .get(&start.activity_type)
                    .ok_or_else(|| {
                        anyhow!(
                            "No function registered for activity type {}",
                            start.activity_type
                        )
                    })?
                    .clone();
                let ct = CancellationToken::new();
                self.task_tokens_to_cancels
                    .insert(activity.task_token.clone().into(), ct.clone());

                tokio::spawn(ACT_CANCEL_TOK.scope(ct, async move {
                    let mut inputs = start.input;
                    let arg = inputs.pop().unwrap_or_default();
                    let output = (&act_fn.act_func)(arg).await;
                    let result = match output {
                        Ok(res) => ActivityExecutionResult::ok(res),
                        Err(err) => match err.downcast::<ActivityCancelledError>() {
                            Ok(ce) => ActivityExecutionResult::cancel_from_details(ce.details),
                            Err(other_err) => ActivityExecutionResult::fail(other_err.into()),
                        },
                    };
                    worker
                        .complete_activity_task(ActivityTaskCompletion {
                            task_token: activity.task_token,
                            result: Some(result),
                        })
                        .await?;
                    Result::<_, anyhow::Error>::Ok(())
                }));
            }
            Some(activity_task::Variant::Cancel(_)) => {
                if let Some(ct) = self.task_tokens_to_cancels.get(&activity.task_token.into()) {
                    ct.cancel();
                }
            }
            None => bail!("Undefined activity task variant"),
        }
        Ok(())
    }
/// Spawns off a task to handle the provided activity task
Spawns off a task to handle the provided activity task
Spawns off a task to handle the provided activity task
fn activity_task_handler(
        &mut self,
        worker: Arc<dyn CoreWorker>,
        activity: ActivityTask,
    ) -> Result<(), anyhow::Error> {
        match activity.variant {
            Some(activity_task::Variant::Start(start)) => {
                let act_fn = self
                    .activity_fns
                    .get(&start.activity_type)
                    .ok_or_else(|| {
                        anyhow!(
                            "No function registered for activity type {}",
                            start.activity_type
                        )
                    })?
                    .clone();
                let ct = CancellationToken::new();
                self.task_tokens_to_cancels
                    .insert(activity.task_token.clone().into(), ct.clone());
                tokio::spawn(ACT_CANCEL_TOK.scope(ct, async move {
                    let mut inputs = start.input;
                    let arg = inputs.pop().unwrap_or_default();
                    let output = (&act_fn.act_func)(arg).await;
                    let result = match output {
                        Ok(res) => ActivityExecutionResult::ok(res),
                        Err(err) => match err.downcast::<ActivityCancelledError>() {
                            Ok(ce) => ActivityExecutionResult::cancel_from_details(ce.details),
                            Err(other_err) => ActivityExecutionResult::fail(other_err.into()),
                        },
                    };
                    worker
                        .complete_activity_task(ActivityTaskCompletion {
                            task_token: activity.task_token,
                            result: Some(result),
                        })
                        .await?;
                    Result::<_, anyhow::Error>::Ok(())
                }));
            }
            Some(activity_task::Variant::Cancel(_)) => {
                if let Some(ct) = self.task_tokens_to_cancels.get(&activity.task_token.into()) {
                    ct.cancel();
                }
            }
            None => bail!("Undefined activity task variant"),
        }
        Ok(())
    }
fn activity_task_handler ( & mut self , worker : Arc < dyn CoreWorker > , activity : ActivityTask , ) -> Result < ( ) , anyhow :: Error > { match activity . variant { Some ( activity_task :: Variant :: Start ( start ) ) => { let act_fn = self . activity_fns . get ( & start . activity_type ) . ok_or_else ( | | { anyhow ! ( " " start activity_type ) } ) ? . clone ( ) ; let ct = CancellationToken :: new ( ) ; self . task_tokens_to_cancels . insert ( activity . task_token . clone ( ) . into ( ) , ct . clone ( ) ) ; tokio :: spawn ( ACT_CANCEL_TOK . scope ( ct , async move { let mut inputs = start . input ; let arg = inputs . pop ( ) . unwrap_or_default ( ) ; let output = ( & act_fn . act_func ) ( arg ) . await ; let result = match output { Ok ( res ) => ActivityExecutionResult :: ok ( res ) , Err ( err ) => match err . downcast :: < ActivityCancelledError > ( ) { Ok ( ce ) => ActivityExecutionResult :: cancel_from_details ( ce . details ) , Err ( other_err ) => ActivityExecutionResult :: fail ( other_err . into ( ) ) , } , } ; worker . complete_activity_task ( ActivityTaskCompletion { task_token : activity . task_token , result : Some ( result ) , } ) . await ? ; Result :: < _ , anyhow :: Error > :: Ok ( ( ) ) } ) ) ; } Some ( activity_task :: Variant :: Cancel ( _ ) ) => { if let Some ( ct ) = self . task_tokens_to_cancels . get ( & activity . task_token . into ( ) ) { ct . cancel ( ) ; } } None => bail ! ( " " ) , } Ok ( ( ) ) }
Spawns off a task to handle the provided activity task
Spawns off a task to handle the provided activity task
&mut self
worker
Arc<dyn CoreWorker>
activity
ActivityTask
&mut self
worker
Arc<dyn CoreWorker>
activity
ActivityTask