Struct Worker 

pub struct Worker<H>
where
    H: JobHandler + Send + Sync + 'static,
{ /* private fields */ }

The Worker is responsible for fetching jobs from Zeebe and processing them with the associated handler. /// A worker implementation for processing Zeebe jobs with configurable concurrency and state management.

The Worker is responsible for:

• Polling for new jobs from the Zeebe broker
• Managing job activation and processing
• Handling concurrent job execution
• Maintaining worker state across job executions

The worker consists of two main components:

• WorkProducer: Handles job polling and queue management
• WorkConsumer: Manages job execution and concurrency

Architecture

The worker uses a producer-consumer pattern where:

1. The producer polls for jobs at regular intervals
2. Jobs are queued in an internal channel
3. The consumer processes jobs concurrently up to the configured limit

Concurrency

Job processing is controlled by:

• A semaphore limiting concurrent job executions
• Channel-based communication between components
• Configurable maximum jobs to activate

Example

let worker = client
    .worker()
    .with_job_timeout(Duration::from_secs(60))
    .with_request_timeout(Duration::from_secs(10))
    .with_max_jobs_to_activate(5)
    .with_concurrency_limit(3)
    .with_job_type("example-service")
    .with_handler(|client, job| async move {
        // Process job here
        client.complete_job().with_job_key(job.key()).send().await;
    })
    .build();

// Start the worker
worker.run().await?;

Error Handling

The worker implements automatic error handling for:

• Job activation timeouts
• Network errors during polling
• Job processing failures

Implementations

impl<H> Worker<H>

where H: JobHandler + Send + Sync + 'static,

pub async fn run(self)

Starts the worker by running both the poller and dispatcher concurrently.

This method uses tokio::join! to run the poller and dispatcher concurrently. The poller continuously polls the Zeebe broker for new jobs, while the dispatcher processes the jobs using the provided callback.

Example

#[tokio::main]
async fn main() {
    client
        .worker()
        //Worker configuration...
        .build()
        .run()
        .await;
}

Examples found in repository

examples/custom_auto_handler.rs (line 76)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    unsafe { std::env::set_var("RUST_BACKTRACE", "1") };

    // Default configuration using Camunda's docker compose
    // https://github.com/camunda/camunda-platform/blob/5dc74fe71667e18fbb5c8d4694068d662d83ad00/README.md
    let client = Client::builder()
        .with_address("http://localhost", 26500)
        .with_oauth(
            String::from("zeebe"),
            String::from("zecret"),
            String::from(
                "http://localhost:18080/auth/realms/camunda-platform/protocol/openid-connect/token",
            ),
            String::from("zeebe-api"),
            Duration::from_secs(30),
            None,
        )
        .build()
        .await?;

    // Wait until first OAuth token has been retrieved
    client.auth_initialized().await;

    let worker = client
        .worker()
        .with_request_timeout(Duration::from_secs(1))
        .with_job_timeout(Duration::from_secs(1))
        .with_max_jobs_to_activate(1)
        .with_concurrency_limit(1)
        .with_job_type(String::from("placeholder"))
        .with_handler(always_fail)
        .build();

    tokio::join!(worker.run());

    Ok(())
}

