2095718284

Build # 2095718284

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/5cab8c66c3a8e6fc67cf7366083e8d629a4f50b7">5cab8c66c<a href="https://github.com/tremor-rs/tremor-runtime/commit/5cab8c66c3a8e6fc67cf7366083e8d629a4f50b7">">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/999f4770d32db264fc502fd95596e6353d0b5e9a">999f4770d</a>">999f4770d</a><a href="https://github.com/tremor-rs/tremor-runtime/commit/5cab8c66c3a8e6fc67cf7366083e8d629a4f50b7"> into ca5234b15">ca5234b15</a>

Pull Request Pull Request #1077: Connectors Implementation

Run Details

14720 of 14720 new or added lines in 166 files covered. (100.0%)

24355 of 27854 relevant lines covered (87.44%)

3.76 hits per line

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

35.71

/tremor-script/src/ast/deploy.rs

// Copyright 2020-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.

// We want to keep the names here
#![allow(clippy::module_name_repetitions)]

use super::{
    docs::Docs, helper::Scope, node_id::BaseRef, raw::BaseExpr, CreationalWith, DefinitionalArgs,
    DefinitionalArgsWith, NodeMeta,
};
use super::{node_id::NodeId, PipelineDefinition};
use super::{HashMap, Value};
use crate::{impl_expr, impl_expr_no_lt};
pub(crate) mod raw;

/// A Tremor deployment
#[derive(Clone, Debug, PartialEq, Serialize)]
pub struct Deploy<'script> {
    // TODO handle configuration directives for troy definitions
    /// Configuration directives
    pub config: HashMap<String, Value<'script>>,
    /// Statements
    pub stmts: DeployStmts<'script>,
    /// Scope
    pub scope: Scope<'script>,
    #[serde(skip)]
    /// Documentation comments
    pub docs: Docs,
}

impl<'script> Deploy<'script> {
    /// Provides a `GraphViz` dot file representation of the deployment graph
    #[must_use]
    #[allow(clippy::unused_self)]
    pub fn dot(&self) -> String {
        "todo".to_string() // TODO convert to graphviz dot file
    }
}

/// A tremor deployment language ( troy ) statement
#[derive(Clone, Debug, PartialEq, Serialize)]
pub enum DeployStmt<'script> {
    /// A flow definition
    FlowDefinition(Box<FlowDefinition<'script>>),
    /// A pipeline definition
    PipelineDefinition(Box<PipelineDefinition<'script>>),
    /// A connector definition
    ConnectorDefinition(Box<ConnectorDefinition<'script>>),
    /// The create instance constructor
    DeployFlowStmt(Box<DeployFlow<'script>>),
}

impl<'script> BaseRef for DeployStmt<'script> {
    /// Returns the user provided `fqn` of this statement
    #[must_use]
    fn fqn(&self) -> String {
        match self {
            DeployStmt::FlowDefinition(stmt) => stmt.id.clone(),
            DeployStmt::PipelineDefinition(stmt) => stmt.id.clone(),
            DeployStmt::ConnectorDefinition(stmt) => stmt.id.clone(),
            DeployStmt::DeployFlowStmt(stmt) => stmt.fqn(),
        }
    }
}

#[cfg(not(tarpaulin_include))] // this is a simple passthrough
impl<'script> BaseExpr for DeployStmt<'script> {
    fn meta(&self) -> &NodeMeta {
        match self {
            DeployStmt::PipelineDefinition(s) => s.meta(),
            DeployStmt::ConnectorDefinition(s) => s.meta(),
            DeployStmt::FlowDefinition(s) => s.meta(),
            DeployStmt::DeployFlowStmt(s) => s.meta(),
        }
    }
}

/// A connector definition
#[derive(Clone, Debug, PartialEq, Serialize)]
pub struct ConnectorDefinition<'script> {
    pub(crate) mid: Box<NodeMeta>,
    /// Identifer for the connector
    pub id: String,
    /// Resolved argument defaults
    pub params: DefinitionalArgsWith<'script>,
    /// Internal / intrinsic builtin name
    pub builtin_kind: String,
    /// The rendered config of this connector
    pub config: Value<'script>,
    /// Documentation comments
    #[serde(skip)]
    pub docs: Option<String>,
}
impl_expr!(ConnectorDefinition);

type DeployStmts<'script> = Vec<DeployStmt<'script>>;

/// A deployment link
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
pub enum ConnectStmt {
    /// Connector to Pipeline connection
    ConnectorToPipeline {
        /// Metadata ID
        mid: Box<NodeMeta>,
        /// The instance we're connecting to
        from: DeployEndpoint,
        /// The instance being connected
        to: DeployEndpoint,
    },
    /// Pipeline to connector connection
    PipelineToConnector {
        /// Metadata ID
        mid: Box<NodeMeta>,
        /// The instance we're connecting to
        from: DeployEndpoint,
        /// The instance being connected
        to: DeployEndpoint,
    },
    /// Pipeline to Pipeline connection
    PipelineToPipeline {
        /// Metadata ID
        mid: Box<NodeMeta>,
        /// The instance we're connecting to
        from: DeployEndpoint,
        /// The instance being connected
        to: DeployEndpoint,
    },
}

impl ConnectStmt {
    pub(crate) fn from_mut(&mut self) -> &mut DeployEndpoint {
        match self {
            ConnectStmt::ConnectorToPipeline { from, .. }
            | ConnectStmt::PipelineToConnector { from, .. }
            | ConnectStmt::PipelineToPipeline { from, .. } => from,
        }
    }
    pub(crate) fn to_mut(&mut self) -> &mut DeployEndpoint {
        match self {
            ConnectStmt::ConnectorToPipeline { to, .. }
            | ConnectStmt::PipelineToConnector { to, .. }
            | ConnectStmt::PipelineToPipeline { to, .. } => to,
        }
    }
}

/// A deployment endpoint
#[derive(Clone, Debug, PartialEq, Eq, Hash, Serialize)]
pub struct DeployEndpoint {
    alias: String,
    /// Refers to a local artefact being deployed in a troy definition
    port: String,
    #[serde(skip_serializing)]
    mid: Box<NodeMeta>,
}
impl_expr_no_lt!(DeployEndpoint);

impl std::fmt::Display for DeployEndpoint {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}/{}", self.alias, self.port)
    }
}

