src/pipeline/pipeline.oxl

use std.stream { map, filter, scan, dedup, batch }

/// Sensor readings, one every few milliseconds; some readings repeat.
pub fn readings() -> stream<i32>  time {
  for v in [3, 3, 4, 8, 8, 8, 15, 16, 23, 42] {
    sleep(2ms)
    yield v
  }
}

/// Distinct readings above `floor`, scaled by ten.
pub fn scaled(floor: i32) -> stream<i32>  time {
  readings() |> dedup() |> filter(x => *x > floor) |> map(x => x * 10)
}

/// Running sum of the readings.
pub fn running() -> stream<i32>  time {
  readings() |> scan(0, (acc, x) => acc + x)
}

/// Readings in groups of at most `n` (or what arrives within one second).
pub fn groups(n: usize) -> stream<List<i32>>  time {
  readings() |> batch(n, 1s)
}

/// Collects a stream into a list.
pub fn collect(s: stream<i32>) -> List<i32>  time {
  var out: List<i32> = []
  for await x in s { out.push(x) }
  out
}