Struct ActivatedJob 

pub struct ActivatedJob { /* private fields */ }

Represents an activated job

Implementations

impl ActivatedJob

pub fn key(&self) -> i64

The key, a unique identifier for the job.

Returns

• i64 - The unique key of the job.

Examples found in repository

examples/custom_auto_handler.rs (line 32)

    async fn handle_result(self, client: Client, job: ActivatedJob) {
        match *self {
            Ok(_) => unreachable!("This will always fail!"),
            Err(_) => {
                let _ = client.fail_job().with_job_key(job.key()).send().await;
            }
        }
    }

examples/pizza.rs (line 196)

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(())
}

async fn confirm_order(
    _client: Client,
    job: ActivatedJob,
    state: Arc<SharedState<Mutex<Stock>>>,
) -> Result<OrderResult, WorkerError<()>> {
    let order = job
        .data::<Order>()
        .expect("For demonstration purposes we're not handling a missing order here");
    let mut stock: tokio::sync::MutexGuard<Stock> = state.lock().await;

    let mut order_accepted = true;
    let mut message = None;
    for item in order.items {
        if let Some(quantity) = stock.items.get_mut(&item) {
            if *quantity > 0 {
                *quantity -= 1;
            } else {
                message = Some(format!("We're out of stock of {}", item));
                order_accepted = false;
            }
        } else {
            message = Some(format!("We don't serve {}", item));
            order_accepted = false;
        }
    }

    println!("Confirmed order");
    Ok(OrderResult {
        order_accepted,
        message,
    })
}

async fn bake_pizzas(client: Client, job: ActivatedJob) {
    let order = job
        .data::<Order>()
        .expect("We shouldn't start baking pizzas if there is no order!");

    println!("Baking pizzas");
    sleep(Duration::from_secs(10 * order.items.len() as u64)).await;
    println!("Finished baking {} pizzas", order.items.len());

    let _ = client.complete_job().with_job_key(job.key()).send().await;
}

pub fn job_type(&self) -> &str

The type of the job (should match what was requested).

Returns

• &str - The type of the job.

pub fn process_instance_key(&self) -> i64

The job’s process instance key.

Returns

• i64 - The process instance key of the job.

pub fn bpmn_process_id(&self) -> &str

The BPMN process ID of the job process definition.

Returns

• &str - The BPMN process ID.

pub fn process_definition_version(&self) -> i32

The version of the job process definition.

Returns

• i32 - The version of the process definition.

pub fn process_definition_key(&self) -> i64

The key of the job process definition.

Returns

• i64 - The key of the process definition.

pub fn element_id(&self) -> &str

The associated task element ID.

Returns

• &str - The element ID.

pub fn element_instance_key(&self) -> i64

The unique key identifying the associated task, unique within the scope of the process instance.

Returns

• i64 - The element instance key.

pub fn custom_headers(&self) -> &str

A set of custom headers defined during modelling; returned as a serialized JSON document.

Returns

• &str - The custom headers as a JSON document.

pub fn worker(&self) -> &str

The name of the worker which activated this job.

Returns

• &str - The worker name.

pub fn retries(&self) -> i32

The amount of retries left to this job (should always be positive).

Returns

• i32 - The number of retries left.

Examples found in repository

examples/pizza.rs (line 310)

async fn deliver_order(
    _client: Client,
    job: ActivatedJob,
) -> Result<(), WorkerError<serde_json::Value>> {
    let data: Result<Customer, ClientError> = job.data();
    let customer = data.unwrap();

    if customer.address.is_empty() {
        return Err(WorkerError::ThrowError {
            error_code: String::from("invalid_address"),
            error_message: Some(String::from("Missing address")),
        });
    }

    if customer.bad_tipper && job.retries() > 1 {
        println!("Customer is a bad tipper, let's delay their order");
        sleep(Duration::from_secs(10)).await;

        return Err(WorkerError::FailJobWithData {
            error_message: String::from("Bad tipper, delaying delivery"),
            data: json!({"apology": "Oops"}),
        });
    }
    Ok(())
}

pub fn deadline(&self) -> i64

When the job can be activated again, sent as a UNIX epoch timestamp.

Returns

• i64 - The deadline as a UNIX epoch timestamp.

pub fn variables(&self) -> &str

JSON document, computed at activation time, consisting of all visible variables to the task scope.

Returns

• &str - The variables as a JSON document.

pub fn data<T: DeserializeOwned>(&self) -> Result<T, ClientError>

Deserializes the variables JSON document into a specified type.

Type Parameters

• T - The type to deserialize into.

Returns

• Result<T, ClientError> - The deserialized data or an error if deserialization fails.

Examples found in repository

examples/pizza.rs (line 190)

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(())
}

async fn confirm_order(
    _client: Client,
    job: ActivatedJob,
    state: Arc<SharedState<Mutex<Stock>>>,
) -> Result<OrderResult, WorkerError<()>> {
    let order = job
        .data::<Order>()
        .expect("For demonstration purposes we're not handling a missing order here");
    let mut stock: tokio::sync::MutexGuard<Stock> = state.lock().await;

    let mut order_accepted = true;
    let mut message = None;
    for item in order.items {
        if let Some(quantity) = stock.items.get_mut(&item) {
            if *quantity > 0 {
                *quantity -= 1;
            } else {
                message = Some(format!("We're out of stock of {}", item));
                order_accepted = false;
            }
        } else {
            message = Some(format!("We don't serve {}", item));
            order_accepted = false;
        }
    }

    println!("Confirmed order");
    Ok(OrderResult {
        order_accepted,
        message,
    })
}

async fn bake_pizzas(client: Client, job: ActivatedJob) {
    let order = job
        .data::<Order>()
        .expect("We shouldn't start baking pizzas if there is no order!");

    println!("Baking pizzas");
    sleep(Duration::from_secs(10 * order.items.len() as u64)).await;
    println!("Finished baking {} pizzas", order.items.len());

    let _ = client.complete_job().with_job_key(job.key()).send().await;
}

async fn deliver_order(
    _client: Client,
    job: ActivatedJob,
) -> Result<(), WorkerError<serde_json::Value>> {
    let data: Result<Customer, ClientError> = job.data();
    let customer = data.unwrap();

    if customer.address.is_empty() {
        return Err(WorkerError::ThrowError {
            error_code: String::from("invalid_address"),
            error_message: Some(String::from("Missing address")),
        });
    }

    if customer.bad_tipper && job.retries() > 1 {
        println!("Customer is a bad tipper, let's delay their order");
        sleep(Duration::from_secs(10)).await;

        return Err(WorkerError::FailJobWithData {
            error_message: String::from("Bad tipper, delaying delivery"),
            data: json!({"apology": "Oops"}),
        });
    }
    Ok(())
}

pub fn tenant_id(&self) -> &str

The id of the tenant that owns the job.

Returns

• &str - The tenant ID.

Trait Implementations

impl Clone for ActivatedJob

fn clone(&self) -> ActivatedJob

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for ActivatedJob

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.