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
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
use crate::{client::Client, proto, Error, Result};
use tracing::{debug, trace};

/// An activate Zeebe job that is ready to be worked on by a worker.
#[derive(Clone, Debug)]
pub struct Job(proto::ActivatedJob);

impl Job {
    /// Create a new job from a GRPC response
    pub(crate) fn new(proto: proto::ActivatedJob) -> Self {
        Job(proto)
    }

    /// the key, a unique identifier for the job
    pub fn key(&self) -> i64 {
        self.0.key
    }

    /// the type of the job (should match what was requested)
    pub fn job_type(&self) -> &str {
        &self.0.r#type
    }

    /// the job's process instance key
    pub fn process_instance_key(&self) -> i64 {
        self.0.process_instance_key
    }

    /// the bpmn process ID of the job process definition
    pub fn bpmn_process_id(&self) -> &str {
        &self.0.bpmn_process_id
    }

    /// the version of the job process definition
    pub fn process_definition_version(&self) -> i32 {
        self.0.process_definition_version
    }

    /// the key of the job process definition
    pub fn process_definition_key(&self) -> i64 {
        self.0.process_definition_key
    }

    /// the associated task element ID
    pub fn element_id(&self) -> &str {
        &self.0.element_id
    }

    /// the unique key identifying the associated task, unique within the scope of
    /// the process instance
    pub fn element_instance_key(&self) -> i64 {
        self.0.element_instance_key
    }

    /// a set of custom headers defined during modelling; returned as a serialized
    /// JSON document
    pub fn custom_headers(&self) -> &str {
        &self.0.custom_headers
    }

    /// the name of the worker which activated this job
    pub fn worker(&self) -> &str {
        &self.0.worker
    }

    /// the amount of retries left to this job (should always be positive)
    pub fn retries(&self) -> i32 {
        self.0.retries
    }

    /// when the job can be activated again, sent as a UNIX epoch timestamp
    pub fn deadline(&self) -> i64 {
        self.0.deadline
    }

    /// Serialized JSON document, computed at activation time, consisting of all
    /// visible variables to the task scope
    pub fn variables_str(&self) -> &str {
        &self.0.variables
    }

    /// JSON document, computed at activation time, consisting of all visible
    /// variables to the task scope
    pub fn variables(&self) -> serde_json::Value {
        serde_json::from_str(&self.0.variables).unwrap_or_else(|_| serde_json::json!({}))
    }

    /// Deserialize encoded json variables as a given type
    pub fn variables_as<'a, T: serde::de::Deserialize<'a>>(&'a self) -> Option<T> {
        serde_json::from_str::<'a, T>(&self.0.variables).ok()
    }
}

/// Configuration to complete a job
#[derive(Debug)]
pub struct CompleteJobBuilder {
    client: Client,
    job_key: Option<i64>,
    variables: Option<serde_json::Value>,
}

impl CompleteJobBuilder {
    /// Create a new complete job builder.
    pub fn new(client: Client) -> Self {
        CompleteJobBuilder {
            client,
            job_key: None,
            variables: None,
        }
    }

    /// Set the unique job identifier, as obtained from [`Job::key`].
    pub fn with_job_key(self, job_key: i64) -> Self {
        CompleteJobBuilder {
            job_key: Some(job_key),
            ..self
        }
    }

    /// Set the JSON document representing the variables in the current task scope.
    pub fn with_variables<T: Into<serde_json::Value>>(self, variables: T) -> Self {
        CompleteJobBuilder {
            variables: Some(variables.into()),
            ..self
        }
    }

    /// Submit the complete job request.
    #[tracing::instrument(skip(self), name = "complete_job", err)]
    pub async fn send(mut self) -> Result<CompleteJobResponse> {
        if self.job_key.is_none() && self.client.current_job_key.is_none() {
            return Err(Error::InvalidParameters("`job_key` must be set"));
        }
        let req = proto::CompleteJobRequest {
            job_key: self.job_key.or(self.client.current_job_key).unwrap(),
            variables: self
                .variables
                .map_or(String::new(), |vars| vars.to_string()),
        };

        debug!(job_key = req.job_key, "completing job:");
        trace!(?req, "request:");
        let res = self
            .client
            .gateway_client
            .complete_job(tonic::Request::new(req))
            .await?;

        Ok(CompleteJobResponse(res.into_inner()))
    }
}

/// Completed job instance data.
#[derive(Debug)]
pub struct CompleteJobResponse(proto::CompleteJobResponse);

/// Configuration to fail a job
#[derive(Debug)]
pub struct FailJobBuilder {
    client: Client,
    job_key: Option<i64>,
    retries: Option<u32>,
    error_message: Option<String>,
}

impl FailJobBuilder {
    /// Create a new fail job builder.
    pub fn new(client: Client) -> Self {
        FailJobBuilder {
            client,
            job_key: None,
            retries: None,
            error_message: None,
        }
    }

    /// Set the unique job identifier, as obtained from [`Job::key`].
    pub fn with_job_key(self, job_key: i64) -> Self {
        FailJobBuilder {
            job_key: Some(job_key),
            ..self
        }
    }

    /// Set the amount of retries the job should have left.
    pub fn with_retries(self, retries: u32) -> Self {
        FailJobBuilder {
            retries: Some(retries),
            ..self
        }
    }