examples/pizza.rs (line 241)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    unsafe { std::env::set_var("RUST_BACKTRACE", "1") };

    // Default configuration using Camunda's docker compose
    // https://github.com/camunda/camunda-platform/blob/5dc74fe71667e18fbb5c8d4694068d662d83ad00/README.md
    let client = Client::builder()
        .with_address("http://localhost", 26500)
        .with_oauth(
            String::from("zeebe"),
            String::from("zecret"),
            String::from(
                "http://localhost:18080/auth/realms/camunda-platform/protocol/openid-connect/token",
            ),
            String::from("zeebe-api"),
            Duration::from_secs(30),
            None,
        )
        .build()
        .await?;

    // Wait until first OAuth token has been retrieved
    client.auth_initialized().await;

    // Deploy our pizza ordering process
    // We just discard the result for brevity
    let deploy_resource_response = client
        .deploy_resource()
        .with_resource_file(PathBuf::from("examples/resources/pizza-order.bpmn"))
        .read_resource_files()?
        .send()
        .await?;

    let process_definition_key =
        if let Some(deployment) = deploy_resource_response.deployments().first() {
            match deployment.metadata() {
                Some(metadata) => match metadata {
                    zeebe_rs::Metadata::Process(process_metadata) => {
                        process_metadata.process_definition_key()
                    }
                    _ => {
                        unreachable!("We're only deploying a bpmn here");
                    }
                },
                _ => -1,
            }
        } else {
            -1
        };

    // Let's define our shared state, how much stock we have
    let mut initial_stock = HashMap::new();
    initial_stock.insert(String::from("Pepperoni"), 5);
    initial_stock.insert(String::from("Margherita"), 8);
    initial_stock.insert(String::from("Hawaiian"), 3);
    initial_stock.insert(String::from("Quattro Formaggi"), 0);
    initial_stock.insert(String::from("Vegetarian"), 6);

    let stock = Arc::new(SharedState(Mutex::new(Stock {
        items: initial_stock,
    })));

    // This will share the stock among all the workers confirming orders
    // counting it down for each order.
    let confirm_worker = client
        .worker()
        .with_request_timeout(Duration::from_secs(10))
        .with_job_timeout(Duration::from_secs(10))
        .with_max_jobs_to_activate(4)
        .with_concurrency_limit(2)
        .with_job_type(String::from("confirm_order"))
        .with_state(stock)
        .with_handler(confirm_order)
        .with_fetch_variable(String::from("items"))
        .build();

    // We can reuse the worker builder for sharing some configurations
    let base_worker = client
        .worker()
        .with_request_timeout(Duration::from_secs(10))
        .with_job_timeout(Duration::from_secs(10));

    let bake_worker = base_worker
        .clone()
        .with_max_jobs_to_activate(10)
        .with_concurrency_limit(10)
        .with_job_type(String::from("bake_pizzas"))
        .with_handler(bake_pizzas)
        .build();

    // Handler can be both function callbacks or closures
    // Handlers have to share return type for this to work
    let reject_order_worker = client
        .worker()
        .with_request_timeout(Duration::from_secs(10))
        .with_job_timeout(Duration::from_secs(10))
        .with_max_jobs_to_activate(5)
        .with_concurrency_limit(5)
        .with_job_type(String::from("reject_order"))
        .with_handler(|_client, _job| async move { Ok::<(), WorkerError<()>>(()) })
        .build();

    let deliver_worker = client
        .worker()
        .with_request_timeout(Duration::from_secs(10))
        .with_job_timeout(Duration::from_secs(10))
        .with_max_jobs_to_activate(1)
        .with_concurrency_limit(1)
        .with_job_type(String::from("deliver_order"))
        .with_handler(deliver_order)
        .build();

    let clarify_address_worker = base_worker
        .clone()
        .with_max_jobs_to_activate(5)
        .with_concurrency_limit(5)
        .with_job_type(String::from("call_customer"))
        .with_handler(|client, job| async move {
            let mut customer = job.data::<Customer>().unwrap();

            //Same guy keeps leaving out his address for some reason
            customer.address = String::from("1337 Coolsville, USA");
            if let Ok(req) = client
                .complete_job()
                .with_job_key(job.key())
                .with_variables(customer)
            {
                let _ = req.send().await;
            }
        })
        .build();

    tokio::spawn(place_order(
        client.clone(),
        process_definition_key,
        "Jane Doe",
        "100 Coolsville, USA",
        true,
        vec!["Pepperoni"],
    ));

    tokio::spawn(place_order(
        client.clone(),
        process_definition_key,
        "John Doe",
        "999 NotCoolsville, USA",
        false,
        vec!["Quattro Formaggi"; 100],
    ));

    tokio::spawn(place_order(
        client.clone(),
        process_definition_key,
        "Lorem Ipsum",
        "",
        false,
        vec!["Hawaiian", "Pepperoni"],
    ));

    tokio::spawn(place_order(
        client.clone(),
        process_definition_key,
        "George W Bush",
        "White House",
        true,
        vec!["Burger"],
    ));

    let _ = tokio::join!(
        confirm_worker.run(),
        reject_order_worker.run(),
        bake_worker.run(),
        deliver_worker.run(),
        clarify_address_worker.run(),
    );

    Ok(())
}