2079882825

Build # 2079882825

Build Type

github

Committed by GitHub

Commit Message

<a href="https://github.com/tremor-rs/tremor-runtime/commit/<a class=hub.com/tremor-rs/tremor-runtime/commit/46381822621f529909f503c586dde50bba4759df">463818226<a href="https://github.com/tremor-rs/tremor-runtime/commit/46381822621f529909f503c586dde50bba4759df">">Merge </a><a class="double-link" href="https://github.com/tremor-rs/tremor-runtime/commit/<a class="double-link" href="https://github.com/tremor-rs/tremor-runtime/commit/e2e40600378240a3455a8ef1e067df40505f7802">e2e406003</a>">e2e406003</a><a href="https://github.com/tremor-rs/tremor-runtime/commit/46381822621f529909f503c586dde50bba4759df"> into ca5234b15">ca5234b15</a>

Pull Request Pull Request #1077: Connectors Implementation

Run Details

14158 of 14158 new or added lines in 164 files covered. (100.0%)

23670 of 27365 relevant lines covered (86.5%)

3.75 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

81.25

/src/connectors/utils.rs

// Copyright 2021, The Tremor Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use std::net::SocketAddr;

/// Metrics facilities
pub(crate) mod metrics;

/// Quiescence support facilities
pub(crate) mod quiescence;

/// Reconnection facilities
pub(crate) mod reconnect;

/// Transport Level Security facilities
pub(crate) mod tls;

/// MIME encoding utilities
pub(crate) mod mime;

/// Protocol Buffer utilities
pub(crate) mod pb;

#[derive(Clone, Debug, Hash, PartialEq, Eq)]
pub(crate) struct ConnectionMeta {
    pub(crate) host: String,
    pub(crate) port: u16,
}

impl From<SocketAddr> for ConnectionMeta {
    fn from(sa: SocketAddr) -> Self {
        Self {
            host: sa.ip().to_string(),
            port: sa.port(),
        }
    }
}

pub(crate) mod url {

    use crate::errors::Result;
    use regex::Regex;
    use serde::{Deserialize, Serialize};
    use std::marker::PhantomData;

    lazy_static! {
        // ALLOW: we know this regex is valid
        static ref URL_SCHEME_REGEX: Regex = Regex::new("^[A-Za-z-]+://").expect("Invalid Regex");
    }

    pub(crate) trait Defaults {
        /// Default scheme
        const SCHEME: &'static str;
        /// Default host
        const HOST: &'static str;
        /// Default port
        const PORT: u16;
    }

    // Default HTTP
    pub(crate) struct HttpDefaults;
    impl Defaults for HttpDefaults {
        const HOST: &'static str = "localhost";
        const SCHEME: &'static str = "http";
        const PORT: u16 = 80;
    }
    pub(crate) struct HttpsDefaults;
    impl Defaults for HttpsDefaults {
        const SCHEME: &'static str = "http";
        const HOST: &'static str = "localhost";
        const PORT: u16 = 443;
    }
    /// Endpoint URL
    #[derive(Serialize)]
    pub(crate) struct Url<D: Defaults = HttpDefaults> {
        url: url::Url,
        #[serde(skip)]
        _marker: PhantomData<D>,
    }

    // We have a custom deserializer since we want to not have it nested as\
    // ```json
    // {"url": "http..."}
    // ```
    impl<'de, Dflt: Defaults> Deserialize<'de> for Url<Dflt> {
        fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
        where
            D: serde::Deserializer<'de>,
        {
            let input = String::deserialize(deserializer)?;
            Url::parse(&input).map_err(serde::de::Error::custom)
        }
    }

    impl<D: Defaults, T> AsRef<T> for Url<D>
    where
        url::Url: AsRef<T>,
    {
        fn as_ref(&self) -> &T {
            self.url.as_ref()
        }
    }

    impl<D: Defaults, T> PartialEq<T> for Url<D>
    where
        url::Url: PartialEq<T>,
    {
        fn eq(&self, other: &T) -> bool {
            self.url == *other
        }
    }

    impl<D: Defaults> std::ops::Deref for Url<D> {
        type Target = url::Url;

        fn deref(&self) -> &Self::Target {
            &self.url
        }
    }

    impl<D: Defaults> Clone for Url<D> {
        fn clone(&self) -> Self {
            Self {
                url: self.url.clone(),
                _marker: PhantomData::default(),
            }
        }
    }

