2089250034

Build # 2089250034

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/b21022303f2d44c41f369b07313dd03ece4cb705">b21022303<a href="https://github.com/tremor-rs/tremor-runtime/commit/b21022303f2d44c41f369b07313dd03ece4cb705">">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/15ac8e1f3865d0a85bced70c1c35f64efbdc0a2b">15ac8e1f3</a>">15ac8e1f3</a><a href="https://github.com/tremor-rs/tremor-runtime/commit/b21022303f2d44c41f369b07313dd03ece4cb705"> into ca5234b15">ca5234b15</a>

Pull Request Pull Request #1077: Connectors Implementation

Run Details

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

23738 of 27357 relevant lines covered (86.77%)

3.76 hits per line

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

94.64

/src/connectors/impls/ws/server.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 super::{WsReader, WsWriter};
use crate::connectors::utils::tls::{load_server_config, TLSServerConfig};
use crate::connectors::{prelude::*, utils::ConnectionMeta};
use async_std::task::JoinHandle;
use async_std::{net::TcpListener, prelude::FutureExt};
use async_tls::TlsAcceptor;
use async_tungstenite::accept_async;
use futures::StreamExt;
use rustls::ServerConfig;
use simd_json::ValueAccess;
use std::net::SocketAddr;
use std::sync::Arc;

const URL_SCHEME: &str = "tremor-ws-server";

#[derive(Deserialize, Debug)]
#[serde(deny_unknown_fields)]
pub(crate) struct Config {
    // kept as a str, so it is re-resolved upon each connect
    url: Url<super::WsDefaults>,
    tls: Option<TLSServerConfig>,
}

impl ConfigImpl for Config {}

#[allow(clippy::module_name_repetitions)]
pub(crate) struct WsServer {
    config: Config,
    accept_task: Option<JoinHandle<()>>,
    sink_runtime: Option<ChannelSinkRuntime<ConnectionMeta>>,
    source_runtime: Option<ChannelSourceRuntime>,
    tls_server_config: Option<ServerConfig>,
}

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

#[async_trait::async_trait]
impl ConnectorBuilder for Builder {
    fn connector_type(&self) -> ConnectorType {
        "ws_server".into()
    }
    async fn from_config(
        &self,
        _id: &str,
        raw_config: &ConnectorConfig,
    ) -> crate::errors::Result<Box<dyn Connector>> {
        if let Some(raw_config) = &raw_config.config {
            let config = Config::new(raw_config)?;

            let tls_server_config = if let Some(tls_config) = config.tls.as_ref() {
                Some(load_server_config(tls_config)?)
            } else {
                None
            };

            Ok(Box::new(WsServer {
                config,
                accept_task: None,  // not yet started
                sink_runtime: None, // replaced in create_sink()
                source_runtime: None,
                tls_server_config,
            }))
        } else {
            Err(crate::errors::ErrorKind::MissingConfiguration(String::from("WsServer")).into())
        }
    }
}

fn resolve_connection_meta(meta: &Value) -> Option<ConnectionMeta> {
    let peer = meta.get("peer");
    peer.get_u16("port")
        .zip(peer.get_str("host"))
        .map(|(port, host)| -> ConnectionMeta {
            ConnectionMeta {
                host: host.to_string(),
                port,
            }
        })
}

impl WsServer {
    fn meta(peer: SocketAddr, has_tls: bool) -> Value<'static> {
        let peer_ip = peer.ip().to_string();
        let peer_port = peer.port();

        literal!({
            "tls": has_tls,
            "peer": {
                "host": peer_ip,
                "port": peer_port
            }
        })
    }
}

#[async_trait::async_trait()]
impl Connector for WsServer {
    async fn on_stop(&mut self, _ctx: &ConnectorContext) -> Result<()> {
        if let Some(accept_task) = self.accept_task.take() {
            // stop acceptin' new connections
            accept_task.cancel().await;
        }
        Ok(())
    }

    async fn create_source(
        &mut self,
        ctx: SourceContext,
        builder: SourceManagerBuilder,
    ) -> Result<Option<SourceAddr>> {
        let source = ChannelSource::new(builder.qsize());
        self.source_runtime = Some(source.runtime());
        let addr = builder.spawn(source, ctx)?;

        Ok(Some(addr))
    }

    async fn create_sink(
        &mut self,
        ctx: SinkContext,
        builder: SinkManagerBuilder,
    ) -> Result<Option<SinkAddr>> {
        let sink = ChannelSink::new_with_meta(
            builder.qsize(),
            resolve_connection_meta,
            builder.reply_tx(),
        );

        self.sink_runtime = Some(sink.runtime());
        let addr = builder.spawn(sink, ctx)?;
        Ok(Some(addr))
    }