    /// Set an optional message describing why the job failed. This is particularly
    /// useful if a job runs out of retries and an incident is raised, as it this
    /// message can help explain why an incident was raised.
    pub fn with_error_message<T: Into<String>>(self, error_message: T) -> Self {
        FailJobBuilder {
            error_message: Some(error_message.into()),
            ..self
        }
    }

    /// Submit the fail job request.
    #[tracing::instrument(skip(self), name = "fail_job", err)]
    pub async fn send(mut self) -> Result<FailJobResponse> {
        if self.job_key.is_none() && self.client.current_job_key.is_none() {
            return Err(Error::InvalidParameters("`job_key` must be set"));
        }
        let req = proto::FailJobRequest {
            job_key: self.job_key.or(self.client.current_job_key).unwrap(),
            retries: self.retries.unwrap_or_default() as i32,
            error_message: self.error_message.unwrap_or_default(),
        };

        debug!(job_key = req.job_key, "failing job:");
        trace!(?req, "request:");
        let res = self
            .client
            .gateway_client
            .fail_job(tonic::Request::new(req))
            .await?;

        Ok(FailJobResponse(res.into_inner()))
    }
}

/// Failed job instance data.
#[derive(Debug)]
pub struct FailJobResponse(proto::FailJobResponse);

/// Configuration to throw an error in the context of a job.
#[derive(Debug)]
pub struct ThrowErrorBuilder {
    client: Client,
    job_key: Option<i64>,
    error_code: Option<String>,
    error_message: Option<String>,
}

impl ThrowErrorBuilder {
    /// Create a new throw error builder.
    pub fn new(client: Client) -> Self {
        ThrowErrorBuilder {
            client,
            job_key: None,
            error_code: None,
            error_message: None,
        }
    }

    /// Set the unique job identifier, as obtained from [`Job::key`].
    pub fn with_job_key(self, job_key: i64) -> Self {
        ThrowErrorBuilder {
            job_key: Some(job_key),
            ..self
        }
    }

    /// Set the error code that will be matched with an error catch event.
    pub fn with_error_code<T: Into<String>>(self, error_code: T) -> Self {
        ThrowErrorBuilder {
            error_code: Some(error_code.into()),
            ..self
        }
    }

    /// Set an optional message describing why the error was thrown.
    pub fn with_error_message<T: Into<String>>(self, error_message: T) -> Self {
        ThrowErrorBuilder {
            error_message: Some(error_message.into()),
            ..self
        }
    }

    /// Submit the throw error request.
    #[tracing::instrument(skip(self), name = "throw_error", err)]
    pub async fn send(mut self) -> Result<ThrowErrorResponse> {
        if self.job_key.is_none() && self.client.current_job_key.is_none() {
            return Err(Error::InvalidParameters("`job_key` must be set"));
        }
        let req = proto::ThrowErrorRequest {
            job_key: self.job_key.or(self.client.current_job_key).unwrap(),
            error_code: self.error_code.unwrap_or_default(),
            error_message: self.error_message.unwrap_or_default(),
        };

        debug!(?req, "sending request:");
        let res = self
            .client
            .gateway_client
            .throw_error(tonic::Request::new(req))
            .await?;

        Ok(ThrowErrorResponse(res.into_inner()))
    }
}

/// Throw error response data.
#[derive(Debug)]
pub struct ThrowErrorResponse(proto::ThrowErrorResponse);

/// Updates the number of retries a job has left. This is mostly useful for jobs
/// that have run out of retries, should the underlying problem be solved.
#[derive(Debug)]
pub struct UpdateJobRetriesBuilder {
    client: Client,
    job_key: Option<i64>,
    retries: Option<u32>,
}

impl UpdateJobRetriesBuilder {
    /// Create a new update retries builder.
    pub fn new(client: Client) -> Self {
        UpdateJobRetriesBuilder {
            client,
            job_key: None,
            retries: None,
        }
    }

    /// Set the unique job identifier, as obtained from [`Job::key`].
    pub fn with_job_key(self, job_key: i64) -> Self {
        UpdateJobRetriesBuilder {
            job_key: Some(job_key),
            ..self
        }
    }

    /// Set the new amount of retries for the job
    pub fn with_retries(self, retries: u32) -> Self {
        UpdateJobRetriesBuilder {
            retries: Some(retries),
            ..self
        }
    }

    /// Submit the update job retries request.
    #[tracing::instrument(skip(self), name = "update_job_retries", err)]
    pub async fn send(mut self) -> Result<UpdateJobRetriesResponse> {
        if (self.job_key.is_none() && self.client.current_job_key.is_none())
            || self.retries.is_none()
        {
            return Err(Error::InvalidParameters(
                "`job_key` and `retries` must be set",
            ));
        }
        let req = proto::UpdateJobRetriesRequest {
            job_key: self.job_key.or(self.client.current_job_key).unwrap(),
            retries: self.retries.unwrap() as i32,
        };

        debug!(?req, "sending request:");
        let res = self
            .client
            .gateway_client
            .update_job_retries(tonic::Request::new(req))
            .await?;

        Ok(UpdateJobRetriesResponse(res.into_inner()))
    }
}

/// Update job retries data.
#[derive(Debug)]
pub struct UpdateJobRetriesResponse(proto::UpdateJobRetriesResponse);