    impl<D: Defaults> std::fmt::Debug for Url<D> {
        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
            f.debug_struct("Url").field("url", &self.url).finish()
        }
    }

    impl<D: Defaults> std::fmt::Display for Url<D> {
        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
            write!(f, "{}", self.url)
        }
    }

    impl<D: Defaults> Default for Url<D> {
        fn default() -> Self {
            Self {
                url: url::Url::parse(&format!("{}://{}:{}", D::SCHEME, D::HOST, D::PORT))
                    // ALLOW: this is a known safe url
                    .expect("DEFAULT URL INVALID"),
                _marker: PhantomData::default(),
            }
        }
    }

    impl<D: Defaults> Url<D> {
        pub(crate) fn parse(input: &str) -> Result<Self> {
            let parsed = if URL_SCHEME_REGEX.is_match(input) {
                url::Url::parse(input)
            } else {
                url::Url::parse(&format!("{}://{}", D::SCHEME, input))
            };
            match parsed {
                Ok(url) => Ok(Self {
                    url,
                    ..Self::default()
                }),
                Err(e) => Err(e.into()),
            }
        }
        pub(crate) fn port_or_dflt(&self) -> u16 {
            self.url.port().unwrap_or(D::PORT)
        }
        pub(crate) fn host_or_local(&self) -> &str {
            self.url.host_str().unwrap_or(D::HOST)
        }

        pub(crate) fn url(&self) -> &url::Url {
            &self.url
        }
    }

    #[cfg(test)]
    mod test {
        use super::*;
        #[test]
        fn default() {
            Url::<HttpDefaults>::default();
            Url::<HttpsDefaults>::default();
        }

        use test_case::test_case;

        #[test_case("127.0.0.1", "http://127.0.0.1/"; "ensure scheme without port")]
        #[test_case("localhost:42", "http://localhost:42/"; "ensure scheme")]
        #[test_case("scheme://host:42/path?query=1&query=2#fragment", "scheme://host:42/path?query=1&query=2#fragment"; "all the url features")]
        fn serialize_deserialize(input: &str, expected: &str) -> Result<()> {
            let mut input = format!("\"{input}\""); // prepare for json compat
            let url: Url = simd_json::from_str(&mut input)?;

            let serialized = url.to_string();
            assert_eq!(expected, &serialized);
            Ok(())
        }
    }
}

