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

use super::{ItemType, ParsedRegisteredItem, RegisteredItem, ENV};
use crate::Error;
use serde_yaml::Value;
use std::collections::HashMap;
use tracing::trace;

/// The function takes the raw hashmap configuration item and looks up the registered
/// plugin and returns the ParsedRegisteredItem with the execution logic for used during
/// processing.
/// ```compile_fail
/// # use fiddler::config::{ConfigSpec, ItemType, ExecutionType};
/// # use fiddler::modules::inputs;
/// # inputs::register_plugins().unwrap();
/// # use std::collections::HashMap;
/// use fiddler::config::parse_configuration_item;
///
/// use serde_yaml::Value;
/// let conf_str = r#"file:
///    filename: tests/data/input.txt
///    codec: ToEnd"#;
/// let parsed_config: HashMap<String, Value> = serde_yaml::from_str(conf_str).unwrap();
///
/// # tokio_test::block_on(async {
/// parse_configuration_item(ItemType::Input, &parsed_config).unwrap();
/// # })
/// ```
pub async fn parse_configuration_item(
    itype: ItemType,
    map: &HashMap<String, Value>,
) -> Result<ParsedRegisteredItem, Error> {
    let keys: Vec<String> = map.keys().cloned().collect();
    let first_key = keys.first().ok_or(Error::ConfigFailedValidation(format!(
        "unable to determine {} key",
        itype
    )))?;
    trace!("validating item {} of type {}", first_key, itype);
    let item = get_item(&itype, first_key)?;

    let content = map
        .get(first_key)
        .ok_or(Error::ConfigFailedValidation(format!(
            "unable to validate {} key {}",
            itype, first_key
        )))?;

    let content_str = serde_yaml::to_string(content)?;
    item.format.validate(&content_str)?;
    trace!("Format for {} validated", first_key);
    Ok(ParsedRegisteredItem {
        creator: item.creator,
        config: content.clone(),
    })
}

fn get_item(itype: &ItemType, key: &String) -> Result<RegisteredItem, Error> {
    match ENV.read() {
        Ok(lock) => {
            match lock.get(itype) {
                Some(i) => {
                    if let Some(item) = i.get(key) {
                        return Ok(item.clone());
                    }
                }
                None => return Err(Error::UnableToSecureLock),
            };
        }
        Err(_) => return Err(Error::UnableToSecureLock),
    };
    Err(Error::ConfigurationItemNotFound(key.clone()))
}