Enum StreamProcessor 

pub enum StreamProcessor<'a, A: 'a, B> {
    Get(Box<dyn FnOnce(A) -> StreamProcessor<'a, A, B> + 'a>),
    Put(B, Box<dyn FnOnce() -> StreamProcessor<'a, A, B> + 'a>),
}

StreamProcessor<A, B> defines (the syntax of) a language describing the domain of stream processors, that is, terms which can be interpreted to turn streams of type A into streams of type B.

Variants

Get(Box<dyn FnOnce(A) -> StreamProcessor<'a, A, B> + 'a>)

This stream processor first reads the A from the head of the input stream and subsequently applies its function argument to that element yielding a stream processor. The resulting stream processor is then used to process the input stream further depending on its shape: if it is a

• Get, it is applied to the tail of the input stream.
• Put, it is applied to the whole input stream.

Put(B, Box<dyn FnOnce() -> StreamProcessor<'a, A, B> + 'a>)

This stream processor writes the B from its first argument to the output list. Then, to construct the rest of the output list, it uses its second argument to process the input stream depending on its shape: if it is a

• Get, it is applied to the tail of the input stream.
• Put, it is applied to the whole input stream.

Implementations

impl<'a, A, B> StreamProcessor<'a, A, B>

pub fn get<F>(f: F) -> Self

where F: FnOnce(A) -> Self + 'a,

The same as StreamProcessor::Get but with boxing of f hidden to make the resulting code less verbose.

pub fn put<T>(b: B, lazy_sp: T) -> Self

where B: 'a, T: FnOnce() -> Self + 'a,

The same as StreamProcessor::Put but with boxing of lazy_sp hidden to make the resulting code less verbose.

impl<'a, A, B> StreamProcessor<'a, A, B>

pub fn eval<S: Stream<A> + 'a>(self, stream: S) -> InfiniteList<'a, B>

where A: Clone,

Evaluate self on an input stream essentially implementing a semantic of StreamProcessor<A, B>.

• stream is the input stream.

Note that the function can block the current thread if the respective implementation of Stream::tail can.

Panics

A panic may occur if

• the stream processor contains Rust-terms which can panic.
• the respective implementation of Stream::head or Stream::tail can panic.

Examples

Negating a stream of trues to obtain a stream of falses:

use rspl::StreamProcessor;

fn negate<'a>() -> StreamProcessor<'a, bool, bool> {
    StreamProcessor::get(|b: bool| StreamProcessor::put(!b, negate))
}

let trues = rspl::streams::infinite_lists::InfiniteList::constant(true);

negate().eval(trues);

Examples found in repository

examples/hics.rs (line 78)

        fn execute(self, mut cs: impl System<'a, Space>, epsilon: f64) {
            let mut status;

            // Here the measurements are generated (lazily).
            let mut positions = cs.meter().eval(InfiniteList::constant(()));

            loop {
                // Here the actual measurement is made.
                positions = positions.tail();
                let position = *positions.head();

                status = cs.quantity(position);
                let setpoint = cs.reference();
                let deviation = status - setpoint;

                if f64::abs(deviation) < epsilon {
                    break;
                }

                cs = cs.controller(deviation, status, position);

                thread::sleep(self.dwell_time);
            }
        }
    }
}

use control::Strategy;

use rspl::streams::infinite_lists::InfiniteList;
use rspl::streams::Stream;
use rspl::StreamProcessor;

use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::Duration;

use crossbeam::channel;
use crossbeam::channel::Sender;

// This constant is the window of tolerance for the heat index.
const EPSILON: f64 = 0.5;

const REFERENCE_HEAT_INDEX_DAY: f64 = 91.0;
const REFERENCE_HEAT_INDEX_NIGHT: f64 = 83.0;

const MINIMAL_TEMPERATURE: f64 = 80.0;
const MINIMAL_HUMIDITY: f64 = 50.0;

const INITIAL_TEMPERATURE: f64 = 87.0;
const INITIAL_HUMIDITY: f64 = 72.0;

const ACTUATOR_DECREASE: HeatIndexSpace = HeatIndexSpace {
    temperature: 0.25,
    humidity: 1.5,
};
const NATURAL_INCREASE: HeatIndexSpace = HeatIndexSpace {
    temperature: 0.02,
    humidity: 0.1,
};

// This block defines time-related constants. In particular, note that the `TICK` is intended to
// represent 10 real seconds.
const TICK_LENGTH: u64 = 5; // in (real) millis
const TICK: u64 = 1;
const DAY: u64 = 8640 * TICK;
const DWELL_TIME: u64 = 6 * TICK;
const CONTROL_PERIOD: u64 = 180 * TICK;
const NATURAL_INCREASE_PERIOD: u64 = 3 * TICK;

const UNSAFE_BARRIER: usize = 100_000;
const SERVICE_BARRIER: usize = UNSAFE_BARRIER - 5000;

type HeatIndex = f64;
type Time = u64;
type Clock = AtomicU64;

#[derive(Copy, Clone)]
struct HeatIndexSpace {
    temperature: f64, // in degree Fahrenheit
    humidity: f64,    // in percent
}

