aion-rs 0.10.0

Transport-agnostic Aion workflow engine with durability, replay, timers, and supervision.
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
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168

use super::support::*;

pub(in super::super) fn reenvelope(event: Event, workflow_id: &WorkflowId, seq: u64) -> Event {
    let envelope = EventEnvelope {
        seq,
        recorded_at: chrono::Utc::now(),
        workflow_id: workflow_id.clone(),
    };
    match event {
        Event::WorkflowStarted {
            workflow_type,
            input,
            run_id,
            parent_run_id,
            package_version,
            ..
        } => Event::WorkflowStarted {
            envelope,
            workflow_type,
            input,
            run_id,
            parent_run_id,
            package_version,
        },
        Event::ActivityScheduled {
            activity_id,
            activity_type,
            input,
            task_queue,
            node,
            ..
        } => Event::ActivityScheduled {
            envelope,
            activity_id,
            activity_type,
            input,
            task_queue,
            node,
        },
        Event::ActivityStarted {
            activity_id,
            attempt,
            ..
        } => Event::ActivityStarted {
            envelope,
            activity_id,
            attempt,
        },
        Event::ActivityCompleted {
            activity_id,
            result,
            attempt,
            ..
        } => Event::ActivityCompleted {
            envelope,
            activity_id,
            result,
            attempt,
        },
        Event::ActivityFailed {
            activity_id,
            error,
            attempt,
            ..
        } => Event::ActivityFailed {
            envelope,
            activity_id,
            error,
            attempt,
        },
        Event::ActivityCancelled {
            activity_id,
            attempt,
            ..
        } => Event::ActivityCancelled {
            envelope,
            activity_id,
            attempt,
        },
        Event::TimerFired { timer_id, .. } => Event::TimerFired { envelope, timer_id },
        Event::SearchAttributesUpdated {
            workflow_id: attribute_workflow_id,
            attributes,
            ..
        } => Event::SearchAttributesUpdated {
            envelope,
            workflow_id: attribute_workflow_id,
            attributes,
        },
        other => other,
    }
}

pub(in super::super) fn started_event(
    workflow_id: &WorkflowId,
    run_id: &RunId,
) -> Result<Event, Box<dyn std::error::Error>> {
    Ok(Event::WorkflowStarted {
        envelope: EventEnvelope {
            seq: 1,
            recorded_at: chrono::Utc::now(),
            workflow_id: workflow_id.clone(),
        },
        workflow_type: "collect-parent".to_owned(),
        input: Payload::from_json(&json!({ "fixture": "input" }))?,
        run_id: run_id.clone(),
        parent_run_id: None,
        package_version: aion_core::PackageVersion::new("a".repeat(64)),
    })
}

pub(in super::super) fn placeholder_envelope() -> EventEnvelope {
    EventEnvelope {
        seq: 0,
        recorded_at: chrono::Utc::now(),
        workflow_id: WorkflowId::new_v4(),
    }
}

pub(in super::super) fn scheduled_started(ordinal: u64, name: &str) -> Vec<Event> {
    vec![
        Event::ActivityScheduled {
            envelope: placeholder_envelope(),
            activity_id: ActivityId::from_sequence_position(ordinal),
            activity_type: name.to_owned(),
            input: Payload::new(ContentType::Json, br#""in""#.to_vec()),
            task_queue: String::from("default"),
            node: None,
        },
        Event::ActivityStarted {
            envelope: placeholder_envelope(),
            activity_id: ActivityId::from_sequence_position(ordinal),
            attempt: 1,
        },
    ]
}

pub(in super::super) fn completed(ordinal: u64, result: &str) -> Event {
    Event::ActivityCompleted {
        envelope: placeholder_envelope(),
        activity_id: ActivityId::from_sequence_position(ordinal),
        result: Payload::new(ContentType::Json, result.as_bytes().to_vec()),
        attempt: 1,
    }
}

pub(in super::super) fn scheduled_started_on(
    ordinal: u64,
    name: &str,
    task_queue: &str,
    node: Option<&str>,
) -> Vec<Event> {
    vec![
        Event::ActivityScheduled {
            envelope: placeholder_envelope(),
            activity_id: ActivityId::from_sequence_position(ordinal),
            activity_type: name.to_owned(),
            input: Payload::new(ContentType::Json, br#""in""#.to_vec()),
            task_queue: task_queue.to_owned(),
            node: node.map(str::to_owned),
        },
        Event::ActivityStarted {
            envelope: placeholder_envelope(),
            activity_id: ActivityId::from_sequence_position(ordinal),
            attempt: 1,
        },
    ]
}