durable-workflow 2.0.0-rc.34

First-party Rust SDK for Durable Workflow workers and clients
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63

use std::time::Duration;

use durable_workflow::{
    json, ChildWorkflowOptions, Error, ParallelOperation, Result, Value, WorkflowContext,
};

#[allow(dead_code)]
async fn nested_parallel(ctx: WorkflowContext) -> Result<Value> {
    let results = ctx
        .join(vec![
            ParallelOperation::activity("trip.quote-flight", json!([])),
            ParallelOperation::group(vec![
                ParallelOperation::child_workflow(
                    "trip.quote-hotel",
                    ChildWorkflowOptions::new("trip-workers"),
                    json!([]),
                ),
                ParallelOperation::timer(Duration::from_secs(1)),
            ]),
        ])
        .await?;

    Ok(json!({"top_level_members": results.len()}))
}

#[allow(dead_code)]
async fn inspect_parallel_failure(ctx: WorkflowContext) -> Result<Value> {
    match nested_parallel(ctx).await {
        Ok(value) => Ok(value),
        Err(Error::ParallelFailed(failure)) => Ok(json!({
            "failed_member_path": failure.member_path,
            "completed_members": failure.completed.len(),
            "cause": failure.cause.to_string(),
        })),
        Err(error) => Err(error),
    }
}

#[allow(dead_code)]
async fn trip_saga(ctx: WorkflowContext) -> Result<Value> {
    let mut saga = ctx.saga();
    let outcome = async {
        let flight = ctx.activity("trip.reserve-flight", json!([])).await?;
        saga.add_compensation("trip.cancel-flight", json!([flight]))?;

        let hotel = ctx.activity("trip.reserve-hotel", json!([])).await?;
        saga.add_compensation("trip.cancel-hotel", json!([hotel]))?;

        ctx.throw_if_cancellation_requested()?;
        ctx.activity("trip.charge", json!([])).await?;
        Ok(json!({"status": "booked"}))
    }
    .await;

    // Replay and worker restart resume the next ordinary compensation command.
    // A compensation failure returns Error::SagaCompensationFailed with both
    // the initiating failure and the compensation failure.
    saga.finish(outcome).await
}

fn main() {
    println!("register nested_parallel and trip_saga on a Durable Workflow worker");
}