662574276

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/bd124a838459fffa8f24737a5bdb8c693152caf6">bd124a838<a href="https://github.com/tremor-rs/tremor-runtime/commit/bd124a838459fffa8f24737a5bdb8c693152caf6">">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/8b8bd04c670fe582c0c63a55c93db38c50e778b7">8b8bd04c6</a>">8b8bd04c6</a><a href="https://github.com/tremor-rs/tremor-runtime/commit/bd124a838459fffa8f24737a5bdb8c693152caf6"> into c58b345b5">c58b345b5</a>

Pull Request Pull Request #841: [WIP] OpenTelemetry connectors

Run Details

1086 of 1086 new or added lines in 8 files covered. (100.0%)

15560 of 18886 relevant lines covered (82.39%)

3.82 hits per line

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

83.17

/src/onramp.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.
use crate::errors::Result;
use crate::metrics::RampReporter;
use crate::pipeline;
use crate::repository::ServantId;
use crate::source::prelude::*;
use crate::source::{
    blaster, crononome, discord, file, kafka, metronome, otel, postgres, rest, tcp, udp, ws,
};
use crate::url::TremorURL;
use async_std::task::{self, JoinHandle};
use serde_yaml::Value;
use std::fmt;
use tremor_common::ids::OnrampIdGen;
use tremor_pipeline::EventId;

pub(crate) type Sender = async_channel::Sender<ManagerMsg>;

pub(crate) trait Impl {
    fn from_config(id: &TremorURL, config: &Option<Value>) -> Result<Box<dyn Onramp>>;
}

#[derive(Clone, Debug)]
pub enum Msg {
    Connect(Cow<'static, str>, Vec<(TremorURL, pipeline::Addr)>),
    Disconnect {
        id: TremorURL,
        tx: async_channel::Sender<bool>,
    },
    Cb(CBAction, EventId),
    // TODO pick good naming here: LinkedEvent / Response / Result?
    Response(tremor_pipeline::Event),
}

pub type Addr = async_channel::Sender<Msg>;

pub(crate) struct OnrampConfig<'cfg> {
    pub onramp_uid: u64,
    pub codec: &'cfg str,
    pub codec_map: halfbrown::HashMap<String, String>,
    pub processors: Processors<'cfg>,
    pub metrics_reporter: RampReporter,
    pub is_linked: bool,
    pub err_required: bool,
}
#[async_trait::async_trait]
pub(crate) trait Onramp: Send {
    async fn start(&mut self, config: OnrampConfig<'_>) -> Result<Addr>;
    fn default_codec(&self) -> &str;
}

// just a lookup

pub(crate) fn lookup(
    name: &str,
    id: &TremorURL,
    config: &Option<Value>,
) -> Result<Box<dyn Onramp>> {
    match name {
        "blaster" => blaster::Blaster::from_config(id, config),
        "file" => file::File::from_config(id, config),
        "kafka" => kafka::Kafka::from_config(id, config),
        "postgres" => postgres::Postgres::from_config(id, config),
        "metronome" => metronome::Metronome::from_config(id, config),
        "crononome" => crononome::Crononome::from_config(id, config),
        "udp" => udp::Udp::from_config(id, config),
        "tcp" => tcp::Tcp::from_config(id, config),
        "rest" => rest::Rest::from_config(id, config),
        "ws" => ws::Ws::from_config(id, config),
        "discord" => discord::Discord::from_config(id, config),
        "otel" => otel::OpenTelemetry::from_config(id, config),
        _ => Err(format!("[onramp:{}] Onramp type {} not known", id, name).into()),
    }
}

pub(crate) struct Create {
    pub id: ServantId,
    pub stream: Box<dyn Onramp>,
    pub codec: String,
    pub codec_map: halfbrown::HashMap<String, String>,
    pub preprocessors: Vec<String>,
    pub postprocessors: Vec<String>,
    pub metrics_reporter: RampReporter,
    pub is_linked: bool,
    pub err_required: bool,
}

impl fmt::Debug for Create {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        write!(f, "StartOnramp({})", self.id)
    }
}

