2073282375

Build # 2073282375

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/454d7a855f0eac3a058d85aea4b64e36d38aabd5">454d7a855<a href="https://github.com/tremor-rs/tremor-runtime/commit/454d7a855f0eac3a058d85aea4b64e36d38aabd5">">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/ee3521267f174b79a36ee71a1012d6f5de4ea805">ee3521267</a>">ee3521267</a><a href="https://github.com/tremor-rs/tremor-runtime/commit/454d7a855f0eac3a058d85aea4b64e36d38aabd5"> into ca5234b15">ca5234b15</a>

Pull Request Pull Request #1077: Connectors Implementation

Run Details

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

23849 of 27546 relevant lines covered (86.58%)

4.33 hits per line

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

85.19

/src/connectors/impls/exit.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.

//! # Exit Connector
//!
//! Will stop the Tremor process gracefully when it receives an Event
//!
//! ## Expected Event format
//!
//! This Connector doesnt have any requirements on the event format.
//! But options can be provided within a record with the following fields:
//!
//! * delay: milliseconds to wait before stopping the process
use crate::connectors::{prelude::*, spawn_task};
use crate::system::{ShutdownMode, World};
use async_std::task;
use std::time::Duration;

use value_trait::ValueAccess;

#[derive(Clone)]
pub struct Exit {
    world: World,
    done: bool,
}

#[async_trait::async_trait]
impl Connector for Exit {
    async fn create_sink(
        &mut self,
        sink_context: SinkContext,
        builder: SinkManagerBuilder,
    ) -> Result<Option<SinkAddr>> {
        let sink = self.clone();

        Ok(Some(builder.spawn(sink, sink_context)?))
    }

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

impl Exit {
    const DELAY: &'static str = "delay";
    const GRACEFUL: &'static str = "graceful";
}

#[async_trait::async_trait()]
impl Sink for Exit {
    fn auto_ack(&self) -> bool {
        true
    }
    async fn on_event(
        &mut self,
        _input: &str,
        event: tremor_pipeline::Event,
        ctx: &SinkContext,
        _serializer: &mut EventSerializer,
        _start: u64,
    ) -> Result<SinkReply> {
        if self.done {
            debug!("{} Already exited.", ctx);
        } else if let Some((value, _meta)) = event.value_meta_iter().next() {
            if let Some(delay) = value.get_u64(Self::DELAY) {
                info!(
                    "{} Sleeping for {}ns before triggering shutdown.",
                    ctx, delay
                );
                task::sleep(Duration::from_nanos(delay)).await;
            }
            let mode = if value.get_bool(Self::GRACEFUL).unwrap_or(true) {
                ShutdownMode::Graceful
            } else {
                ShutdownMode::Forceful
            };
            // this should stop the whole server process
            let world = self.world.clone();
            // we spawn this out into another task, so we don't block the sink loop handling control plane messages
            spawn_task(ctx.clone(), async move { world.stop(mode).await });
            self.done = true;
        }

        Ok(SinkReply::default())
    }
}

#[derive(Debug)]
pub(crate) struct Builder {
    world: World,
}
impl Builder {
    pub(crate) fn new(world: &World) -> Self {
        Self {
            world: world.clone(),
        }
    }
}

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

    async fn from_config(
        &self,
        _id: &str,
        _config: &ConnectorConfig,
    ) -> Result<Box<dyn Connector>> {
        Ok(Box::new(Exit {
            world: self.world.clone(),
            done: 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
15	//! # Exit Connector
16	//!
17	//! Will stop the Tremor process gracefully when it receives an Event
18	//!
19	//! ## Expected Event format
20	//!
21	//! This Connector doesnt have any requirements on the event format.
22	//! But options can be provided within a record with the following fields:
23	//!
24	//! * delay: milliseconds to wait before stopping the process
25	use crate::connectors::{prelude::*, spawn_task};
26	use crate::system::{ShutdownMode, World};
27	use async_std::task;
28	use std::time::Duration;
29
30	use value_trait::ValueAccess;
31
32	#[derive(Clone)]
33	pub struct Exit {
34	world: World,
35	done: bool,
36	}
37
38	#[async_trait::async_trait]
39	impl Connector for Exit {
40	async fn create_sink(	1✔
41	&mut self,
42	sink_context: SinkContext,
43	builder: SinkManagerBuilder,
44	) -> Result<Option<SinkAddr>> {
45	let sink = self.clone();	2✔
46
47	Ok(Some(builder.spawn(sink, sink_context)?))	2✔
48	}
49
50	fn codec_requirements(&self) -> CodecReq {	1✔
51	CodecReq::Structured	1✔
52	}
53	}
54
55	impl Exit {
56	const DELAY: &'static str = "delay";
57	const GRACEFUL: &'static str = "graceful";
58	}
59
60	#[async_trait::async_trait()]
61	impl Sink for Exit {
62	fn auto_ack(&self) -> bool {	×
63	true
64	}
65	async fn on_event(	1✔
66	&mut self,
67	_input: &str,
68	event: tremor_pipeline::Event,
69	ctx: &SinkContext,
70	_serializer: &mut EventSerializer,
71	_start: u64,
72	) -> Result<SinkReply> {
73	if self.done {	2✔
74	debug!("{} Already exited.", ctx);	2✔
75	} else if let Some((value, _meta)) = event.value_meta_iter().next() {	3✔
76	if let Some(delay) = value.get_u64(Self::DELAY) {	3✔
77	info!(	×
78	"{} Sleeping for {}ns before triggering shutdown.",
79	ctx, delay
80	);
81	task::sleep(Duration::from_nanos(delay)).await;	×
82	}
83	let mode = if value.get_bool(Self::GRACEFUL).unwrap_or(true) {	2✔
84	ShutdownMode::Graceful	1✔
85	} else {
86	ShutdownMode::Forceful	×
87	};
88	// this should stop the whole server process
89	let world = self.world.clone();	1✔
90	// we spawn this out into another task, so we don't block the sink loop handling control plane messages
91	spawn_task(ctx.clone(), async move { world.stop(mode).await });	5✔
92	self.done = true;	1✔
93	}
94
95	Ok(SinkReply::default())	2✔
96	}
97	}
98
99	#[derive(Debug)]
100	pub(crate) struct Builder {
101	world: World,
102	}
103	impl Builder {
104	pub(crate) fn new(world: &World) -> Self {	4✔
105	Self {
106	world: world.clone(),	4✔
107	}
108	}
109	}
110
111	#[async_trait::async_trait]
112	impl ConnectorBuilder for Builder {
113	fn connector_type(&self) -> ConnectorType {	4✔
114	"exit".into()	4✔
115	}
116
117	async fn from_config(	1✔
118	&self,
119	_id: &str,
120	_config: &ConnectorConfig,
121	) -> Result<Box<dyn Connector>> {
122	Ok(Box::new(Exit {	2✔
123	world: self.world.clone(),	1✔
124	done: false,
125	}))
126	}
127	}

tremor-rs / tremor-runtime / 2073282375

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