fiddler 4.9.1

Data Stream processor written in rust
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

use crate::config::register_plugin;
use crate::config::{parse_configuration_item, Item, ItemType};
use crate::config::{ConfigSpec, ExecutionType};
use crate::Message;
use crate::MessageBatch;
use crate::{Closer, Error, Processor};
use async_trait::async_trait;
use fiddler_macros::fiddler_registration_func;
use serde::Deserialize;
use serde_yaml::Value;
use tracing::debug;

#[derive(Deserialize)]
struct TryConfig {
    processor: Value,
    catch: Option<Vec<Item>>,
}
pub struct Try {
    processor: Box<dyn Processor + Send + Sync>,
    catch: Vec<Box<dyn Processor + Send + Sync>>,
}

#[async_trait]
impl Processor for Try {
    async fn process(&self, message: Message) -> Result<MessageBatch, Error> {
        match self.processor.process(message.clone()).await {
            Ok(m) => Ok(m),
            Err(e) => {
                debug!("caught error {e}");
                let mut messages = vec![message];
                for p in &self.catch {
                    let mut new_messages = Vec::new();
                    for m in messages.drain(..) {
                        let processed = p.process(m).await?;
                        new_messages.extend(processed);
                    }
                    messages = new_messages;
                }
                Ok(messages)
            }
        }
    }
}

#[async_trait]
impl Closer for Try {
    async fn close(&mut self) -> Result<(), Error> {
        self.processor.close().await?;
        for p in &mut self.catch {
            p.close().await?;
        }
        Ok(())
    }
}

#[fiddler_registration_func]
fn create_try(conf: Value) -> Result<ExecutionType, Error> {
    let try_conf: TryConfig = serde_yaml::from_value(conf.clone())?;
    let proc: Item = serde_yaml::from_value(try_conf.processor)?;

    let proc_registered_item = parse_configuration_item(ItemType::Processor, &proc.extra).await?;
    let p = match ((proc_registered_item.creator)(proc_registered_item.config)).await? {
        ExecutionType::Processor(rp) => rp,
        _ => {
            return Err(Error::ConfigFailedValidation(
                "invalid execution type provided".into(),
            ))
        }
    };

    let catch: Vec<Box<dyn Processor + Send + Sync>> = match try_conf.catch {
        Some(processors) => {
            let mut steps = Vec::new();
            for p in processors {
                let ri = parse_configuration_item(ItemType::Processor, &p.extra).await?;
                let proc = ((ri.creator)(ri.config.clone())).await?;
                if let ExecutionType::Processor(rp) = proc {
                    steps.push(rp);
                };
            }
            steps
        }
        None => Vec::new(),
    };

    Ok(ExecutionType::Processor(Box::new(Try {
        processor: p,
        catch,
    })))
}

pub(super) fn register_try() -> Result<(), Error> {
    let config = "type: object
properties:
  processor: 
    type: object
  catch:
    type: array
required:
  - processor";

    let conf_spec = ConfigSpec::from_schema(config)?;

    register_plugin("try".into(), ItemType::Processor, conf_spec, create_try)
}

#[cfg(test)]
mod test {
    use super::*;

    #[test]
    fn register_plugin() {
        register_try().unwrap()
    }
}