impl DeployEndpoint {
    /// Creates a new endpoint
    pub fn new<A, P>(alias: &A, port: &P, mid: &NodeMeta) -> Self
    where
        A: ToString,
        P: ToString,
    {
        Self {
            alias: alias.to_string(),
            port: port.to_string(),
            mid: Box::new(mid.clone()),
        }
    }
    /// The artefact
    #[must_use]
    pub fn alias(&self) -> &str {
        &self.alias
    }
    /// The port
    #[must_use]
    pub fn port(&self) -> &str {
        &self.port
    }
}

/// A flow definition
#[derive(Clone, Debug, PartialEq, Serialize)]
pub struct FlowDefinition<'script> {
    pub(crate) mid: Box<NodeMeta>,
    /// Identifer for the flow
    pub id: String,
    /// Resolved argument defaults
    pub params: DefinitionalArgs<'script>,
    /// Links between artefacts in the flow
    pub connections: Vec<ConnectStmt>,
    /// Deployment atoms
    pub creates: Vec<CreateStmt<'script>>,
    /// Documentation comments
    #[serde(skip)]
    pub docs: Option<String>,
}
impl_expr!(FlowDefinition);

#[derive(Clone, Debug, PartialEq, Serialize)]
/// A connect target
pub enum CreateTargetDefinition<'script> {
    /// A connector
    Connector(ConnectorDefinition<'script>),
    /// A Pipeline
    Pipeline(Box<PipelineDefinition<'script>>),
}
/// A create statement
#[derive(Clone, Debug, PartialEq, Serialize)]
pub struct CreateStmt<'script> {
    pub(crate) mid: Box<NodeMeta>,
    /// Target of the artefact definition being deployed
    pub from_target: NodeId,
    /// The name of the created entity (aka local alias)
    pub instance_alias: String,
    /// creational args
    pub with: CreationalWith<'script>,
    /// Atomic unit of deployment
    pub defn: CreateTargetDefinition<'script>,
}
impl_expr!(CreateStmt);
impl crate::ast::node_id::BaseRef for CreateStmt<'_> {
    fn fqn(&self) -> String {
        self.instance_alias.clone()
    }
}

/// A create statement
#[derive(Clone, Debug, PartialEq, Serialize)]
pub struct DeployFlow<'script> {
    pub(crate) mid: Box<NodeMeta>,
    /// Target of the artefact definition being deployed
    pub from_target: NodeId,
    /// Target for creation
    pub instance_alias: String,
    /// Atomic unit of deployment
    pub defn: FlowDefinition<'script>,
    /// Documentation comments
    #[serde(skip)]
    pub docs: Option<String>,
}
impl_expr!(DeployFlow);
impl crate::ast::node_id::BaseRef for DeployFlow<'_> {
    fn fqn(&self) -> String {
        self.instance_alias.clone()
    }
}

