pipedash_plugin_api/
support.rs

1use std::collections::BTreeSet;
2use std::future::Future;
3
4use futures::stream::{
5    self,
6    StreamExt,
7    TryStreamExt,
8};
9use serde::de::DeserializeOwned;
10
11use crate::error::{
12    PluginError,
13    PluginResult,
14};
15use crate::health::ClientOptions;
16use crate::logs::{
17    LogCursor,
18    LogSink,
19    RunRef,
20};
21use crate::types::{
22    AvailablePipeline,
23    Organization,
24    TriggerParams,
25    WorkflowParameter,
26    WorkflowParameterType,
27};
28
29pub const CONCURRENCY: usize = 8;
30pub const LOG_CHUNK_BYTES: usize = 64 * 1024;
31
32pub fn run_metadata_str<'a>(run: &RunRef<'a>, key: &str) -> Option<&'a str> {
33    run.metadata
34        .get(key)
35        .and_then(serde_json::Value::as_str)
36        .filter(|value| !value.is_empty())
37}
38
39pub fn client_options(origin: Option<&str>, insecure: bool) -> ClientOptions {
40    ClientOptions {
41        origin: origin.and_then(|url| reqwest::Url::parse(url).ok()),
42        accept_invalid_certs: insecure,
43        ..ClientOptions::default()
44    }
45}
46
47pub fn input_str<'a>(params: &'a TriggerParams, key: &str) -> Option<&'a str> {
48    params
49        .inputs
50        .as_ref()
51        .and_then(|inputs| inputs.get(key))
52        .and_then(serde_json::Value::as_str)
53}
54
55pub fn string_parameter(
56    name: &str, label: &str, description: &str, default: Option<String>, required: bool,
57) -> WorkflowParameter {
58    WorkflowParameter {
59        name: name.to_string(),
60        label: Some(label.to_string()),
61        description: Some(description.to_string()),
62        param_type: WorkflowParameterType::String { default },
63        required,
64    }
65}
66
67pub async fn write_in_chunks<I>(
68    sink: &mut LogSink, items: impl IntoIterator<Item = I>,
69    mut render: impl FnMut(&mut Vec<u8>, I) -> PluginResult<()>,
70) -> PluginResult<()> {
71    let mut chunk = Vec::new();
72    for item in items {
73        render(&mut chunk, item)?;
74        if chunk.len() >= LOG_CHUNK_BYTES {
75            sink.write(std::mem::take(&mut chunk).into()).await?;
76        }
77    }
78    sink.write(chunk.into()).await
79}
80
81pub fn organizations<'a>(
82    names: impl IntoIterator<Item = &'a str>, description: Option<&str>,
83) -> Vec<Organization> {
84    names
85        .into_iter()
86        .collect::<BTreeSet<_>>()
87        .into_iter()
88        .map(|name| Organization {
89            id: name.to_string(),
90            name: name.to_string(),
91            description: description.map(str::to_string),
92        })
93        .collect()
94}
95
96pub fn require_configured<T>(items: Vec<T>, what: &str) -> PluginResult<Vec<T>> {
97    if items.is_empty() {
98        return Err(PluginError::InvalidConfig(format!("No {what} configured")));
99    }
100    Ok(items)
101}
102
103pub fn retain_matching(items: &mut Vec<AvailablePipeline>, search: &str) {
104    let search = search.to_lowercase();
105    items.retain(|pipeline| {
106        pipeline.name.to_lowercase().contains(&search)
107            || pipeline.id.to_lowercase().contains(&search)
108            || pipeline
109                .description
110                .as_ref()
111                .is_some_and(|description| description.to_lowercase().contains(&search))
112    });
113}
114
115pub async fn join_buffered<F: Future>(requests: Vec<F>, limit: usize) -> Vec<F::Output> {
116    stream::iter(requests).buffered(limit).collect().await
117}
118
119pub async fn try_buffered<F, T>(requests: Vec<F>, limit: usize) -> PluginResult<Vec<T>>
120where
121    F: Future<Output = PluginResult<T>>,
122{
123    stream::iter(requests).buffered(limit).try_collect().await
124}
125
126pub async fn collect_tolerant<F, T>(
127    requests: Vec<F>, limit: usize, mut on_error: impl FnMut(&PluginError),
128) -> PluginResult<Vec<T>>
129where
130    F: Future<Output = PluginResult<T>>,
131{
132    let results: Vec<_> = stream::iter(requests).buffered(limit).collect().await;
133    let mut items = Vec::new();
134    let mut first_error = None;
135
136    for result in results {
137        match result {
138            Ok(item) => items.push(item),
139            Err(error) if error.is_transient() => return Err(error),
140            Err(error) => {
141                on_error(&error);
142                first_error.get_or_insert(error);
143            }
144        }
145    }
146
147    match first_error {
148        Some(error) if items.is_empty() => Err(error),
149        _ => Ok(items),
150    }
151}
152
153pub fn resume_position<T: DeserializeOwned>(
154    from: Option<&LogCursor>, sink: &mut LogSink, keep: impl FnOnce(&T) -> bool,
155) -> PluginResult<Option<T>> {
156    let position = from
157        .and_then(|cursor| cursor.parse::<T>())
158        .filter(|position| keep(position));
159    if from.is_some() && position.is_none() {
160        sink.restart()?;
161    }
162    Ok(position)
163}
164
165pub fn digit_id<'a>(id: &'a str, what: &str) -> PluginResult<&'a str> {
166    checked_id(id, what, |byte| byte.is_ascii_digit())
167}
168
169pub fn numeric_id<T: std::str::FromStr>(id: &str, what: &str) -> PluginResult<T> {
170    digit_id(id, what)?
171        .parse()
172        .map_err(|_| PluginError::InvalidConfig(format!("Invalid {what}: {id}")))
173}
174
175pub fn uuid_id<'a>(id: &'a str, what: &str) -> PluginResult<&'a str> {
176    checked_id(id, what, |byte| byte.is_ascii_hexdigit() || byte == b'-')
177}
178
179fn checked_id<'a>(id: &'a str, what: &str, valid: fn(u8) -> bool) -> PluginResult<&'a str> {
180    if id.is_empty() || !id.bytes().all(valid) {
181        return Err(PluginError::InvalidConfig(format!("Invalid {what}: {id}")));
182    }
183    Ok(id)
184}
185
186#[cfg(test)]
187mod tests;
pipedash_plugin_api/support.rs

pipedash_plugin_api/
support.rs