    #[allow(clippy::too_many_lines)]
    async fn connect(&mut self, ctx: &ConnectorContext, _attempt: &Attempt) -> Result<bool> {
        // TODO: this can be simplified as the connect can be moved into the source
        let path = vec![self.config.url.port_or_dflt().to_string()];

        let source_runtime = self
            .source_runtime
            .clone()
            .ok_or("Source runtime not initialized")?;
        let sink_runtime = self
            .sink_runtime
            .clone()
            .ok_or("sink runtime not initialized")?;

        // cancel last accept task if necessary, this will drop the previous listener
        if let Some(previous_handle) = self.accept_task.take() {
            previous_handle.cancel().await;
        }

        // TODO: allow for other sockets
        let host = self.config.url.host_or_local();
        let port = self
            .config
            .url
            .port()
            .unwrap_or(if self.config.url.scheme() == "wss" {
                443
            } else {
                80
            });
        let listener = TcpListener::bind((host, port)).await?;

        let ctx = ctx.clone();
        let tls_server_config = self.tls_server_config.clone();

        // accept task
        self.accept_task = Some(spawn_task(ctx.clone(), async move {
            let mut stream_id_gen = StreamIdGen::default();
            while ctx.quiescence_beacon.continue_reading().await {
                match listener.accept().timeout(ACCEPT_TIMEOUT).await {
                    Ok(Ok((tcp_stream, peer_addr))) => {
                        let stream_id: u64 = stream_id_gen.next_stream_id();
                        let connection_meta: ConnectionMeta = peer_addr.into();

                        // Async<T> allows us to read in one thread and write in another concurrently - see its documentation
                        // So we don't need no BiLock like we would when using `.split()`
                        let origin_uri = EventOriginUri {
                            scheme: URL_SCHEME.to_string(),
                            host: peer_addr.ip().to_string(),
                            port: Some(peer_addr.port()),
                            path: path.clone(), // captures server port
                        };

                        let tls_acceptor: Option<TlsAcceptor> = tls_server_config
                            .clone()
                            .map(|sc| TlsAcceptor::from(Arc::new(sc)));
                        if let Some(acceptor) = tls_acceptor {
                            let meta = ctx.meta(WsServer::meta(peer_addr, true));
                            // TODO: this should live in its own task, as it requires rome roundtrips :()
                            let tls_stream = acceptor.accept(tcp_stream).await?;
                            let ws_stream = accept_async(tls_stream).await?;
                            debug!("{ctx} new connection from {peer_addr}");

                            let (ws_write, ws_read) = ws_stream.split();

                            let ws_writer = WsWriter::new_tls_server(ws_write);
                            sink_runtime.register_stream_writer(
                                stream_id,
                                Some(connection_meta.clone()),
                                &ctx,
                                ws_writer,
                            );

                            let ws_reader = WsReader::new(
                                ws_read,
                                sink_runtime.clone(),
                                origin_uri.clone(),
                                meta,
                                ctx.clone(),
                            );
                            source_runtime.register_stream_reader(stream_id, &ctx, ws_reader);
                        } else {
                            let ws_stream = match accept_async(tcp_stream).await {
                                Ok(s) => s,
                                Err(e) => {
                                    error!("{ctx} Websocket connection error: {e}");
                                    continue;
                                }
                            };
                            debug!("{ctx} new connection from {peer_addr}",);

                            let (ws_write, ws_read) = ws_stream.split();

                            let meta = ctx.meta(WsServer::meta(peer_addr, false));

                            let ws_writer = WsWriter::new(ws_write);

                            sink_runtime.register_stream_writer(
                                stream_id,
                                Some(connection_meta.clone()),
                                &ctx,
                                ws_writer,
                            );

                            let ws_reader = WsReader::new(
                                ws_read,
                                sink_runtime.clone(),
                                origin_uri.clone(),
                                meta,
                                ctx.clone(),
                            );
                            source_runtime.register_stream_reader(stream_id, &ctx, ws_reader);
                        }
                    }
                    Ok(Err(e)) => return Err(e.into()),
                    Err(_) => continue,
                };
            }
            Ok(())
        }));

        Ok(true)
    }