/// This is control plane
pub(crate) enum ManagerMsg {
    Create(async_channel::Sender<Result<Addr>>, Box<Create>),
    Stop,
}

#[derive(Debug, Default)]
pub(crate) struct Manager {
    qsize: usize,
}

impl Manager {
    pub fn new(qsize: usize) -> Self {
        Self { qsize }
    }
    pub fn start(self) -> (JoinHandle<Result<()>>, Sender) {
        let (tx, rx) = bounded(self.qsize);

        let h = task::spawn::<_, Result<()>>(async move {
            let mut onramp_id_gen = OnrampIdGen::new();
            info!("Onramp manager started");
            loop {
                match rx.recv().await {
                    Ok(ManagerMsg::Stop) => {
                        info!("Stopping onramps...");
                        break;
                    }
                    Ok(ManagerMsg::Create(r, c)) => {
                        let Create {
                            codec,
                            codec_map,
                            mut stream,
                            preprocessors,
                            postprocessors,
                            metrics_reporter,
                            is_linked,
                            id,
                            err_required,
                        } = *c;

                        match stream
                            .start(OnrampConfig {
                                onramp_uid: onramp_id_gen.next_id(),
                                codec: &codec,
                                codec_map,
                                processors: Processors {
                                    pre: &preprocessors,
                                    post: &postprocessors,
                                },
                                metrics_reporter,
                                is_linked,
                                err_required,
                            })
                            .await
                        {
                            Ok(addr) => {
                                info!("Onramp {} started.", id);
                                r.send(Ok(addr)).await?;
                            }
                            Err(e) => error!("Creating an onramp failed: {}", e),
                        }
                    }
                    Err(e) => {
                        info!("Stopping onramps... {}", e);
                        break;
                    }
                }
            }
            info!("Onramp manager stopped.");
            Ok(())
        });

        (h, tx)
    }
}

#[cfg(test)]
mod test {
    use super::*;
    use crate::config::Binding;
    use crate::config::MappingMap;
    use crate::repository::BindingArtefact;
    use crate::system;
    use crate::url::TremorURL;
    use simd_json::json;
    use std::io::Write;
    use std::net::TcpListener;
    use std::net::TcpStream;
    use std::net::UdpSocket;
    use std::ops::Range;

    fn port_is_free(port: u16) -> bool {
        match TcpListener::bind(format!("127.0.0.1:{}", port)) {
            Ok(_x) => true,
            _otherwise => false,
        }
    }

    fn find_free_port(mut range: Range<u16>) -> Option<u16> {
        range.find(|port| port_is_free(*port))
    }

    struct TcpInjector {
        stream: TcpStream,
    }

    impl TcpInjector {
        fn with_port(port: u16) -> Self {
            let hostport = format!("localhost:{}", port);
            let stream = TcpStream::connect(hostport).expect("could not connect");
            stream.set_nodelay(true).expect("could not disable nagle");
            stream
                .set_nonblocking(true)
                .expect("could not set non-blocking");
            TcpInjector { stream }
        }
    }

    struct UdpInjector {
        datagram: UdpSocket,
    }

    impl UdpInjector {
        fn with_port(port: u16) -> Self {
            let ephemeral_port = format!(
                "localhost:{}",
                find_free_port(30000..31000).expect("no free ports in range")
            );
            let hostport = format!("localhost:{}", port);
            let datagram = UdpSocket::bind(ephemeral_port).expect("could not bind");
            datagram.connect(hostport).expect("could not connect");
            datagram
                .set_nonblocking(true)
                .expect("could not set non-blocking");
            UdpInjector { datagram }
        }
    }

