2148901975

Build # 2148901975

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/66d46902907f16494659e4c6aa974b39af378c88">66d469029<a href="https://github.com/tremor-rs/tremor-runtime/commit/66d46902907f16494659e4c6aa974b39af378c88">">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/4eaab39b142dae5f2290c7c3052436eefab8e58d">4eaab39b1</a>">4eaab39b1</a><a href="https://github.com/tremor-rs/tremor-runtime/commit/66d46902907f16494659e4c6aa974b39af378c88"> into ca5234b15">ca5234b15</a>

Pull Request Pull Request #1077: Connectors Implementation

Run Details

14883 of 14883 new or added lines in 167 files covered. (100.0%)

24177 of 27694 relevant lines covered (87.3%)

3.8 hits per line

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

88.24

/src/connectors/impls/crononome.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.
mod handler;

use crate::connectors::prelude::*;
use handler::{ChronomicQueue, CronEntryInt};
use serde_yaml::Value as YamlValue;
use tremor_common::time::nanotime;

#[derive(Deserialize, Debug, Clone)]
#[serde(deny_unknown_fields)]
pub(crate) struct Config {
    /// Cron entries
    pub(crate) entries: Option<YamlValue>,
}

impl ConfigImpl for Config {}

#[derive(Debug, Default)]
pub(crate) struct Builder {}

#[async_trait::async_trait]
impl ConnectorBuilder for Builder {
    fn connector_type(&self) -> ConnectorType {
        "crononome".into()
    }

    async fn build(&self, id: &str, config: &ConnectorConfig) -> Result<Box<dyn Connector>> {
        if let Some(raw) = &config.config {
            let raw = Config::new(raw)?;
            let payload = if let Some(yaml_entries) = raw.entries {
                let mut entries = simd_json::to_vec(&yaml_entries)?;
                let tremor_entries = tremor_value::parse_to_value(&mut entries)?;
                Some(tremor_entries.into_static())
            } else {
                None
            };

            let entries = if let Some(Value::Array(entries)) = &payload {
                entries
                    .iter()
                    .cloned()
                    .map(CronEntryInt::try_from)
                    .collect::<Result<Vec<CronEntryInt>>>()?
            } else {
                return Err(ErrorKind::InvalidConfiguration(
                    id.to_string(),
                    "missing `entries` array".to_string(),
                )
                .into());
            };

            Ok(Box::new(Crononome { entries }))
        } else {
            Err(ErrorKind::MissingConfiguration(id.to_string()).into())
        }
    }
}

#[derive(Clone, Debug)]
pub(crate) struct Crononome {
    entries: Vec<CronEntryInt>,
}

#[async_trait::async_trait()]
impl Connector for Crononome {
    async fn create_source(
        &mut self,
        source_context: SourceContext,
        builder: SourceManagerBuilder,
    ) -> Result<Option<SourceAddr>> {
        let source = CrononomeSource::new(self.entries.clone());
        builder.spawn(source, source_context).map(Some)
    }

    fn codec_requirements(&self) -> CodecReq {
        CodecReq::Structured
    }
}
struct CrononomeSource {
    entries: Vec<CronEntryInt>,
    cq: ChronomicQueue,
    origin_uri: EventOriginUri,
}

impl CrononomeSource {
    fn new(entries: Vec<CronEntryInt>) -> Self {
        Self {
            entries,
            cq: ChronomicQueue::default(),
            origin_uri: EventOriginUri {
                scheme: "tremor-crononome".to_string(),
                host: hostname(),
                port: None,
                path: vec![],
            },
        }
    }
}
#[async_trait::async_trait()]
impl Source for CrononomeSource {
    async fn connect(&mut self, _ctx: &SourceContext, _attempt: &Attempt) -> Result<bool> {
        self.cq = ChronomicQueue::default(); // create a new queue to avoid duplication in case of reconnect
        for entry in &self.entries {
            self.cq.enqueue(entry);
        }
        Ok(true)
    }
    async fn pull_data(&mut self, pull_id: &mut u64, ctx: &SourceContext) -> Result<SourceReply> {
        let mut origin_uri = self.origin_uri.clone();
        if let Some(trigger) = self.cq.wait_for_next().await {
            origin_uri.path.push(trigger.0.clone());

            let mut tr: Value<'static> = Value::object_with_capacity(2);
            tr.try_insert("name", trigger.0.clone());
            if let Some(payload) = trigger.1 {
                tr.try_insert("payload", payload);
            }
            let data = literal!({
                "connector": "crononome",
                "ingest_ns": nanotime(),
                "id": *pull_id,
                "trigger": tr
            });
            let meta = ctx.meta(literal!({
                "trigger": trigger.0
            }));
            Ok(SourceReply::Structured {
                origin_uri,
                payload: (data, meta).into(),
                stream: DEFAULT_STREAM_ID,
                port: None,
            })
        } else {
            Ok(SourceReply::Finished)
        }
    }

    fn is_transactional(&self) -> bool {
        false
    }