    fn codec_requirements(&self) -> CodecReq {
        CodecReq::Required
    }
}

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 super::{WsReader, WsWriter};
16	use crate::connectors::utils::tls::{load_server_config, TLSServerConfig};
17	use crate::connectors::{prelude::*, utils::ConnectionMeta};
18	use async_std::task::JoinHandle;
19	use async_std::{net::TcpListener, prelude::FutureExt};
20	use async_tls::TlsAcceptor;
21	use async_tungstenite::accept_async;
22	use futures::StreamExt;
23	use rustls::ServerConfig;
24	use simd_json::ValueAccess;
25	use std::net::SocketAddr;
26	use std::sync::Arc;
27
28	const URL_SCHEME: &str = "tremor-ws-server";
29
30	#[derive(Deserialize, Debug)]
31	#[serde(deny_unknown_fields)]
32	pub(crate) struct Config {
33	// kept as a str, so it is re-resolved upon each connect
34	url: Url<super::WsDefaults>,
35	tls: Option<TLSServerConfig>,
36	}
37
38	impl ConfigImpl for Config {}
39
40	#[allow(clippy::module_name_repetitions)]
41	pub(crate) struct WsServer {
42	config: Config,
43	accept_task: Option<JoinHandle<()>>,
44	sink_runtime: Option<ChannelSinkRuntime<ConnectionMeta>>,
45	source_runtime: Option<ChannelSourceRuntime>,
46	tls_server_config: Option<ServerConfig>,
47	}
48
49	#[derive(Debug, Default)]
50	pub(crate) struct Builder {}
51
52	#[async_trait::async_trait]
53	impl ConnectorBuilder for Builder {
54	fn connector_type(&self) -> ConnectorType {	3✔
55	"ws_server".into()	3✔
56	}
57	async fn from_config(	1✔
58	&self,
59	_id: &str,
60	raw_config: &ConnectorConfig,
61	) -> crate::errors::Result<Box<dyn Connector>> {
62	if let Some(raw_config) = &raw_config.config {	3✔
63	let config = Config::new(raw_config)?;	2✔
64
65	let tls_server_config = if let Some(tls_config) = config.tls.as_ref() {	4✔
66	Some(load_server_config(tls_config)?)	2✔
67	} else {
68	None	1✔
69	};
70
71	Ok(Box::new(WsServer {	2✔
72	config,	1✔
73	accept_task: None, // not yet started	1✔
74	sink_runtime: None, // replaced in create_sink()	1✔
75	source_runtime: None,	1✔
76	tls_server_config,	1✔
77	}))
78	} else {
79	Err(crate::errors::ErrorKind::MissingConfiguration(String::from("WsServer")).into())	×
80	}
81	}
82	}
83
84	fn resolve_connection_meta(meta: &Value) -> Option<ConnectionMeta> {	1✔
85	let peer = meta.get("peer");	1✔
86	peer.get_u16("port")	2✔
87	.zip(peer.get_str("host"))	1✔
88	.map(\|(port, host)\| -> ConnectionMeta {	2✔
89	ConnectionMeta {	1✔
90	host: host.to_string(),	1✔
91	port,
92	}
93	})
94	}
95
96	impl WsServer {
97	fn meta(peer: SocketAddr, has_tls: bool) -> Value<'static> {	1✔
98	let peer_ip = peer.ip().to_string();	1✔
99	let peer_port = peer.port();	1✔
100
101	literal!({	3✔
102	"tls": has_tls,
103	"peer": {
104	"host": peer_ip,	1✔
105	"port": peer_port
106	}
107	})
108	}
109	}
110
111	#[async_trait::async_trait()]
112	impl Connector for WsServer {
113	async fn on_stop(&mut self, _ctx: &ConnectorContext) -> Result<()> {	7✔
114	if let Some(accept_task) = self.accept_task.take() {	4✔
115	// stop acceptin' new connections
116	accept_task.cancel().await;	4✔
117	}
118	Ok(())	1✔
119	}
120
121	async fn create_source(	1✔
122	&mut self,
123	ctx: SourceContext,
124	builder: SourceManagerBuilder,
125	) -> Result<Option<SourceAddr>> {
126	let source = ChannelSource::new(builder.qsize());	3✔
127	self.source_runtime = Some(source.runtime());	2✔
128	let addr = builder.spawn(source, ctx)?;	2✔
129
130	Ok(Some(addr))	1✔
131	}
132
133	async fn create_sink(	1✔
134	&mut self,
135	ctx: SinkContext,
136	builder: SinkManagerBuilder,
137	) -> Result<Option<SinkAddr>> {
138	let sink = ChannelSink::new_with_meta(
139	builder.qsize(),	1✔
140	resolve_connection_meta,
141	builder.reply_tx(),	1✔
142	);
143
144	self.sink_runtime = Some(sink.runtime());	2✔
145	let addr = builder.spawn(sink, ctx)?;	2✔
146	Ok(Some(addr))	1✔
147	}
148
149	#[allow(clippy::too_many_lines)]
150	async fn connect(&mut self, ctx: &ConnectorContext, _attempt: &Attempt) -> Result<bool> {	7✔
151	// TODO: this can be simplified as the connect can be moved into the source
152	let path = vec![self.config.url.port_or_dflt().to_string()];	3✔
153
154	let source_runtime = self	2✔
155	.source_runtime
156	.clone()
157	.ok_or("Source runtime not initialized")?;
158	let sink_runtime = self	3✔
159	.sink_runtime
160	.clone()
161	.ok_or("sink runtime not initialized")?;
162
163	// cancel last accept task if necessary, this will drop the previous listener
164	if let Some(previous_handle) = self.accept_task.take() {	3✔
165	previous_handle.cancel().await;	×
166	}
167
168	// TODO: allow for other sockets
169	let host = self.config.url.host_or_local();	2✔
170	let port = self	2✔
171	.config
172	.url
173	.port()
174	.unwrap_or(if self.config.url.scheme() == "wss" {	2✔
175	443	1✔
176	} else {
177	80	1✔
178	});
179	let listener = TcpListener::bind((host, port)).await?;	3✔
180
181	let ctx = ctx.clone();	2✔
182	let tls_server_config = self.tls_server_config.clone();	2✔
183
184	// accept task
185	self.accept_task = Some(spawn_task(ctx.clone(), async move {	7✔
186	let mut stream_id_gen = StreamIdGen::default();	2✔
187	while ctx.quiescence_beacon.continue_reading().await {	3✔
188	match listener.accept().timeout(ACCEPT_TIMEOUT).await {	5✔
189	Ok(Ok((tcp_stream, peer_addr))) => {	1✔
190	let stream_id: u64 = stream_id_gen.next_stream_id();	2✔
191	let connection_meta: ConnectionMeta = peer_addr.into();	1✔
192
193	// Async<T> allows us to read in one thread and write in another concurrently - see its documentation
194	// So we don't need no BiLock like we would when using `.split()`
195	let origin_uri = EventOriginUri {	1✔
196	scheme: URL_SCHEME.to_string(),	1✔
197	host: peer_addr.ip().to_string(),	2✔
198	port: Some(peer_addr.port()),	1✔
199	path: path.clone(), // captures server port	1✔
200	};
201
202	let tls_acceptor: Option<TlsAcceptor> = tls_server_config	2✔
203	.clone()
204	.map(\|sc\| TlsAcceptor::from(Arc::new(sc)));	3✔
205	if let Some(acceptor) = tls_acceptor {	1✔
206	let meta = ctx.meta(WsServer::meta(peer_addr, true));	3✔
207	// TODO: this should live in its own task, as it requires rome roundtrips :()
208	let tls_stream = acceptor.accept(tcp_stream).await?;	4✔
209	let ws_stream = accept_async(tls_stream).await?;	4✔
210	debug!("{ctx} new connection from {peer_addr}");	3✔
211
212	let (ws_write, ws_read) = ws_stream.split();	1✔
213
214	let ws_writer = WsWriter::new_tls_server(ws_write);	1✔
215	sink_runtime.register_stream_writer(	2✔
216	stream_id,	1✔
217	Some(connection_meta.clone()),	2✔
218	&ctx,
219	ws_writer,	1✔
220	);
221
222	let ws_reader = WsReader::new(	1✔
223	ws_read,	1✔
224	sink_runtime.clone(),	1✔
225	origin_uri.clone(),	2✔
226	meta,	1✔
227	ctx.clone(),	1✔
228	);
229	source_runtime.register_stream_reader(stream_id, &ctx, ws_reader);	1✔
230	} else {
231	let ws_stream = match accept_async(tcp_stream).await {	2✔
232	Ok(s) => s,	1✔
233	Err(e) => {	×
234	error!("{ctx} Websocket connection error: {e}");	×
235	continue;
236	}
237	};
238	debug!("{ctx} new connection from {peer_addr}",);	3✔
239
240	let (ws_write, ws_read) = ws_stream.split();	1✔
241
242	let meta = ctx.meta(WsServer::meta(peer_addr, false));	2✔
243
244	let ws_writer = WsWriter::new(ws_write);	1✔
245
246	sink_runtime.register_stream_writer(	2✔
247	stream_id,	1✔
248	Some(connection_meta.clone()),	2✔
249	&ctx,
250	ws_writer,	1✔
251	);
252
253	let ws_reader = WsReader::new(	1✔
254	ws_read,	1✔
255	sink_runtime.clone(),	1✔
256	origin_uri.clone(),	2✔
257	meta,	1✔
258	ctx.clone(),	1✔
259	);
260	source_runtime.register_stream_reader(stream_id, &ctx, ws_reader);	1✔
261	}
262	}
263	Ok(Err(e)) => return Err(e.into()),	×
264	Err(_) => continue,
265	};
266	}
267	Ok(())	×
268	}));
269
270	Ok(true)	1✔
271	}
272
273	fn codec_requirements(&self) -> CodecReq {	1✔
274	CodecReq::Required	1✔
275	}
276	}

tremor-rs / tremor-runtime / 2089250034

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