Struct rtlola_interpreter::Monitor

pub struct Monitor<Source, SourceTime, Verdict = Incremental, VerdictTime = RelativeFloat>where
    Source: Input,
    SourceTime: TimeRepresentation,
    Verdict: VerdictRepresentation,
    VerdictTime: OutputTimeRepresentation + 'static,{ /* private fields */ }

The Monitor is the central object exposed by the API.

The Monitor accepts new events and computes streams. It can compute event-based streams based on new events through accept_event. It can also simply advance periodic streams up to a given timestamp through accept_time. The generic argument Source implements the Input trait describing the input source of the API. The generic argument SourceTime implements the TimeRepresentation trait defining the input time format. The generic argument Verdict implements the VerdictRepresentation trait describing the output format of the API that is by default Incremental. The generic argument VerdictTime implements the TimeRepresentation trait defining the output time format. It defaults to RelativeFloat

Implementations

impl<Source, SourceTime, Verdict, VerdictTime> Monitor<Source, SourceTime, Verdict, VerdictTime>where
    Source: Input,
    SourceTime: TimeRepresentation,
    Verdict: VerdictRepresentation,
    VerdictTime: OutputTimeRepresentation,

Crate-public interface

pub fn setup(
    config: Config<SourceTime, VerdictTime>,
    setup_data: Source::CreationData
) -> Result<Monitor<Source, SourceTime, Verdict, VerdictTime>, Source::Error>

setup

Examples found in repository

src/configuration/config.rs (line 74)

    pub fn monitor_with_data(
        self,
        data: Source::CreationData,
    ) -> Result<Monitor<Source, SourceTime, Verdict, VerdictTime>, Source::Error> {
        Monitor::setup(self.config, data)
    }

    /// Transforms the configuration into a [Monitor]
    pub fn monitor(self) -> Result<Monitor<Source, SourceTime, Verdict, VerdictTime>, Source::Error>
    where
        Source: Input<CreationData = ()>,
    {
        Monitor::setup(self.config, ())
    }

    #[cfg(feature = "queued-api")]
    /// Transforms the configuration into a [QueuedMonitor] using the provided data to setup the input source.
    pub fn queued_monitor_with_data(
        self,
        data: Source::CreationData,
    ) -> QueuedMonitor<Source, SourceTime, Verdict, VerdictTime> {
        QueuedMonitor::setup(self.config, data)
    }

    #[cfg(feature = "queued-api")]
    /// Transforms the configuration into a [QueuedMonitor]
    pub fn queued_monitor(self) -> QueuedMonitor<Source, SourceTime, Verdict, VerdictTime>
    where
        Source: Input<CreationData = ()>,
    {
        QueuedMonitor::setup(self.config, ())
    }
}

/// The execution mode of the interpreter. See the `README` for more details.
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub enum ExecutionMode {
    /// Time provided by input source
    Offline,
    /// Time taken by evaluator
    Online,
}

impl Config<RelativeFloat, RelativeFloat> {
    /// Creates a new Debug config.
    pub fn debug(ir: RtLolaMir) -> Self {
        Config {
            ir,
            mode: ExecutionMode::Offline,
            input_time_representation: RelativeFloat::default(),
            output_time_representation: PhantomData::<RelativeFloat>::default(),
            start_time: None,
        }
    }
}

impl<InputTime: TimeRepresentation, OutputTime: OutputTimeRepresentation> Config<InputTime, OutputTime> {
    /// Creates a new API config
    pub fn api(ir: RtLolaMir) -> Self {
        Config {
            ir,
            mode: ExecutionMode::Offline,
            input_time_representation: InputTime::default(),
            output_time_representation: PhantomData::default(),
            start_time: None,
        }
    }

    /// Turn the configuration into the [Monitor] API.
    pub fn monitor<Source: Input, Verdict: VerdictRepresentation>(
        self,
        data: Source::CreationData,
    ) -> Result<Monitor<Source, InputTime, Verdict, OutputTime>, Source::Error> {
        Monitor::setup(self, data)
    }

src/api/queued.rs (line 593)

    fn init(&mut self) -> Result<(), QueueError> {
        // Setup evaluator
        let monitor: Monitor<Source, SourceTime, Verdict, VerdictTime> =
            Monitor::setup(self.config.clone(), self.setup_data.clone())
                .map_err(|e| QueueError::SourceError(Box::new(e)))?;
        self.monitor.replace(monitor);
        Ok(())
    }

impl<Source, SourceTime, Verdict, VerdictTime> Monitor<Source, SourceTime, Verdict, VerdictTime>where
    Source: Input,
    SourceTime: TimeRepresentation,
    Verdict: VerdictRepresentation,
    VerdictTime: OutputTimeRepresentation,

Public interface

pub fn accept_event(
    &mut self,
    ev: Source::Record,
    ts: SourceTime::InnerTime
) -> Result<Verdicts<Verdict, VerdictTime>, Source::Error>

Computes all periodic streams up through the new timestamp and then handles the input event.

The new event is therefore not seen by periodic streams up through a new timestamp.

Examples found in repository

src/api/queued.rs (line 609)