1	// Copyright 2021, The Tremor Team
2	//
3	// Licensed under the Apache License, Version 2.0 (the "License");
4	// you may not use this file except in compliance with the License.
5	// You may obtain a copy of the License at
6	//
7	// http://www.apache.org/licenses/LICENSE-2.0
8	//
9	// Unless required by applicable law or agreed to in writing, software
10	// distributed under the License is distributed on an "AS IS" BASIS,
11	// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12	// See the License for the specific language governing permissions and
13	// limitations under the License.
14
15	use std::net::SocketAddr;
16
17	/// Metrics facilities
18	pub(crate) mod metrics;
19
20	/// Quiescence support facilities
21	pub(crate) mod quiescence;
22
23	/// Reconnection facilities
24	pub(crate) mod reconnect;
25
26	/// Transport Level Security facilities
27	pub(crate) mod tls;
28
29	/// MIME encoding utilities
30	pub(crate) mod mime;
31
32	/// Protocol Buffer utilities
33	pub(crate) mod pb;
34
35	#[derive(Clone, Debug, Hash, PartialEq, Eq)]
36	pub(crate) struct ConnectionMeta {
37	pub(crate) host: String,
38	pub(crate) port: u16,
39	}
40
41	impl From<SocketAddr> for ConnectionMeta {
42	fn from(sa: SocketAddr) -> Self {	1✔
43	Self {
44	host: sa.ip().to_string(),	1✔
45	port: sa.port(),	1✔
46	}
47	}
48	}
49
50	pub(crate) mod url {
51
52	use crate::errors::Result;
53	use regex::Regex;
54	use serde::{Deserialize, Serialize};
55	use std::marker::PhantomData;
56
57	lazy_static! {
58	// ALLOW: we know this regex is valid
59	static ref URL_SCHEME_REGEX: Regex = Regex::new("^[A-Za-z-]+://").expect("Invalid Regex");	1✔
60	}
61
62	pub(crate) trait Defaults {
63	/// Default scheme
64	const SCHEME: &'static str;
65	/// Default host
66	const HOST: &'static str;
67	/// Default port
68	const PORT: u16;
69	}
70
71	// Default HTTP
72	pub(crate) struct HttpDefaults;
73	impl Defaults for HttpDefaults {
74	const HOST: &'static str = "localhost";
75	const SCHEME: &'static str = "http";
76	const PORT: u16 = 80;
77	}
78	pub(crate) struct HttpsDefaults;
79	impl Defaults for HttpsDefaults {
80	const SCHEME: &'static str = "http";
81	const HOST: &'static str = "localhost";
82	const PORT: u16 = 443;
83	}
84	/// Endpoint URL
85	#[derive(Serialize)]
86	pub(crate) struct Url<D: Defaults = HttpDefaults> {
87	url: url::Url,
88	#[serde(skip)]
89	_marker: PhantomData<D>,
90	}
91
92	// We have a custom deserializer since we want to not have it nested as\
93	// ```json
94	// {"url": "http..."}
95	// ```
96	impl<'de, Dflt: Defaults> Deserialize<'de> for Url<Dflt> {
97	fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>	6✔
98	where
99	D: serde::Deserializer<'de>,
100	{
101	let input = String::deserialize(deserializer)?;	6✔
102	Url::parse(&input).map_err(serde::de::Error::custom)	18✔
103	}
104	}
105
106	impl<D: Defaults, T> AsRef<T> for Url<D>
107	where
108	url::Url: AsRef<T>,
109	{
110	fn as_ref(&self) -> &T {	×
111	self.url.as_ref()	×
112	}
113	}
114
115	impl<D: Defaults, T> PartialEq<T> for Url<D>
116	where
117	url::Url: PartialEq<T>,
118	{
119	fn eq(&self, other: &T) -> bool {	×
120	self.url == *other	×
121	}
122	}
123
124	impl<D: Defaults> std::ops::Deref for Url<D> {
125	type Target = url::Url;
126
127	fn deref(&self) -> &Self::Target {	4✔
128	&self.url	×
129	}
130	}
131
132	impl<D: Defaults> Clone for Url<D> {
133	fn clone(&self) -> Self {	3✔
134	Self {
135	url: self.url.clone(),	3✔
136	_marker: PhantomData::default(),	3✔
137	}
138	}
139	}
140
141	impl<D: Defaults> std::fmt::Debug for Url<D> {
142	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {	×
143	f.debug_struct("Url").field("url", &self.url).finish()	×
144	}
145	}
146
147	impl<D: Defaults> std::fmt::Display for Url<D> {
148	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {	1✔
149	write!(f, "{}", self.url)	1✔
150	}
151	}
152
153	impl<D: Defaults> Default for Url<D> {
154	fn default() -> Self {	6✔
155	Self {
156	url: url::Url::parse(&format!("{}://{}:{}", D::SCHEME, D::HOST, D::PORT))	12✔
157	// ALLOW: this is a known safe url
158	.expect("DEFAULT URL INVALID"),
159	_marker: PhantomData::default(),	6✔
160	}
161	}
162	}
163
164	impl<D: Defaults> Url<D> {
165	pub(crate) fn parse(input: &str) -> Result<Self> {	5✔
166	let parsed = if URL_SCHEME_REGEX.is_match(input) {	5✔
167	url::Url::parse(input)	4✔
168	} else {
169	url::Url::parse(&format!("{}://{}", D::SCHEME, input))	6✔
170	};
171	match parsed {	5✔
172	Ok(url) => Ok(Self {	10✔
173	url,	5✔
174	..Self::default()	5✔
175	}),
176	Err(e) => Err(e.into()),	×
177	}
178	}
179	pub(crate) fn port_or_dflt(&self) -> u16 {	3✔
180	self.url.port().unwrap_or(D::PORT)	3✔
181	}
182	pub(crate) fn host_or_local(&self) -> &str {	4✔
183	self.url.host_str().unwrap_or(D::HOST)	4✔
184	}
185
186	pub(crate) fn url(&self) -> &url::Url {	2✔
187	&self.url	×
188	}
189	}
190
191	#[cfg(test)]
192	mod test {
193	use super::*;
194	#[test]
195	fn default() {	3✔
196	Url::<HttpDefaults>::default();	1✔
197	Url::<HttpsDefaults>::default();	1✔
198	}
199
200	use test_case::test_case;
201
202	#[test_case("127.0.0.1", "http://127.0.0.1/"; "ensure scheme without port")]	12✔
203	#[test_case("localhost:42", "http://localhost:42/"; "ensure scheme")]
204	#[test_case("scheme://host:42/path?query=1&query=2#fragment", "scheme://host:42/path?query=1&query=2#fragment"; "all the url features")]
205	fn serialize_deserialize(input: &str, expected: &str) -> Result<()> {	1✔
206	let mut input = format!("\"{input}\""); // prepare for json compat	1✔
207	let url: Url = simd_json::from_str(&mut input)?;	2✔
208
209	let serialized = url.to_string();	1✔
210	assert_eq!(expected, &serialized);	2✔
211	Ok(())	1✔
212	}
213	}
214	}

tremor-rs / tremor-runtime / 2079882825

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous