flake.lock: Update

ARCHITECTURE.md

@@ -30,7 +30,7 @@ The entire library is built around Effekt's effect system. Instead of calling fu
 
 The main effects used are:
 
-- **`read[Event]`**: Pull-based stream input. When a component needs the next event, it performs `do read[Event]()`. A handler somewhere up the call stack provides the actual value.
+- **`next[Event]`**: Pull-based stream input. When a component needs the next event, it performs `do next[Event]()`. A handler somewhere up the call stack provides the actual value.
 - **`emit[Event]`**: Push-based stream output. Components emit results using `do emit(event)`. Handlers decide what to do with each value.
 - **`AnomalyDetection`**: Interface with `anomaly()` and `noAnomaly()` operations. Anomaly detectors call these, and handlers decide how to react (log, alert, store, etc.).
 - **`stop`**: Used to signal the end of the stream.
@@ -60,13 +60,13 @@ For example, all anomaly loggers handle the `AnomalyDetection` effect the same w
 
 Switching from console to file logging is just changing one line. The rest of the pipeline stays exactly the same.
 
-Similarly for the rest of the library - whether you use `csvFeedPull()`, `normalDistributedPull()`, or `cpuUsagePullStream()`, they all provide events via `read[Event]`, so any aggregator works with any input source and any anomaly detector works with any aggregator.
+Similarly for the rest of the library - whether you use `csvFeedPull()`, `normalDistributedPull()`, or `cpuUsagePullStream()`, they all provide events via `next[Event]`, so any aggregator works with any input source and any anomaly detector works with any aggregator.
 
 ### Stream Model
 
 The library uses two stream models:
 
-1. **Pull streams**: The consumer requests data by performing `read[Event]`. The handler decides where the data comes from. This is lazy—no work happens until someone reads.
+1. **Pull streams**: The consumer requests data by performing `next[Event]`. The handler decides where the data comes from. This is lazy—no work happens until someone reads.
 
 2. **Push streams**: The producer emits data by performing `emit[Event]`. The handler decides what to do with each value.
 
@@ -139,10 +139,10 @@ The library uses minimal FFI:
 To add your own detection strategy, write a function that reads events and calls the `AnomalyDetection` effect:
 
 ```js
-def myCustomDetector(param: Double): Unit / { read[Event], AnomalyDetection } = {
+def myCustomDetector(param: Double): Unit / { next[Event], AnomalyDetection } = {
   with boundary
   while (true) {
-    val ev = do read[Event]()
+    val ev = do next[Event]()
     // your detection logic here
     if (isAnomalous) {
       do anomaly(AnomalyEvaluation(ev, true, score))
@@ -155,11 +155,11 @@ def myCustomDetector(param: Double): Unit / { read[Event], AnomalyDetection } =
 
 ### Writing a Custom Input Source
 
-Implement a handler for `read[Event]`:
+Implement a handler for `next[Event]`:
 
 ```js
-def myDataSource() { body: () => Unit / read[Event] }: Unit = {
-  try body() with read[Event] {
+def myDataSource() { body: () => Unit / next[Event] }: Unit = {
+  try body() with next[Event] {
     val nextEvent = // fetch from somewhere
     resume { nextEvent }
   }

README.md

@@ -119,7 +119,7 @@ You can run the examples by calling the function in `src/main.effekt` and then e
 
 ### Data Input Sources
 
-The library provides various input sources that handle `read[Event]`. All input handlers share the same interface, so they can be swapped without changing the rest of the pipeline.
+The library provides various input sources that handle `next[Event]`. All input handlers share the same interface, so they can be swapped without changing the rest of the pipeline.
 
 #### CSV Input (`src/lib/stream_input`)
 
@@ -159,7 +159,7 @@ Usage examples are provided in the `examples` namespace within each component fi
 
 ### Aggregation Functions
 
-All aggregators read events via `read[Event]` and emit aggregated results via `emit[Event]`.
+All aggregators read events via `next[Event]` and emit aggregated results via `emit[Event]`.
 
 | Function | Description | Space | Time per event |
 |----------|-------------|-------|----------------|
@@ -175,7 +175,7 @@ All aggregators read events via `read[Event]` and emit aggregated results via `e
 
 ### Anomaly Detection
 
-The library provides three anomaly detection strategies (`src/lib/anomaly_detection`). Each detector reads events via `read[Event]` and calls the `AnomalyDetection` interface (`anomaly()` or `noAnomaly()`).
+The library provides three anomaly detection strategies (`src/lib/anomaly_detection`). Each detector reads events via `next[Event]` and calls the `AnomalyDetection` interface (`anomaly()` or `noAnomaly()`).
 
 | Detector | Description |
 |----------|-------------|
@@ -193,7 +193,7 @@ The library provides handlers to log events or anomaly evaluations. "Rethrow" va
 
 #### Event Logging (`src/lib/event_logger`)
 
-Handles `emit[Event]` (push) or `read[Event]` (pull) effects.
+Handles `emit[Event]` (push) or `next[Event]` (pull) effects.
 
 | Handler | Description |
 |---------|-------------|
@@ -225,7 +225,7 @@ Usage examples are provided in the `examples` namespace within each logger file.
 
 ## Architecture Overview
 
-The library is built around Effekt's effect system. Components communicate through effects (`read[Event]`, `emit[Event]`, `AnomalyDetection`) that get handled by outer handlers. This design enables:
+The library is built around Effekt's effect system. Components communicate through effects (`next[Event]`, `emit[Event]`, `AnomalyDetection`) that get handled by outer handlers. This design enables:
 
 - **Composability**: Stack handlers to build pipelines
 - **Exchangeability**: Swap any handler for another without changing the rest of the pipeline

src/lib/aggregation.effekt

@@ -17,11 +17,11 @@ extern def negInfinity: Double =
 /// Aggregate the minimum value from an unbounded stream of events
 /// 
 /// O(1) space complexity, O(1) time complexity per read
-def aggregateMin(): Unit / { read[Event], emit[Event] } = {
+def aggregateMin(): Unit / { next[Event], emit[Event] } = {
   with boundary
   var min = infinity()
   while (true) {
-    val ev = do read[Event]()
+    val ev = do next[Event]()
     val value = ev.value
     if (value < min) {
       min = value
@@ -33,14 +33,14 @@ def aggregateMin(): Unit / { read[Event], emit[Event] } = {
 /// Aggregate the minimum value over a sliding window of given size
 /// 
 /// O(windowSize) space complexity, O(1) time complexity on average per read
-def aggregateMinWindow(windowSize: Int) : Unit / { read[Event], emit[Event] } = {
+def aggregateMinWindow(windowSize: Int) : Unit / { next[Event], emit[Event] } = {
   with boundary
   with on[OutOfBounds].panic()
   val windowArr = array[Double](windowSize, infinity()) // Array to cycle through window values
   var min = infinity()
   var index = 0
   while (true) {
-    val ev = do read[Event]()
+    val ev = do next[Event]()
     val value = ev.value
     if (windowArr.get(index) == min) {
       // Recompute min
@@ -63,11 +63,11 @@ def aggregateMinWindow(windowSize: Int) : Unit / { read[Event], emit[Event] } =
 /// Aggregate the maximum value from an unbounded stream of events
 /// 
 /// O(1) space complexity, O(1) time complexity per read
-def aggregateMax(): Unit / { read[Event], emit[Event] } = {
+def aggregateMax(): Unit / { next[Event], emit[Event] } = {
   with boundary
   var max = negInfinity()
   while (true) {
-    val ev = do read[Event]()
+    val ev = do next[Event]()
     val value = ev.value
     if (value > max) {
       max = value
@@ -79,14 +79,14 @@ def aggregateMax(): Unit / { read[Event], emit[Event] } = {
 /// Aggregate the maximum value over a sliding window of given size
 /// 
 /// O(windowSize) space complexity, O(1) time complexity on average per read
-def aggregateMaxWindow(windowSize: Int) : Unit / { read[Event], emit[Event] } = {
+def aggregateMaxWindow(windowSize: Int) : Unit / { next[Event], emit[Event] } = {
   with boundary
   with on[OutOfBounds].panic()
   val windowArr = array[Double](windowSize, negInfinity()) // Array to cycle through window values
   var max = negInfinity()
   var index = 0
   while (true) {
-    val ev = do read[Event]()
+    val ev = do next[Event]()
     val value = ev.value
     if (windowArr.get(index) == max) {
       // Recompute max -> O(windowSize)
@@ -109,11 +109,11 @@ def aggregateMaxWindow(windowSize: Int) : Unit / { read[Event], emit[Event] } =
 /// Aggregate the mean value from an unbounded stream of events
 /// 
 /// O(1) space complexity, O(1) time complexity per read
-def aggregateMean(): Unit / { read[Event], emit[Event] } = {
+def aggregateMean(): Unit / { next[Event], emit[Event] } = {
   with boundary
   var counter: Counter = counter::empty()
   while (true) {
-    val ev = do read[Event]()
+    val ev = do next[Event]()
     val value = ev.value
     counter = counter.add(value)
     val meanVal = counter.mean()
@@ -124,14 +124,14 @@ def aggregateMean(): Unit / { read[Event], emit[Event] } = {
 /// Aggregate the mean value over a sliding window of given size
 /// 
 /// O(windowSize) space complexity, O(1) time complexity per read
-def aggregateMeanWindow(windowSize: Int): Unit / { read[Event], emit[Event] } = {
+def aggregateMeanWindow(windowSize: Int): Unit / { next[Event], emit[Event] } = {
   with boundary
   with on[OutOfBounds].panic()
   val windowArr = array[Double](windowSize, negInfinity()) // Array to cycle through window values
   var counter = counter::empty()
   var index = 0
   while (true) {
-    val ev = do read[Event]()
+    val ev = do next[Event]()
     val value = ev.value
     if (windowArr.get(index) != negInfinity()) {
       counter = counter.remove(windowArr.get(index))
@@ -146,13 +146,13 @@ def aggregateMeanWindow(windowSize: Int): Unit / { read[Event], emit[Event] } =
 /// Aggregate the median value from an unbounded stream of events
 /// 
 /// O(n) space complexity, O(n log n) time complexity per read
-def aggregateMedian(): Unit / { read[Event], emit[Event] } = {
+def aggregateMedian(): Unit / { next[Event], emit[Event] } = {
   with boundary
   with on[OutOfBounds].panic()
   val windowList: List[Double] = empty[Double]()
   var lst = windowList
   while (true) {
-    val ev = do read[Event]()
+    val ev = do next[Event]()
     lst = Cons(ev.value, lst)
     val medianVal = internal::listMedian(lst)
     do emit(Event(ev.timestamp, medianVal))
@@ -162,12 +162,12 @@ def aggregateMedian(): Unit / { read[Event], emit[Event] } = {
 /// Aggregate the median value over a sliding window of given size
 /// 
 /// O(windowSize) space complexity, O(n log n) time complexity per read
-def aggregateMedianWindow(windowSize: Int): Unit / { read[Event], emit[Event] } = {
+def aggregateMedianWindow(windowSize: Int): Unit / { next[Event], emit[Event] } = {
   with boundary
   with on[OutOfBounds].panic()
   var lst: List[Double] = empty[Double]()
   while (true) {
-    val ev = do read[Event]()
+    val ev = do next[Event]()
     lst = Cons(ev.value, lst)
     if (lst.size() > windowSize) {
       lst = lst.deleteAt(lst.size() - 1) // remove oldest element
@@ -192,7 +192,7 @@ namespace examples {
     ]
     try {
       aggregateMin()
-    } with read[Event] { () =>
+    } with next[Event] { () =>
       if (lst.isEmpty()) {
         println("No more values to read, stopping...")
         resume { do stop() }
@@ -219,7 +219,7 @@ namespace examples {
     ]
     try {
       aggregateMeanWindow(2)
-    } with read[Event] { () =>
+    } with next[Event] { () =>
       if (values.isEmpty()) {
         println("No further input values; shutting down...")
         resume { do stop() }

src/lib/cpu_utilization_input.effekt

@@ -48,8 +48,8 @@ extern def getCpuUsage() at async: Double =
     })"""
 
 /// Infinite pull stream that yields CPU usage events
-def cpuUsagePullStream() { body: () => Unit / { read[Event] } }: Unit = {
-  try body() with read[Event] {
+def cpuUsagePullStream() { body: () => Unit / { next[Event] } }: Unit = {
+  try body() with next[Event] {
     resume {
       val cpuUsage = getCpuUsage()
       val timestamp = currentTimestampMillis()
@@ -59,9 +59,9 @@ def cpuUsagePullStream() { body: () => Unit / { read[Event] } }: Unit = {
 }
 
 /// Pull stream that yields at most `limit` CPU usage events
-def cpuUsagePullStream(limit: Int) { body: () => Unit / { read[Event] } }: Unit = {
+def cpuUsagePullStream(limit: Int) { body: () => Unit / { next[Event] } }: Unit = {
   var count = 0
-  try body() with read[Event] {
+  try body() with next[Event] {
     resume {
       if (count >= limit) {
         do stop()
@@ -81,7 +81,7 @@ namespace examples {
     var i = 0
     cpuUsagePullStream() {
       while (i < 10) {
-        println("CPU Usage Event received: " ++ do read[Event]().show())
+        println("CPU Usage Event received: " ++ do next[Event]().show())
         i = i + 1
         time::sleep(500)
       }

src/lib/csv.effekt

@@ -114,9 +114,9 @@ def readUntil { shouldStop: Char => Bool }: String / Scan[Char] =
 def readQuotedString(): Unit / { Scan[Char], emit[Char], Exception[WrongFormat] } = {
   try {
     expect("Expected \"") { readIf('"') }
-    while(do read[Char]() is c and c != '"') {
+    while(do next[Char]() is c and c != '"') {
       c match {
-        case '\\' => do read[Char]() match {
+        case '\\' => do next[Char]() match {
           case '"' => do emit('\"')
           case '\\' => do emit('\\')
           case '/' => do emit('/')
@@ -302,7 +302,7 @@ def main() = {
   // Empty row!
   val realCsv = ",name,age\n\n,John,42\n,Joe,10\n,Jolene,27"
   val res = feed(realCsv) {
-    with returning::scanner[Char, CsvValue]
+    with returning::peeking[Char, CsvValue]
     build { decodeCsv() }.second
   }
   inspect(res)

src/lib/event_logger.effekt

@@ -28,12 +28,12 @@ def logToConsoleRethrow() {body: () => Unit / emit[Event]}: Unit / emit[Event] =
 }
 
 /// An AnomalyDetection handler that logs all events to the console for pull-based streams
-def logToConsolePull() {body: () => Unit / read[Event]}: Unit / read[Event] = {
+def logToConsolePull() {body: () => Unit / next[Event]}: Unit / next[Event] = {
   with boundary
   try {
     body()
-  } with read[Event] {
-    val e = do read[Event]()
+  } with next[Event] {
+    val e = do next[Event]()
     resume {
       println("Event: " ++ e.value.show() ++ " at t=" ++ e.timestamp.show())
       e
@@ -96,15 +96,15 @@ def logToFileCsvRethrow(filePath: String) {body: () => Unit / emit[Event]}: Unit
   }
 }
 
-def logToFileCsvPull(filePath: String) {body: () => Unit / read[Event]}: Unit / { read[Event], Exception[IOError] } = {
+def logToFileCsvPull(filePath: String) {body: () => Unit / next[Event]}: Unit / { next[Event], Exception[IOError] } = {
     var buffer: List[CsvRow] = [["timestamp", "value"]]
     var first = true
     val maxBufferSize = 1000
   try { // Double try is required because the read in the with block throws stop when the stream ends
     try {
       body()
-    } with read[Event] {
-      val e = do read[Event]()
+    } with next[Event] {
+      val e = do next[Event]()
       resume {
         val row: CsvRow = [
           e.timestamp.show(),