Add proper state handling to 'tremor run'.
So state is not reset to 'null' after every event but lasts the whole 'tremor run' session when running a tremor script.
@@ -40,7 +40,8 @@ struct Ingress {
codec: Box<dyn Codec>,
}
-type IngressHandler<T> = dyn Fn(&mut T, &mut u64, &mut Egress, u64, Value) -> Result<()>;
+type IngressHandler<T> =
+ dyn Fn(&mut T, &mut u64, &mut Egress, &mut Value<'static>, u64, Value) -> Result<()>;
impl Ingress {
fn from_args(matches: &ArgMatches) -> Result<Self> {
@@ -86,6 +87,7 @@ impl Ingress {
mut egress: &mut Egress,
handler: &IngressHandler<T>,
) -> Result<()> {
+ let mut state: Value<'static> = Value::null();
loop {
match self.buffer.read(&mut self.buf) {
Ok(0) => {
@@ -104,7 +106,6 @@ impl Ingress {
Ok(None) => continue,
Err(e) => return Err(e.into()),
};
- let event = event.clone();
if self.is_interactive {
eprintln!(
@@ -114,7 +115,7 @@ impl Ingress {
);
highlight(self.is_pretty, &event)?;
}
- handler(runnable, &mut id, &mut egress, at, event)?;
+ handler(runnable, &mut id, &mut egress, &mut state, at, event)?;
}
}
Err(e) => {
@@ -245,15 +246,14 @@ fn run_tremor_source(matches: &ArgMatches, src: String) -> Result<()> {
&mut script,
id,
&mut egress,
- &move |runnable, _id, egress, at, event| {
+ &move |runnable, _id, egress, state, at, event| {
let mut global_map = Value::object();
- let mut state = Value::null();
let mut event = event.clone_static();
match runnable.run(
&EventContext::new(at, None),
AggrType::Tick,
&mut event,
- &mut state,
+ state,
&mut global_map,
) {
Ok(r) => egress.process(&src, &event, r),
@@ -340,7 +340,7 @@ fn run_trickle_source(matches: &ArgMatches, src: String) -> Result<()> {
&mut pipeline,
id,
&mut egress,
- &move |runnable, id, egress, at, event| {
+ &move |runnable, id, egress, _state, at, event| {
let value = EventPayload::new(vec![], |_| ValueAndMeta::from(event.clone_static()));
let mut continuation = vec![];
