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

use crate::config::register_plugin;
use crate::config::ItemType;
use crate::config::{ConfigSpec, ExecutionType};
use crate::Message;
use crate::MessageBatch;
use crate::{Closer, Error, Processor};
use async_trait::async_trait;
use base64::{prelude::BASE64_STANDARD, Engine};
use fiddler_macros::fiddler_registration_func;
use serde::Deserialize;
use serde_yaml::Value;

#[derive(Clone, Default, Deserialize)]
pub struct DecoderConfig {
    algorithm: Option<Algoritym>,
}

#[derive(Clone, Default, Deserialize)]
pub enum Algoritym {
    #[default]
    Base64,
}

#[derive(Clone, Default)]
pub struct Decoder {
    algorithm: Algoritym,
}

#[async_trait]
impl Processor for Decoder {
    async fn process(&self, mut message: Message) -> Result<MessageBatch, Error> {
        match self.algorithm {
            Algoritym::Base64 => {
                let content = String::from_utf8(message.bytes)
                    .map_err(|e| Error::ProcessingError(format!("{e}")))?;
                let result = BASE64_STANDARD
                    .decode(content)
                    .map_err(|e| Error::ProcessingError(format!("{e}")))?;

                message.bytes = result;
                Ok(vec![message])
            }
        }
    }
}

impl Closer for Decoder {}

#[fiddler_registration_func]
fn create_decode(conf: Value) -> Result<ExecutionType, Error> {
    let mut proc = Decoder::default();
    let c: DecoderConfig = serde_yaml::from_value(conf)?;
    if let Some(a) = c.algorithm {
        proc.algorithm = a;
    };
    Ok(ExecutionType::Processor(Box::new(proc)))
}

pub(super) fn register_decode() -> Result<(), Error> {
    let config = "type: object";
    let conf_spec = ConfigSpec::from_schema(config)?;

    register_plugin(
        "decode".into(),
        ItemType::Processor,
        conf_spec,
        create_decode,
    )
}

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

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

    #[tokio::test]
    async fn decoded() {
        let input_str = "SGVsbG8gV29ybGQ=";
        let msg = Message {
            bytes: input_str.as_bytes().into(),
            ..Default::default()
        };
        let processor = Decoder::default();
        let output = processor.process(msg).await.unwrap();
        assert_eq!(
            output,
            vec![Message {
                bytes: "Hello World".as_bytes().into(),
                ..Default::default()
            },]
        )
    }
}