// This type defines the output signals of the hics. A signal is either status information
// (`Show(...)`) or orders for the actuator to execute (`Dehumidfy` and `Cool`).
enum HeatIndexSignal {
    Show(Time, HeatIndex),
    Dehumidify,
    Cool,
}

// This type is the actual hics. Essentially, it is the communication interface to its environment.
#[derive(Clone)]
struct Hics {
    clock_finger: Arc<Clock>,
    thermohygrometer_finger: Arc<Mutex<HeatIndexSpace>>,
    signals_s: Sender<HeatIndexSignal>,
}

impl<'a> control::System<'a, HeatIndexSpace> for Hics {
    fn meter(&self) -> StreamProcessor<'a, (), HeatIndexSpace> {
        fn read_out<'a, X: 'a + Copy>(finger: Arc<Mutex<X>>) -> StreamProcessor<'a, (), X> {
            StreamProcessor::Put(
                *Arc::clone(&finger).lock().unwrap(),
                Box::new(|| read_out(finger)),
            )
        }

        read_out(Arc::clone(&self.thermohygrometer_finger))
    }
    fn reference(&self) -> f64 {
        let time = self.clock_finger.load(Ordering::SeqCst);

        if time % DAY < DAY / 2 {
            REFERENCE_HEAT_INDEX_DAY
        } else {
            REFERENCE_HEAT_INDEX_NIGHT
        }
    }
    fn quantity(&self, position: HeatIndexSpace) -> f64 {
        // The body is the heat index formula from https://en.wikipedia.org/wiki/Heat_index.
        const C_1: f64 = -42.379;
        const C_2: f64 = 2.049_015_23;
        const C_3: f64 = 10.143_331_27;
        const C_4: f64 = -0.224_755_41;
        const C_5: f64 = -0.006_837_83;
        const C_6: f64 = -0.054_817_17;
        const C_7: f64 = 0.001_228_74;
        const C_8: f64 = 0.000_852_82;
        const C_9: f64 = -0.000_001_99;

        let t = position.temperature;
        let r = position.humidity;

        C_1 + C_2 * t
            + C_3 * r
            + C_4 * t * r
            + C_5 * t * t
            + C_6 * r * r
            + C_7 * t * t * r
            + C_8 * t * r * r
            + C_9 * t * t * r * r
    }
    fn controller(self, deviation: f64, status: f64, position: HeatIndexSpace) -> Self {
        let time = self.clock_finger.load(Ordering::SeqCst);
        self.signals_s
            .send(HeatIndexSignal::Show(time, status))
            .unwrap();

        if deviation > 0.0 {
            if position.humidity > MINIMAL_HUMIDITY {
                self.signals_s.send(HeatIndexSignal::Dehumidify).unwrap();
            } else if position.temperature > MINIMAL_TEMPERATURE {
                self.signals_s.send(HeatIndexSignal::Cool).unwrap();
            }
        }

        self
    }
}

#[allow(clippy::assertions_on_constants)]
fn driver(hics: Hics) {
    fn control<'a>(hics: Hics, mut counter: usize) -> StreamProcessor<'a, (), usize> {
        control::MeasureOnDemand {
            dwell_time: Duration::from_millis(DWELL_TIME * TICK_LENGTH),
        }
        .execute(hics.clone(), EPSILON);

        counter += 1;

        StreamProcessor::Put(counter, Box::new(move || control(hics, counter)))
    }

    assert!(UNSAFE_BARRIER > SERVICE_BARRIER);

    // Here the runs of the hics are generated (lazily).
    let mut runs = control(hics, 0).eval(InfiniteList::constant(()));

    loop {
        thread::sleep(Duration::from_millis(CONTROL_PERIOD * TICK_LENGTH));

        // Here, an iteration of the hics is started.
        runs = runs.tail();
        let run_count = *runs.head();

        if run_count > SERVICE_BARRIER {
            if run_count > UNSAFE_BARRIER {
                break;
            }
            println!(
                "Warning: Service needed. ({} runs > {} runs)",
                run_count, SERVICE_BARRIER
            );
        }
    }
}

examples/pelican.rs (line 315)

fn driver<S>(events: S, tfeedback: &Sender<Feedback>)
where
    S: Stream<Event>,
{
    // This code starts the machine.
    let mut capabilities = pelican_machine::on().eval(events);

    loop {
        // The following match provides the machine with capabilities. Most notably, it tells the
        // feedback loop to trigger a `Timeout`-event after some time.
        match *capabilities.head() {
            Capability::SetVehicleLights(color) => println!("Vehicles: {}", color),
            Capability::SetPedestrianLights(color) => println!("Pedestrians: {}", color),
            Capability::EmitTimeoutAfter(length) => {
                tfeedback.send(Feedback::TimeoutAfter(length)).unwrap();
            }
            Capability::UnexpectedTimeout(message) => {
                eprintln!("log: unexpected timeout event ({})", message);
            }
            Capability::CallForHelp => {
                println!("Call for help!");
                break;
            }
            Capability::Break => break,
        };
        capabilities = capabilities.tail();
    }
}