    macro_rules! rampercize {
        ($onramp_config:expr, $offramp_config:expr, $test:tt) => {
            let storage_directory = Some("./storage".to_string());
            let (world, _handle) = system::World::start(50, storage_directory).await?;
            let config = serde_yaml::to_value($onramp_config).expect("json to yaml not ok");

            let onramp: crate::config::OnRamp = serde_yaml::from_value(config)?;
            let onramp_url = TremorURL::from_onramp_id("test").expect("bad url");
            world
                .repo
                .publish_onramp(&onramp_url, false, onramp)
                .await?;

            let config2 = serde_yaml::to_value($offramp_config).expect("json to yaml not ok");

            let offramp: crate::config::OffRamp = serde_yaml::from_value(config2)?;
            let offramp_url = TremorURL::from_offramp_id("test").expect("bad url");
            world
                .repo
                .publish_offramp(&offramp_url, false, offramp)
                .await?;

            let id = TremorURL::parse(&format!("/pipeline/{}", "test"))?;

            let module_path = &tremor_script::path::ModulePath { mounts: Vec::new() };
            let aggr_reg = tremor_script::aggr_registry();
            let artefact = tremor_pipeline::query::Query::parse(
                &module_path,
                "select event from in into out;",
                "<test>",
                Vec::new(),
                &*tremor_pipeline::FN_REGISTRY.lock()?,
                &aggr_reg,
            )?;
            world.repo.publish_pipeline(&id, false, artefact).await?;

            let binding: Binding = serde_yaml::from_str(
                r#"
id: test
links:
  '/onramp/test/{instance}/out': [ '/pipeline/test/{instance}/in' ]
  '/pipeline/test/{instance}/out': [ '/offramp/test/{instance}/in' ]
"#,
            )?;

            world
                .repo
                .publish_binding(
                    &TremorURL::parse(&format!("/binding/{}", "test"))?,
                    false,
                    BindingArtefact {
                        binding,
                        mapping: None,
                    },
                )
                .await?;

            let mapping: MappingMap = serde_yaml::from_str(
                r#"
/binding/test/01:
  instance: "01"
"#,
            )?;

            let id = TremorURL::parse(&format!("/binding/{}/01", "test"))?;
            world.link_binding(&id, mapping[&id].clone()).await?;

            std::thread::sleep(std::time::Duration::from_millis(1000));

            $test;

            std::thread::sleep(std::time::Duration::from_millis(1000));

            world.stop().await?;
        };
    }

    // macro_rules! rampercize_with_logs {
    //     ($onramp_config:expr, $offramp_config:expr, $test:tt) => {
    //         env_logger::init();
    //         rampercize!($onramp_config, $offramp_config, $test)
    //     };
    // }