    fn process(&mut self) -> Result<(), QueueError> {
        let monitor = self.monitor.as_mut().expect("Init to be called before process");
        let mut last_event = None;
        let mut done = false;
        while !done {
            let timed = match self.input.recv() {
                Ok(WorkItem::Event(e, ts)) => {
                    // Received Event
                    last_event.replace(ts.clone());
                    let Verdicts { timed, event, ts } = monitor
                        .accept_event(e, ts)
                        .map_err(|e| QueueError::SourceError(Box::new(e)))?;

                    if !event.is_empty() {
                        let verdict = QueuedVerdict {
                            kind: VerdictKind::Event,
                            ts: ts.clone(),
                            verdict: event,
                        };
                        Self::try_send(&self.output, Some(verdict))?;
                    }

                    timed
                },
                Err(_) => {
                    // Channel closed, we are done here
                    done = true;
                    if let Some(last_event) = last_event.as_ref() {
                        monitor.accept_time(last_event.clone())
                    } else {
                        return Ok(());
                    }
                },
                Ok(WorkItem::Start) => {
                    // Received second start command -> abort
                    return Err(QueueError::MultipleStart);
                },
            };

            for (ts, v) in timed {
                let verdict = QueuedVerdict {
                    kind: VerdictKind::Timed,
                    ts,
                    verdict: v,
                };
                Self::try_send(&self.output, Some(verdict))?;
            }
        }
        Ok(())
    }

pub fn accept_time(
    &mut self,
    ts: SourceTime::InnerTime
) -> Vec<(VerdictTime::InnerTime, Verdict)>

Computes all periodic streams up through and including the timestamp.

Examples found in repository

src/api/queued.rs (line 627)

    fn process(&mut self) -> Result<(), QueueError> {
        let monitor = self.monitor.as_mut().expect("Init to be called before process");
        let mut last_event = None;
        let mut done = false;
        while !done {
            let timed = match self.input.recv() {
                Ok(WorkItem::Event(e, ts)) => {
                    // Received Event
                    last_event.replace(ts.clone());
                    let Verdicts { timed, event, ts } = monitor
                        .accept_event(e, ts)
                        .map_err(|e| QueueError::SourceError(Box::new(e)))?;

                    if !event.is_empty() {
                        let verdict = QueuedVerdict {
                            kind: VerdictKind::Event,
                            ts: ts.clone(),
                            verdict: event,
                        };
                        Self::try_send(&self.output, Some(verdict))?;
                    }

                    timed
                },
                Err(_) => {
                    // Channel closed, we are done here
                    done = true;
                    if let Some(last_event) = last_event.as_ref() {
                        monitor.accept_time(last_event.clone())
                    } else {
                        return Ok(());
                    }
                },
                Ok(WorkItem::Start) => {
                    // Received second start command -> abort
                    return Err(QueueError::MultipleStart);
                },
            };

            for (ts, v) in timed {
                let verdict = QueuedVerdict {
                    kind: VerdictKind::Timed,
                    ts,
                    verdict: v,
                };
                Self::try_send(&self.output, Some(verdict))?;
            }
        }
        Ok(())
    }

pub fn ir(&self) -> &RtLolaMir

Returns the underlying representation of the specification as an RtLolaMir

pub fn name_for_input(&self, id: InputReference) -> &str

Get the name of an input stream based on its InputReference.

The reference is valid for the lifetime of the monitor.

pub fn name_for_output(&self, id: OutputReference) -> &str

Get the name of an output stream based on its OutputReference.

The reference is valid for the lifetime of the monitor.

pub fn trigger_message(&self, id: usize) -> &str

Get the message of a trigger based on its index.

The reference is valid for the lifetime of the monitor.

pub fn trigger_stream_index(&self, id: usize) -> usize

Get the OutputReference of a trigger based on its index.

pub fn number_of_input_streams(&self) -> usize

Get the number of input streams.

pub fn number_of_output_streams(&self) -> usize

Get the number of output streams (this includes one output stream for each trigger).

pub fn number_of_triggers(&self) -> usize

Get the number of triggers.

pub fn type_of_input(&self, id: InputReference) -> &Type

Get the type of an input stream based on its InputReference.

The reference is valid for the lifetime of the monitor.

pub fn type_of_output(&self, id: OutputReference) -> &Type

Get the type of an output stream based on its OutputReference.

The reference is valid for the lifetime of the monitor.

pub fn extend_rate_of_output(&self, id: OutputReference) -> Option<Duration>

Get the extend rate of an output stream based on its OutputReference.

The reference is valid for the lifetime of the monitor.

pub fn with_verdict_representation<T: VerdictRepresentation>(
    self
) -> Monitor<Source, SourceTime, T, VerdictTime>

Switch VerdictRepresentations of the Monitor.