2089054829

Build # 2089054829

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/b9213f9bc8bda8e28d0cded3465f14941f8b1834">b9213f9bc<a href="https://github.com/tremor-rs/tremor-runtime/commit/b9213f9bc8bda8e28d0cded3465f14941f8b1834">">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/5f7957a9c1cc665c1ce506d29d73474b77c2c29d">5f7957a9c</a>">5f7957a9c</a><a href="https://github.com/tremor-rs/tremor-runtime/commit/b9213f9bc8bda8e28d0cded3465f14941f8b1834"> into ca5234b15">ca5234b15</a>

Pull Request Pull Request #1077: Connectors Implementation

Run Details

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

23748 of 27430 relevant lines covered (86.58%)

3.75 hits per line

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

90.0

/tremor-pipeline/src/op/trickle/script.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::op::prelude::*;
use std::mem;
use tremor_script::{highlighter, prelude::*};

#[derive(Debug)]
pub struct Script {
    pub id: String,
    pub script: tremor_script::Script,
}

impl Operator for Script {
    fn on_event(
        &mut self,
        _uid: u64,
        _port: &str,
        state: &mut Value<'static>,
        mut event: Event,
    ) -> Result<EventAndInsights> {
        let context = EventContext::new(event.ingest_ns, event.origin_uri.as_ref());

        let port = event.data.rent_mut(|data| {
            let (unwind_event, event_meta): (&mut Value, &mut Value) = data.parts_mut();

            let value = self.script.run(
                &context,
                AggrType::Emit,
                unwind_event, // event
                state,        // state
                event_meta,   // $
            );

            match value {
                Ok(Return::EmitEvent { port }) => Some(port.map_or(OUT, Cow::from)),
                Ok(Return::Emit { value, port }) => {
                    *unwind_event = value;
                    Some(port.map_or(OUT, Cow::from))
                }
                Ok(Return::Drop) => None,
                Err(e) => {
                    let s = highlighter::Dumb::error_to_string(&e).unwrap_or_default();
                    let mut o = literal!({ "error": s });
                    mem::swap(&mut o, unwind_event);
                    if let Some(error) = unwind_event.as_object_mut() {
                        error.insert("event".into(), o);
                    };
                    Some(ERR)
                }
            }
        });

        Ok(if let Some(port) = port {
            EventAndInsights {
                events: vec![(port, event)],
                ..EventAndInsights::default()
            }
        } else if event.transactional {
            EventAndInsights {
                insights: vec![event.insight_ack()],
                ..EventAndInsights::default()
            }
        } else {
            EventAndInsights::default()
        })
    }
}

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	use crate::op::prelude::*;
16	use std::mem;
17	use tremor_script::{highlighter, prelude::*};
18
19	#[derive(Debug)]
20	pub struct Script {
21	pub id: String,
22	pub script: tremor_script::Script,
23	}
24
25	impl Operator for Script {
26	fn on_event(	2✔
27	&mut self,
28	_uid: u64,
29	_port: &str,
30	state: &mut Value<'static>,
31	mut event: Event,
32	) -> Result<EventAndInsights> {
33	let context = EventContext::new(event.ingest_ns, event.origin_uri.as_ref());	2✔
34
35	let port = event.data.rent_mut(\|data\| {	4✔
36	let (unwind_event, event_meta): (&mut Value, &mut Value) = data.parts_mut();	2✔
37
38	let value = self.script.run(	4✔
39	&context,	2✔
40	AggrType::Emit,	2✔
41	unwind_event, // event
42	state, // state	2✔
43	event_meta, // $
44	);
45
46	match value {	2✔
47	Ok(Return::EmitEvent { port }) => Some(port.map_or(OUT, Cow::from)),	1✔
48	Ok(Return::Emit { value, port }) => {	1✔
49	*unwind_event = value;	2✔
50	Some(port.map_or(OUT, Cow::from))	2✔
51	}
52	Ok(Return::Drop) => None,	1✔
53	Err(e) => {	2✔
54	let s = highlighter::Dumb::error_to_string(&e).unwrap_or_default();	4✔
55	let mut o = literal!({ "error": s });	2✔
56	mem::swap(&mut o, unwind_event);	2✔
57	if let Some(error) = unwind_event.as_object_mut() {	4✔
58	error.insert("event".into(), o);	4✔
59	};
60	Some(ERR)	2✔
61	}
62	}
63	});
64
65	Ok(if let Some(port) = port {	6✔
66	EventAndInsights {	2✔
67	events: vec![(port, event)],	4✔
68	..EventAndInsights::default()	2✔
69	}
70	} else if event.transactional {	5✔
71	EventAndInsights {	×
72	insights: vec![event.insight_ack()],	×
73	..EventAndInsights::default()	×
74	}
75	} else {
76	EventAndInsights::default()	2✔
77	})
78	}
79	}

tremor-rs / tremor-runtime / 2089054829

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