emmett 0.1.3-alpha

a simple etl tool
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

/// Specification: https://www.elastic.co/guide/en/logstash/current/plugins-filters-geoip.html
use reqwest::{Client, RedirectPolicy};
use serde_json::{json, value::Value};
use std::path::Path;

#[allow(unused)]
impl<'a> Geoip<'a> {
    pub fn process(self, input: Value) -> Result<Value, ()> {
        if let Some(source) = input.get(&self.source) {
            let source = source
                .as_str()
                .expect("Couldn't parse Geoip source as string.");

            let value = match ip_api(source) {
                Ok(v) => v,
                Err(e) => panic!("{}", e),
            };

            let output_message = json!({ self.target: value });
            Ok(output_message)
        } else {
            Ok(input)
        }
    }
}

#[derive(Debug)]
#[allow(unused)]
pub struct Geoip<'a> {
    cache_size: Option<u64>,
    database: Option<&'a Path>,
    default_database_type: Option<String>,
    fields: Option<Vec<String>>,
    source: String,
    tag_on_failure: Option<Vec<String>>,
    target: String,
}

impl<'a> Default for Geoip<'a> {
    fn default() -> Self {
        Self {
            cache_size: Some(1000),
            database: None,
            default_database_type: Some("City".to_string()),
            fields: None,
            source: "".to_string(),
            tag_on_failure: Some(vec!["_geoip_lookup_failure".to_string()]),
            target: "geoip".to_string(),
        }
    }
}

#[allow(unused)]
fn ip_api(ip: &str) -> Result<Value, reqwest::Error> {
    let client = Client::builder()
        .redirect(RedirectPolicy::limited(10))
        .build()
        .expect("Couldn't build Reqwest client.");

    let uri = format!("http://ip-api.com/json/{}", ip);

    let res = client.get(&uri).send();

    res?.json()
}