springql-core 0.18.1

SpringQL: Open-source stream processor for IoT devices and in-vehicle computers
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

// This file is part of https://github.com/SpringQL/SpringQL which is licensed under MIT OR Apache-2.0. See file LICENSE-MIT or LICENSE-APACHE for full license details.

use std::{collections::HashMap, net::IpAddr, str::FromStr};

use anyhow::Context;
use reqwest::Url;

use crate::{
    api::error::{Result, SpringError},
    pipeline::{option::Options, ColumnName},
};

#[derive(Clone, Eq, PartialEq, Debug)]
pub enum HttpMethod {
    Post,
}
impl FromStr for HttpMethod {
    type Err = SpringError;

    fn from_str(s: &str) -> Result<Self> {
        match s {
            "POST" => Ok(Self::Post),
            _ => Err(SpringError::InvalidOption {
                key: "HTTP_METHOD".to_string(),
                value: s.to_string(),
                source: anyhow::anyhow!("unsupported HTTP method {}", s),
            }),
        }
    }
}

#[derive(Clone, Eq, PartialEq, Debug)]
pub struct Http1ClientOptions {
    pub remote_host: IpAddr,
    pub remote_port: u16,
    pub method: HttpMethod,
    pub url: Url,
    pub headers: HashMap<String, String>,
    pub blob_body_column: ColumnName,
}

impl TryFrom<&Options> for Http1ClientOptions {
    type Error = SpringError;

    fn try_from(options: &Options) -> Result<Self> {
        let headers = Self::parse_headers(options);

        Ok(Self {
            remote_host: options.get("REMOTE_HOST", |remote_host_str| {
                remote_host_str.parse().context("invalid remote host")
            })?,
            remote_port: options.get("REMOTE_PORT", |remote_port_str| {
                remote_port_str.parse().context("invalid remote port")
            })?,
            method: options.get("METHOD", |method_str| {
                method_str.parse().context("invalid HTTP method")
            })?,
            url: options.get("URL", |path_str| {
                Url::parse(path_str.as_str()).context("failed to parse URL")
            })?,
            headers,
            blob_body_column: options.get("BLOB_BODY_COLUMN", |column_str| {
                Ok(ColumnName::new(column_str.to_string()))
            })?,
        })
    }
}

impl Http1ClientOptions {
    fn parse_headers(options: &Options) -> HashMap<String, String> {
        let mut headers = HashMap::new();

        for (key, value) in options.as_key_values() {
            if key.starts_with("HEADER_") {
                let header_key = key.trim_start_matches("HEADER_").to_string();
                headers.insert(header_key, value.to_string());
            }
        }
        headers
    }
}