1	// Copyright 2020-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	// We want to keep the names here
16	#![allow(clippy::module_name_repetitions)]
17
18	use super::{
19	docs::Docs, helper::Scope, node_id::BaseRef, raw::BaseExpr, CreationalWith, DefinitionalArgs,
20	DefinitionalArgsWith, NodeMeta,
21	};
22	use super::{node_id::NodeId, PipelineDefinition};
23	use super::{HashMap, Value};
24	use crate::{impl_expr, impl_expr_no_lt};
25	pub(crate) mod raw;
26
27	/// A Tremor deployment
28	#[derive(Clone, Debug, PartialEq, Serialize)]
29	pub struct Deploy<'script> {
30	// TODO handle configuration directives for troy definitions
31	/// Configuration directives
32	pub config: HashMap<String, Value<'script>>,
33	/// Statements
34	pub stmts: DeployStmts<'script>,
35	/// Scope
36	pub scope: Scope<'script>,
37	#[serde(skip)]
38	/// Documentation comments
39	pub docs: Docs,
40	}
41
42	impl<'script> Deploy<'script> {
43	/// Provides a `GraphViz` dot file representation of the deployment graph
44	#[must_use]
45	#[allow(clippy::unused_self)]
46	pub fn dot(&self) -> String {	×
47	"todo".to_string() // TODO convert to graphviz dot file	×
48	}
49	}
50
51	/// A tremor deployment language ( troy ) statement
52	#[derive(Clone, Debug, PartialEq, Serialize)]
53	pub enum DeployStmt<'script> {
54	/// A flow definition
55	FlowDefinition(Box<FlowDefinition<'script>>),
56	/// A pipeline definition
57	PipelineDefinition(Box<PipelineDefinition<'script>>),
58	/// A connector definition
59	ConnectorDefinition(Box<ConnectorDefinition<'script>>),
60	/// The create instance constructor
61	DeployFlowStmt(Box<DeployFlow<'script>>),
62	}
63
64	impl<'script> BaseRef for DeployStmt<'script> {
65	/// Returns the user provided `fqn` of this statement
66	#[must_use]
67	fn fqn(&self) -> String {	×
68	match self {	×
69	DeployStmt::FlowDefinition(stmt) => stmt.id.clone(),	×
70	DeployStmt::PipelineDefinition(stmt) => stmt.id.clone(),	×
71	DeployStmt::ConnectorDefinition(stmt) => stmt.id.clone(),	×
72	DeployStmt::DeployFlowStmt(stmt) => stmt.fqn(),	×
73	}
74	}
75	}
76
77	#[cfg(not(tarpaulin_include))] // this is a simple passthrough
78	impl<'script> BaseExpr for DeployStmt<'script> {
79	fn meta(&self) -> &NodeMeta {
80	match self {
81	DeployStmt::PipelineDefinition(s) => s.meta(),
82	DeployStmt::ConnectorDefinition(s) => s.meta(),
83	DeployStmt::FlowDefinition(s) => s.meta(),
84	DeployStmt::DeployFlowStmt(s) => s.meta(),
85	}
86	}
87	}
88
89	/// A connector definition
90	#[derive(Clone, Debug, PartialEq, Serialize)]
91	pub struct ConnectorDefinition<'script> {
92	pub(crate) mid: Box<NodeMeta>,
93	/// Identifer for the connector
94	pub id: String,
95	/// Resolved argument defaults
96	pub params: DefinitionalArgsWith<'script>,
97	/// Internal / intrinsic builtin name
98	pub builtin_kind: String,
99	/// The rendered config of this connector
100	pub config: Value<'script>,
101	/// Documentation comments
102	#[serde(skip)]
103	pub docs: Option<String>,
104	}
105	impl_expr!(ConnectorDefinition);
106
107	type DeployStmts<'script> = Vec<DeployStmt<'script>>;
108
109	/// A deployment link
110	#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
111	pub enum ConnectStmt {
112	/// Connector to Pipeline connection
113	ConnectorToPipeline {
114	/// Metadata ID
115	mid: Box<NodeMeta>,
116	/// The instance we're connecting to
117	from: DeployEndpoint,
118	/// The instance being connected
119	to: DeployEndpoint,
120	},
121	/// Pipeline to connector connection
122	PipelineToConnector {
123	/// Metadata ID
124	mid: Box<NodeMeta>,
125	/// The instance we're connecting to
126	from: DeployEndpoint,
127	/// The instance being connected
128	to: DeployEndpoint,
129	},
130	/// Pipeline to Pipeline connection
131	PipelineToPipeline {
132	/// Metadata ID
133	mid: Box<NodeMeta>,
134	/// The instance we're connecting to
135	from: DeployEndpoint,
136	/// The instance being connected
137	to: DeployEndpoint,
138	},
139	}
140
141	impl ConnectStmt {
142	pub(crate) fn from_mut(&mut self) -> &mut DeployEndpoint {	×
143	match self {	×
144	ConnectStmt::ConnectorToPipeline { from, .. }	×
145	\| ConnectStmt::PipelineToConnector { from, .. }
146	\| ConnectStmt::PipelineToPipeline { from, .. } => from,
147	}
148	}
149	pub(crate) fn to_mut(&mut self) -> &mut DeployEndpoint {	×
150	match self {	×
151	ConnectStmt::ConnectorToPipeline { to, .. }	×
152	\| ConnectStmt::PipelineToConnector { to, .. }
153	\| ConnectStmt::PipelineToPipeline { to, .. } => to,
154	}
155	}
156	}
157
158	/// A deployment endpoint
159	#[derive(Clone, Debug, PartialEq, Eq, Hash, Serialize)]
160	pub struct DeployEndpoint {
161	alias: String,
162	/// Refers to a local artefact being deployed in a troy definition
163	port: String,
164	#[serde(skip_serializing)]
165	mid: Box<NodeMeta>,
166	}
167	impl_expr_no_lt!(DeployEndpoint);
168
169	impl std::fmt::Display for DeployEndpoint {
170	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {	3✔
171	write!(f, "{}/{}", self.alias, self.port)	3✔
172	}
173	}
174
175	impl DeployEndpoint {
176	/// Creates a new endpoint
177	pub fn new<A, P>(alias: &A, port: &P, mid: &NodeMeta) -> Self	4✔
178	where
179	A: ToString,
180	P: ToString,
181	{
182	Self {
183	alias: alias.to_string(),	4✔
184	port: port.to_string(),	4✔
185	mid: Box::new(mid.clone()),	4✔
186	}
187	}
188	/// The artefact
189	#[must_use]
190	pub fn alias(&self) -> &str {	3✔
191	&self.alias	3✔
192	}
193	/// The port
194	#[must_use]
195	pub fn port(&self) -> &str {	3✔
196	&self.port	3✔
197	}
198	}
199
200	/// A flow definition
201	#[derive(Clone, Debug, PartialEq, Serialize)]
202	pub struct FlowDefinition<'script> {
203	pub(crate) mid: Box<NodeMeta>,
204	/// Identifer for the flow
205	pub id: String,
206	/// Resolved argument defaults
207	pub params: DefinitionalArgs<'script>,
208	/// Links between artefacts in the flow
209	pub connections: Vec<ConnectStmt>,
210	/// Deployment atoms
211	pub creates: Vec<CreateStmt<'script>>,
212	/// Documentation comments
213	#[serde(skip)]
214	pub docs: Option<String>,
215	}
216	impl_expr!(FlowDefinition);
217
218	#[derive(Clone, Debug, PartialEq, Serialize)]
219	/// A connect target
220	pub enum CreateTargetDefinition<'script> {
221	/// A connector
222	Connector(ConnectorDefinition<'script>),
223	/// A Pipeline
224	Pipeline(Box<PipelineDefinition<'script>>),
225	}
226	/// A create statement
227	#[derive(Clone, Debug, PartialEq, Serialize)]
228	pub struct CreateStmt<'script> {
229	pub(crate) mid: Box<NodeMeta>,
230	/// Target of the artefact definition being deployed
231	pub from_target: NodeId,
232	/// The name of the created entity (aka local alias)
233	pub instance_alias: String,
234	/// creational args
235	pub with: CreationalWith<'script>,
236	/// Atomic unit of deployment
237	pub defn: CreateTargetDefinition<'script>,
238	}
239	impl_expr!(CreateStmt);
240	impl crate::ast::node_id::BaseRef for CreateStmt<'_> {
241	fn fqn(&self) -> String {	×
242	self.instance_alias.clone()	×
243	}
244	}
245
246	/// A create statement
247	#[derive(Clone, Debug, PartialEq, Serialize)]
248	pub struct DeployFlow<'script> {
249	pub(crate) mid: Box<NodeMeta>,
250	/// Target of the artefact definition being deployed
251	pub from_target: NodeId,
252	/// Target for creation
253	pub instance_alias: String,
254	/// Atomic unit of deployment
255	pub defn: FlowDefinition<'script>,
256	/// Documentation comments
257	#[serde(skip)]
258	pub docs: Option<String>,
259	}
260	impl_expr!(DeployFlow);
261	impl crate::ast::node_id::BaseRef for DeployFlow<'_> {
262	fn fqn(&self) -> String {	×
263	self.instance_alias.clone()	×
264	}
265	}

tremor-rs / tremor-runtime / 2095718284

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