pub async fn run(&mut self) -> Result<(), anyhow::Error> {
        let (shutdown_tx, shutdown_rx) = watch::channel(false);
        let pollers = async move {
            let (common, wf_half, act_half) = self.split_apart();
            let (completions_tx, mut completions_rx) = unbounded_channel();
            let (wf_poll_res, act_poll_res) = tokio::join!(
                // Workflow polling loop
                async {
                    loop {
                        info!("Polling");
                        let activation = match common.worker.poll_workflow_activation().await {
                            Err(PollWfError::ShutDown) => {
                                break Result::<_, anyhow::Error>::Ok(());
                            }
                            o => o?,
                        };
                        wf_half
                            .workflow_activation_handler(
                                common,
                                &shutdown_rx,
                                &completions_tx,
                                &mut completions_rx,
                                activation,
                            )
                            .await?;
                    }
                },
                // Only poll on the activity queue if activity functions have been registered. This
                // makes tests which use mocks dramatically more manageable.
                async {
                    let mut shutdown_rx = shutdown_rx.clone();
                    if !act_half.activity_fns.is_empty() {
                        loop {
                            tokio::select! {
                                activity = common.worker.poll_activity_task() => {
                                    if matches!(activity, Err(PollActivityError::ShutDown)) {
                                        break;
                                    }
                                    act_half.activity_task_handler(common.worker.clone(),
                                                                   activity?)?;
                                },
                                _ = shutdown_rx.changed() => { break }
                            }
                        }
                    };
                    Result::<_, anyhow::Error>::Ok(())
                }
            );
            wf_poll_res?;
            // TODO: Activity loop errors don't show up until wf loop exits/errors
            act_poll_res?;
            Result::<_, anyhow::Error>::Ok(self)
        };

        let myself = pollers.await?;
        info!("Polling loop exited");
        let _ = shutdown_tx.send(true);
        while let Some(h) = myself.workflow_half.join_handles.next().await {
            h??;
        }
        myself.common.worker.shutdown().await;
        myself.workflow_half.workflows.clear();
        Ok(())
    }
/// Runs the worker. Eventually resolves after the worker has been explicitly shut down,
/// or may return early with an error in the event of some unresolvable problem.
Runs the worker. Eventually resolves after the worker has been explicitly shut down,
or may return early with an error in the event of some unresolvable problem.
Runs the worker . Eventually resolves after the worker has been explicitly shut down or may return early with an error in the event of some unresolvable problem .
pub async fn run(&mut self) -> Result<(), anyhow::Error> {
        let (shutdown_tx, shutdown_rx) = watch::channel(false);
        let pollers = async move {
            let (common, wf_half, act_half) = self.split_apart();
            let (completions_tx, mut completions_rx) = unbounded_channel();
            let (wf_poll_res, act_poll_res) = tokio::join!(
                // Workflow polling loop
                async {
                    loop {
                        info!("Polling");
                        let activation = match common.worker.poll_workflow_activation().await {
                            Err(PollWfError::ShutDown) => {
                                break Result::<_, anyhow::Error>::Ok(());
                            }
                            o => o?,
                        };
                        wf_half
                            .workflow_activation_handler(
                                common,
                                &shutdown_rx,
                                &completions_tx,
                                &mut completions_rx,
                                activation,
                            )
                            .await?;
                    }
                },
                // Only poll on the activity queue if activity functions have been registered. This
                // makes tests which use mocks dramatically more manageable.
                async {
                    let mut shutdown_rx = shutdown_rx.clone();
                    if !act_half.activity_fns.is_empty() {
                        loop {
                            tokio::select! {
                                activity = common.worker.poll_activity_task() => {
                                    if matches!(activity, Err(PollActivityError::ShutDown)) {
                                        break;
                                    }
                                    act_half.activity_task_handler(common.worker.clone(),
                                                                   activity?)?;
                                },
                                _ = shutdown_rx.changed() => { break }
                            }
                        }
                    };
                    Result::<_, anyhow::Error>::Ok(())
                }
            );
            wf_poll_res?;
            // TODO: Activity loop errors don't show up until wf loop exits/errors
            act_poll_res?;
            Result::<_, anyhow::Error>::Ok(self)
        };
        let myself = pollers.await?;
        info!("Polling loop exited");
        let _ = shutdown_tx.send(true);
        while let Some(h) = myself.workflow_half.join_handles.next().await {
            h??;
        }
        myself.common.worker.shutdown().await;
        myself.workflow_half.workflows.clear();
        Ok(())
    }
pub async fn run ( & mut self ) -> Result < ( ) , anyhow :: Error > { let ( shutdown_tx , shutdown_rx ) = watch :: channel ( false ) ; let pollers = async move { let ( common , wf_half , act_half ) = self . split_apart ( ) ; let ( completions_tx , mut completions_rx ) = unbounded_channel ( ) ; let ( wf_poll_res , act_poll_res ) = tokio :: join ! ( // Workflow polling loop async { loop { info ( " " ) let activation match common worker poll_workflow_activation ( ) await { Err ( PollWfError ShutDown ) { break Result anyhow Error Ok ( ( ) ) } o o } wf_half workflow_activation_handler ( common shutdown_rx completions_tx mut completions_rx activation ) await } } // Only poll on the activity queue if activity functions have been registered. This // makes tests which use mocks dramatically more manageable. async { let mut shutdown_rx shutdown_rx clone ( ) if act_half activity_fns is_empty ( ) { loop { tokio select { activity common worker poll_activity_task ( ) { if matches ( activity Err ( PollActivityError ShutDown ) ) { break } act_half activity_task_handler ( common worker clone ( ) activity ) } shutdown_rx changed ( ) { break } } } } Result anyhow Error Ok ( ( ) ) } ) ; wf_poll_res ? ; // TODO: Activity loop errors don't show up until wf loop exits/errors act_poll_res ? ; Result :: < _ , anyhow :: Error > :: Ok ( self ) } ; let myself = pollers . await ? ; info ! ( " " ) ; let _ = shutdown_tx . send ( true ) ; while let Some ( h ) = myself . workflow_half . join_handles . next ( ) . await { h ? ? ; } myself . common . worker . shutdown ( ) . await ; myself . workflow_half . workflows . clear ( ) ; Ok ( ( ) ) }
Runs the worker.
Runs the worker .
&mut self
&mut self