    #[async_std::test]
    async fn tcp_onramp() -> Result<()> {
        let port = find_free_port(9000..9099).expect("no free port");
        rampercize!(
            // onramp config
            json!({
                "id": "test",
                "type": "tcp",
                "codec": "json",
                "preprocessors": [ "lines" ],
                "config": {
                  "host": "127.0.0.1",
                  "port": port,
                  "is_non_blocking": true,
                  "ttl": 32,
                }
            }),
            // offramp config
            json!({
                "id": "test",
                "type": "stdout",
                "codec": "json",
            }),
            {
                for _ in 0..3 {
                    let mut inject = TcpInjector::with_port(port);
                    inject
                        .stream
                        .write_all(r#"{"snot": "badger"}\n"#.as_bytes())
                        .expect("something bad happened in cli injector");
                    inject.stream.flush().expect("");
                    drop(inject);
                }
            }
        );
        Ok(())
    }

    #[async_std::test]
    async fn udp_onramp() -> Result<()> {
        let port = find_free_port(9100..9199).expect("no free port");
        rampercize!(
            // onramp config
            json!({
                "id": "test",
                "type": "udp",
                "codec": "json",
                "preprocessors": [ "lines" ],
                "config": {
                  "host": "127.0.0.1",
                  "port": port,
                  "is_non_blocking": true,
                  "ttl": 32,
                }
            }),
            // offramp config
            json!({
                "id": "test",
                "type": "stdout",
                "codec": "json",
            }),
            {
                for _ in 0..3 {
                    let inject = UdpInjector::with_port(port);
                    inject
                        .datagram
                        .send(r#"{"snot": "badger"}\n"#.as_bytes())
                        .expect("something bad happened in cli injector");
                    drop(inject);
                }
            }
        );
        Ok(())
    }

    #[ignore]
    #[async_std::test]
    async fn rest_onramp() -> Result<()> {
        let port = find_free_port(9200..9299).expect("no free port");
        rampercize!(
            // onramp config
            json!({
                "id": "test",
                "type": "rest",
                "codec": "json",
                "preprocessors": [ "lines" ],
                "config": {
                  "host": "127.0.0.1",
                  "port": port,
                  "is_non_blocking": true,
                  "ttl": 32,
                  "resources": [

                  ],
                }
            }),
            // offramp config
            json!({
                "id": "test",
                "type": "stdout",
                "codec": "json",
            }),
            {
                for _ in 0..3 {
                    let mut inject = TcpInjector::with_port(port);
                    inject
                        .stream
                        .write_all(
                            r#"{"HTTP/1.1\nContent-type: application\nContent-Length: 3\n\n{}\n"#
                                .as_bytes(),
                        )
                        .expect("something bad happened in cli injector");
                    inject.stream.flush().expect("");
                    drop(inject);
                }
            }
        );
        Ok(())
    }
}

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	use crate::errors::Result;
15	use crate::metrics::RampReporter;
16	use crate::pipeline;
17	use crate::repository::ServantId;
18	use crate::source::prelude::*;
19	use crate::source::{
20	blaster, crononome, discord, file, kafka, metronome, otel, postgres, rest, tcp, udp, ws,
21	};
22	use crate::url::TremorURL;
23	use async_std::task::{self, JoinHandle};
24	use serde_yaml::Value;
25	use std::fmt;
26	use tremor_common::ids::OnrampIdGen;
27	use tremor_pipeline::EventId;
28
29	pub(crate) type Sender = async_channel::Sender<ManagerMsg>;
30
31	pub(crate) trait Impl {
32	fn from_config(id: &TremorURL, config: &Option<Value>) -> Result<Box<dyn Onramp>>;
33	}
34
35	#[derive(Clone, Debug)]
36	pub enum Msg {
37	Connect(Cow<'static, str>, Vec<(TremorURL, pipeline::Addr)>),
38	Disconnect {
39	id: TremorURL,
40	tx: async_channel::Sender<bool>,
41	},
42	Cb(CBAction, EventId),
43	// TODO pick good naming here: LinkedEvent / Response / Result?
44	Response(tremor_pipeline::Event),
45	}
46
47	pub type Addr = async_channel::Sender<Msg>;
48
49	pub(crate) struct OnrampConfig<'cfg> {
50	pub onramp_uid: u64,
51	pub codec: &'cfg str,
52	pub codec_map: halfbrown::HashMap<String, String>,
53	pub processors: Processors<'cfg>,
54	pub metrics_reporter: RampReporter,
55	pub is_linked: bool,
56	pub err_required: bool,
57	}
58	#[async_trait::async_trait]
59	pub(crate) trait Onramp: Send {
60	async fn start(&mut self, config: OnrampConfig<'_>) -> Result<Addr>;
61	fn default_codec(&self) -> &str;
62	}
63
64	// just a lookup
65
66	pub(crate) fn lookup(	1✔
67	name: &str,
68	id: &TremorURL,
69	config: &Option<Value>,
70	) -> Result<Box<dyn Onramp>> {
71	match name {	3✔
72	"blaster" => blaster::Blaster::from_config(id, config),	2✔
73	"file" => file::File::from_config(id, config),	2✔
74	"kafka" => kafka::Kafka::from_config(id, config),	2✔
75	"postgres" => postgres::Postgres::from_config(id, config),	2✔
76	"metronome" => metronome::Metronome::from_config(id, config),	2✔
77	"crononome" => crononome::Crononome::from_config(id, config),	2✔
78	"udp" => udp::Udp::from_config(id, config),	3✔
79	"tcp" => tcp::Tcp::from_config(id, config),	2✔
80	"rest" => rest::Rest::from_config(id, config),	×
81	"ws" => ws::Ws::from_config(id, config),	×
82	"discord" => discord::Discord::from_config(id, config),	×
83	"otel" => otel::OpenTelemetry::from_config(id, config),	×
84	_ => Err(format!("[onramp:{}] Onramp type {} not known", id, name).into()),	×
85	}
86	}
87
88	pub(crate) struct Create {
89	pub id: ServantId,
90	pub stream: Box<dyn Onramp>,
91	pub codec: String,
92	pub codec_map: halfbrown::HashMap<String, String>,
93	pub preprocessors: Vec<String>,
94	pub postprocessors: Vec<String>,
95	pub metrics_reporter: RampReporter,
96	pub is_linked: bool,
97	pub err_required: bool,
98	}
99
100	impl fmt::Debug for Create {
101	fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {	×
102	write!(f, "StartOnramp({})", self.id)	×
103	}
104	}
105
106	/// This is control plane
107	pub(crate) enum ManagerMsg {
108	Create(async_channel::Sender<Result<Addr>>, Box<Create>),
109	Stop,
110	}
111
112	#[derive(Debug, Default)]
113	pub(crate) struct Manager {
114	qsize: usize,
115	}
116
117	impl Manager {
118	pub fn new(qsize: usize) -> Self {	1✔
119	Self { qsize }
120	}
121	pub fn start(self) -> (JoinHandle<Result<()>>, Sender) {	1✔
122	let (tx, rx) = bounded(self.qsize);	1✔
123
124	let h = task::spawn::<_, Result<()>>(async move {	3✔
125	let mut onramp_id_gen = OnrampIdGen::new();	1✔
126	info!("Onramp manager started");	2✔
127	loop {	2✔
128	match rx.recv().await {	3✔
129	Ok(ManagerMsg::Stop) => {	1✔
130	info!("Stopping onramps...");	2✔
131	break;	1✔
132	}
133	Ok(ManagerMsg::Create(r, c)) => {	1✔
134	let Create {	×
135	codec,	1✔
136	codec_map,	1✔
137	mut stream,	1✔
138	preprocessors,	1✔
139	postprocessors,	1✔
140	metrics_reporter,	1✔
141	is_linked,	1✔
142	id,	1✔
143	err_required,	1✔
144	} = *c;	×
145
146	match stream	4✔
147	.start(OnrampConfig {	1✔
148	onramp_uid: onramp_id_gen.next_id(),	1✔
149	codec: &codec,	1✔
150	codec_map,	1✔
151	processors: Processors {	1✔
152	pre: &preprocessors,	1✔
153	post: &postprocessors,	1✔
154	},
155	metrics_reporter,	1✔
156	is_linked,	×
157	err_required,	×
158	})
159	.await	×
160	{
161	Ok(addr) => {	2✔
162	info!("Onramp {} started.", id);	2✔
163	r.send(Ok(addr)).await?;	2✔
164	}
165	Err(e) => error!("Creating an onramp failed: {}", e),	×
166	}
167	}
168	Err(e) => {	×
169	info!("Stopping onramps... {}", e);	×
170	break;	×
171	}
172	}
173	}
174	info!("Onramp manager stopped.");	2✔
175	Ok(())	1✔
176	});
177
178	(h, tx)	2✔
179	}
180	}
181
182	#[cfg(test)]
183	mod test {
184	use super::*;
185	use crate::config::Binding;
186	use crate::config::MappingMap;
187	use crate::repository::BindingArtefact;
188	use crate::system;
189	use crate::url::TremorURL;
190	use simd_json::json;
191	use std::io::Write;
192	use std::net::TcpListener;
193	use std::net::TcpStream;
194	use std::net::UdpSocket;
195	use std::ops::Range;
196
197	fn port_is_free(port: u16) -> bool {	1✔
198	match TcpListener::bind(format!("127.0.0.1:{}", port)) {	2✔
199	Ok(_x) => true,	2✔
200	_otherwise => false,	×
201	}
202	}
203
204	fn find_free_port(mut range: Range<u16>) -> Option<u16> {	1✔
205	range.find(\|port\| port_is_free(*port))	3✔
206	}
207
208	struct TcpInjector {
209	stream: TcpStream,
210	}
211
212	impl TcpInjector {
213	fn with_port(port: u16) -> Self {	1✔
214	let hostport = format!("localhost:{}", port);	1✔
215	let stream = TcpStream::connect(hostport).expect("could not connect");	1✔
216	stream.set_nodelay(true).expect("could not disable nagle");	1✔
217	stream	1✔
218	.set_nonblocking(true)
219	.expect("could not set non-blocking");
220	TcpInjector { stream }
221	}
222	}
223
224	struct UdpInjector {
225	datagram: UdpSocket,
226	}
227
228	impl UdpInjector {
229	fn with_port(port: u16) -> Self {	1✔
230	let ephemeral_port = format!(	2✔
231	"localhost:{}",
232	find_free_port(30000..31000).expect("no free ports in range")	1✔
233	);
234	let hostport = format!("localhost:{}", port);	1✔
235	let datagram = UdpSocket::bind(ephemeral_port).expect("could not bind");	1✔
236	datagram.connect(hostport).expect("could not connect");	1✔
237	datagram	1✔
238	.set_nonblocking(true)
239	.expect("could not set non-blocking");
240	UdpInjector { datagram }
241	}
242	}
243
244	macro_rules! rampercize {
245	($onramp_config:expr, $offramp_config:expr, $test:tt) => {
246	let storage_directory = Some("./storage".to_string());
247	let (world, _handle) = system::World::start(50, storage_directory).await?;
248	let config = serde_yaml::to_value($onramp_config).expect("json to yaml not ok");
249
250	let onramp: crate::config::OnRamp = serde_yaml::from_value(config)?;
251	let onramp_url = TremorURL::from_onramp_id("test").expect("bad url");
252	world
253	.repo
254	.publish_onramp(&onramp_url, false, onramp)
255	.await?;
256
257	let config2 = serde_yaml::to_value($offramp_config).expect("json to yaml not ok");
258
259	let offramp: crate::config::OffRamp = serde_yaml::from_value(config2)?;
260	let offramp_url = TremorURL::from_offramp_id("test").expect("bad url");
261	world
262	.repo
263	.publish_offramp(&offramp_url, false, offramp)
264	.await?;
265
266	let id = TremorURL::parse(&format!("/pipeline/{}", "test"))?;
267
268	let module_path = &tremor_script::path::ModulePath { mounts: Vec::new() };
269	let aggr_reg = tremor_script::aggr_registry();
270	let artefact = tremor_pipeline::query::Query::parse(
271	&module_path,
272	"select event from in into out;",
273	"<test>",
274	Vec::new(),
275	&*tremor_pipeline::FN_REGISTRY.lock()?,
276	&aggr_reg,
277	)?;
278	world.repo.publish_pipeline(&id, false, artefact).await?;
279
280	let binding: Binding = serde_yaml::from_str(
281	r#"
282	id: test
283	links:
284	'/onramp/test/{instance}/out': [ '/pipeline/test/{instance}/in' ]
285	'/pipeline/test/{instance}/out': [ '/offramp/test/{instance}/in' ]
286	"#,
287	)?;
288
289	world
290	.repo
291	.publish_binding(
292	&TremorURL::parse(&format!("/binding/{}", "test"))?,
293	false,
294	BindingArtefact {
295	binding,
296	mapping: None,
297	},
298	)
299	.await?;
300
301	let mapping: MappingMap = serde_yaml::from_str(
302	r#"
303	/binding/test/01:
304	instance: "01"
305	"#,
306	)?;
307
308	let id = TremorURL::parse(&format!("/binding/{}/01", "test"))?;
309	world.link_binding(&id, mapping[&id].clone()).await?;
310
311	std::thread::sleep(std::time::Duration::from_millis(1000));
312
313	$test;
314
315	std::thread::sleep(std::time::Duration::from_millis(1000));
316
317	world.stop().await?;
318	};
319	}
320
321	// macro_rules! rampercize_with_logs {
322	// ($onramp_config:expr, $offramp_config:expr, $test:tt) => {
323	// env_logger::init();
324	// rampercize!($onramp_config, $offramp_config, $test)
325	// };
326	// }
327
328	#[async_std::test]	7✔
329	async fn tcp_onramp() -> Result<()> {
330	let port = find_free_port(9000..9099).expect("no free port");	1✔
331	rampercize!(	22✔
332	// onramp config
333	json!({	2✔
334	"id": "test",
335	"type": "tcp",
336	"codec": "json",
337	"preprocessors": [ "lines" ],
338	"config": {
339	"host": "127.0.0.1",
340	"port": port,
341	"is_non_blocking": true,
342	"ttl": 32,
343	}
344	}),
345	// offramp config
346	json!({	1✔
347	"id": "test",
348	"type": "stdout",
349	"codec": "json",
350	}),
351	{
352	for _ in 0..3 {	2✔
353	let mut inject = TcpInjector::with_port(port);	1✔
354	inject	1✔
355	.stream
356	.write_all(r#"{"snot": "badger"}\n"#.as_bytes())
357	.expect("something bad happened in cli injector");
358	inject.stream.flush().expect("");	1✔
359	drop(inject);	1✔
360	}
361	}
362	);
363	Ok(())	1✔
364	}
365
366	#[async_std::test]	7✔
367	async fn udp_onramp() -> Result<()> {
368	let port = find_free_port(9100..9199).expect("no free port");	1✔
369	rampercize!(	22✔
370	// onramp config
371	json!({	2✔
372	"id": "test",
373	"type": "udp",
374	"codec": "json",
375	"preprocessors": [ "lines" ],
376	"config": {
377	"host": "127.0.0.1",
378	"port": port,
379	"is_non_blocking": true,
380	"ttl": 32,
381	}
382	}),
383	// offramp config
384	json!({	1✔
385	"id": "test",
386	"type": "stdout",
387	"codec": "json",
388	}),
389	{
390	for _ in 0..3 {	2✔
391	let inject = UdpInjector::with_port(port);	1✔
392	inject	1✔
393	.datagram
394	.send(r#"{"snot": "badger"}\n"#.as_bytes())
395	.expect("something bad happened in cli injector");
396	drop(inject);	1✔
397	}
398	}
399	);
400	Ok(())	1✔
401	}
402
403	#[ignore]
404	#[async_std::test]
405	async fn rest_onramp() -> Result<()> {
406	let port = find_free_port(9200..9299).expect("no free port");
407	rampercize!(
408	// onramp config
409	json!({
410	"id": "test",
411	"type": "rest",
412	"codec": "json",
413	"preprocessors": [ "lines" ],
414	"config": {
415	"host": "127.0.0.1",
416	"port": port,
417	"is_non_blocking": true,
418	"ttl": 32,
419	"resources": [
420
421	],
422	}
423	}),
424	// offramp config
425	json!({
426	"id": "test",
427	"type": "stdout",
428	"codec": "json",
429	}),
430	{
431	for _ in 0..3 {
432	let mut inject = TcpInjector::with_port(port);
433	inject
434	.stream
435	.write_all(
436	r#"{"HTTP/1.1\nContent-type: application\nContent-Length: 3\n\n{}\n"#
437	.as_bytes(),
438	)
439	.expect("something bad happened in cli injector");
440	inject.stream.flush().expect("");
441	drop(inject);
442	}
443	}
444	);
445	Ok(())
446	}
447	}

tremor-rs / tremor-runtime / 662574276

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