navactor 0.3.0

A cli tool for creating and updating actors from piped input
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73

//! This module provides the implementation for the `AccumGene` struct, which represents a gene that
//! primarily uses the `Accumulator` operator to update the state of actors based on incoming data
//! from `IoT` devices. The module implements the `Gene` trait to define how to apply the gene's
//! operators to update an actor's state and to provide the gene's time scope. The module also
//! defines a custom error type (`OpError`) that is returned by the `Accumulator` operator when an
//! input is not valid, and uses the `time` crate to work with dates and times. Additionally, the
//! `AccumGene` struct includes a default implementation that sets the gene's time scope to
//! `Forever` and its base time to the current UTC time.
//!
//! The module exports the `AccumGene` struct, which can be used in the larger system for processing
//! incoming data from `IoT` devices.
use crate::actor::State;
use crate::gene::Gene;
use crate::gene::TimeScope;
use crate::message::Message;
use crate::operator::{Accumulator, OpError, Operator, OperatorResult};
use std::ops::Add;
use time::OffsetDateTime;

pub struct AccumGene {
    pub time_scope: TimeScope,
    pub base_time: OffsetDateTime,
}

fn update_state_with_val<T: Add<Output = T> + Copy>(
    in_val: T,
    idx: i32,
    mut state: State<T>,
    datetime: OffsetDateTime,
) -> OperatorResult<State<T>> {
    let new_val = Accumulator::apply(&state, idx, in_val, datetime)?;
    state.insert(idx, new_val);
    Ok(state)
}

impl<T: Add<Output = T> + Copy> Gene<T> for AccumGene {
    fn apply_operators(&self, mut state: State<T>, update: Message<T>) -> OperatorResult<State<T>> {
        match update {
            Message::Update {
                path: _,
                datetime,
                values,
            } => {
                for &idx in values.keys() {
                    let in_val = *values.get(&idx).ok_or_else(|| OpError {
                        reason: format!("unsupported idx: {idx}"),
                    })?;
                    let len = state.keys().len();
                    log::trace!("updating key {idx} of keys {len}");
                    state = update_state_with_val(in_val, idx, state, datetime)?;
                }
            }
            _ => {
                return Err(OpError {
                    reason: "unsupported message type".to_string(),
                })
            }
        };
        Ok(state)
    }
    fn get_time_scope(&self) -> &TimeScope {
        &self.time_scope
    }
}

impl Default for AccumGene {
    fn default() -> Self {
        Self {
            time_scope: TimeScope::Forever,
            base_time: OffsetDateTime::now_utc(),
        }
    }
}