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
use serde::{Deserialize, Serialize};
use utoipa::ToSchema;
use crate::coordination::Step;
/// Response to a completion token creation request.
#[derive(Debug, Serialize, Deserialize, ToSchema)]
pub struct CompletionTokenResponse {
/// Completion token.
///
/// An opaque string associated with the current position in the input stream
/// generated by an input connector.
/// Pass this string to the `/completion_status` endpoint to check whether all
/// inputs associated with the token have been fully processed by the pipeline.
pub token: String,
}
impl CompletionTokenResponse {
pub fn new(token: String) -> Self {
Self { token }
}
}
/// URL-encoded arguments to the `/completion_status` endpoint.
#[derive(Clone, Debug, PartialEq, Deserialize, Serialize, ToSchema)]
pub struct CompletionStatusArgs {
/// Completion token returned by the `/completion_token` or `/ingress`
/// endpoint.
pub token: String,
}
/// Completion token status returned by the `/completion_status` endpoint.
#[derive(Debug, Serialize, Deserialize, ToSchema, PartialEq, Eq)]
pub enum CompletionStatus {
/// All inputs associated with the token have been processed to completion.
#[serde(rename = "complete")]
Complete,
/// The pipeline is still processing input updates associated with the token.
#[serde(rename = "inprogress")]
InProgress,
}
/// Response to a completion token status request.
#[derive(Debug, Serialize, Deserialize, ToSchema)]
pub struct CompletionStatusResponse {
/// Completion token status.
pub status: CompletionStatus,
/// If all of the data associated with the token has been processed through
/// the pipeline, this is the final step that includes at least one record.
/// When the pipeline's `total_completed_steps` reaches this value, the
/// token has been completed.
///
/// This is `None` before the data associated with the token has been
/// processed through the pipeline.
#[schema(value_type = Option<u64>)]
pub step: Option<Step>,
}
impl CompletionStatusResponse {
pub fn new(step: Option<Step>, total_completed_steps: Step) -> Self {
let status = if let Some(step) = step
&& total_completed_steps >= step
{
CompletionStatus::Complete
} else {
CompletionStatus::InProgress
};
Self { step, status }
}
}