    fn asynchronous(&self) -> bool {
        false
    }
}

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	mod handler;
15
16	use crate::connectors::prelude::*;
17	use handler::{ChronomicQueue, CronEntryInt};
18	use serde_yaml::Value as YamlValue;
19	use tremor_common::time::nanotime;
20
21	#[derive(Deserialize, Debug, Clone)]
22	#[serde(deny_unknown_fields)]
23	pub(crate) struct Config {
24	/// Cron entries
25	pub(crate) entries: Option<YamlValue>,
26	}
27
28	impl ConfigImpl for Config {}
29
30	#[derive(Debug, Default)]
31	pub(crate) struct Builder {}
32
33	#[async_trait::async_trait]
34	impl ConnectorBuilder for Builder {
35	fn connector_type(&self) -> ConnectorType {	3✔
36	"crononome".into()	3✔
37	}
38
39	async fn build(&self, id: &str, config: &ConnectorConfig) -> Result<Box<dyn Connector>> {	7✔
40	if let Some(raw) = &config.config {	3✔
41	let raw = Config::new(raw)?;	2✔
42	let payload = if let Some(yaml_entries) = raw.entries {	2✔
43	let mut entries = simd_json::to_vec(&yaml_entries)?;	3✔
44	let tremor_entries = tremor_value::parse_to_value(&mut entries)?;	3✔
45	Some(tremor_entries.into_static())	2✔
46	} else {
47	None	×
48	};
49
50	let entries = if let Some(Value::Array(entries)) = &payload {	3✔
51	entries	2✔
52	.iter()
53	.cloned()
54	.map(CronEntryInt::try_from)
55	.collect::<Result<Vec<CronEntryInt>>>()?
56	} else {
57	return Err(ErrorKind::InvalidConfiguration(	×
58	id.to_string(),	×
59	"missing `entries` array".to_string(),	×
60	)
61	.into());
62	};
63
64	Ok(Box::new(Crononome { entries }))	2✔
65	} else {
66	Err(ErrorKind::MissingConfiguration(id.to_string()).into())	×
67	}
68	}
69	}
70
71	#[derive(Clone, Debug)]
72	pub(crate) struct Crononome {
73	entries: Vec<CronEntryInt>,
74	}
75
76	#[async_trait::async_trait()]
77	impl Connector for Crononome {
78	async fn create_source(	1✔
79	&mut self,
80	source_context: SourceContext,
81	builder: SourceManagerBuilder,
82	) -> Result<Option<SourceAddr>> {
83	let source = CrononomeSource::new(self.entries.clone());	2✔
84	builder.spawn(source, source_context).map(Some)	1✔
85	}
86
87	fn codec_requirements(&self) -> CodecReq {	1✔
88	CodecReq::Structured	1✔
89	}
90	}
91	struct CrononomeSource {
92	entries: Vec<CronEntryInt>,
93	cq: ChronomicQueue,
94	origin_uri: EventOriginUri,
95	}
96
97	impl CrononomeSource {
98	fn new(entries: Vec<CronEntryInt>) -> Self {	1✔
99	Self {
100	entries,
101	cq: ChronomicQueue::default(),	1✔
102	origin_uri: EventOriginUri {	1✔
103	scheme: "tremor-crononome".to_string(),
104	host: hostname(),
105	port: None,
106	path: vec![],
107	},
108	}
109	}
110	}
111	#[async_trait::async_trait()]
112	impl Source for CrononomeSource {
113	async fn connect(&mut self, _ctx: &SourceContext, _attempt: &Attempt) -> Result<bool> {	7✔
114	self.cq = ChronomicQueue::default(); // create a new queue to avoid duplication in case of reconnect	2✔
115	for entry in &self.entries {	4✔
116	self.cq.enqueue(entry);	1✔
117	}
118	Ok(true)	1✔
119	}
120	async fn pull_data(&mut self, pull_id: &mut u64, ctx: &SourceContext) -> Result<SourceReply> {	7✔
121	let mut origin_uri = self.origin_uri.clone();	2✔
122	if let Some(trigger) = self.cq.wait_for_next().await {	8✔
123	origin_uri.path.push(trigger.0.clone());	2✔
124
125	let mut tr: Value<'static> = Value::object_with_capacity(2);	1✔
126	tr.try_insert("name", trigger.0.clone());	3✔
127	if let Some(payload) = trigger.1 {	2✔
128	tr.try_insert("payload", payload);	2✔
129	}
130	let data = literal!({	9✔
131	"connector": "crononome",
132	"ingest_ns": nanotime(),	1✔
133	"id": *pull_id,	1✔
134	"trigger": tr	1✔
135	});
136	let meta = ctx.meta(literal!({	4✔
137	"trigger": trigger.0	1✔
138	}));
139	Ok(SourceReply::Structured {	1✔
140	origin_uri,	1✔
141	payload: (data, meta).into(),	1✔
142	stream: DEFAULT_STREAM_ID,
143	port: None,	1✔
144	})
145	} else {
146	Ok(SourceReply::Finished)	×
147	}
148	}
149
150	fn is_transactional(&self) -> bool {	1✔
151	false
152	}
153
154	fn asynchronous(&self) -> bool {	1✔
155	false
156	}
157	}

tremor-rs / tremor-runtime / 2148901975

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