Expand stream module (#1028)

Author: phischu

effekt/shared/src/main/scala/effekt/core/PolymorphismBoxing.scala

@@ -476,6 +476,7 @@ object PolymorphismBoxing extends Phase[CoreTransformed, CoreTransformed] {
         // Do strings need to be boxed? Really?
         case core.Type.TString  => Pure.Literal("<?nothing>", core.Type.TString)
         case core.Type.TByte    => Pure.Literal(1337, core.Type.TByte)
+        case core.Type.TChar    => Pure.Literal('a', core.Type.TChar)
         case t if box.isDefinedAt(t) => sys error s"No default value defined for ${t}"
         case _ => sys error s"Trying to unbox Nothing to ${t}"
       }

examples/stdlib/stream/sum_of_squares.effekt

@@ -11,7 +11,7 @@ def main() = {
   println("The sum of squares from 1 to " ++ show(max) ++ " is:")
   println(sum {
     with boundary
-    with limit[Int](max + 1)
+    with limit[Int](max)
     squares()
   })
 }

examples/stdlib/stream/taylor.effekt

@@ -10,7 +10,7 @@ def ints(): Unit / emit[Int] = {
 def take[A](n: Int) { stream: () => Unit / emit[A] }: List[A] = {
   with collectList[A]
   with boundary
-  with limit[A](n + 1)
+  with limit[A](n)
   stream()
 }
 

examples/stdlib/stream/tee.check

@@ -3,7 +3,6 @@ a{ ... }
 2
 3
 4
-5
 Cons(1, Cons(2, Cons(3, Cons(4, Nil()))))
 a done
 b{ ... }
@@ -16,59 +15,54 @@ b{ ... }
 7
 8
 9
-10
 Cons(1, Cons(2, Cons(3, Cons(4, Cons(5, Cons(6, Cons(7, Cons(8, Cons(9, Nil())))))))))
 b done
 tee{a}{b}{ ... }
 1
 2
 3
 4
-5
 Cons(1, Cons(2, Cons(3, Cons(4, Nil()))))
 a done
+5
 6
 7
 8
 9
-10
 Cons(1, Cons(2, Cons(3, Cons(4, Cons(5, Cons(6, Cons(7, Cons(8, Cons(9, Nil())))))))))
 b done
 tee{b}{a}{ ... }
 1
 2
 3
 4
-5
 Cons(1, Cons(2, Cons(3, Cons(4, Nil()))))
 a done
+5
 6
 7
 8
 9
-10
 Cons(1, Cons(2, Cons(3, Cons(4, Cons(5, Cons(6, Cons(7, Cons(8, Cons(9, Nil())))))))))
 b done
 a{ teeing{b}{ ... } }
 1
 2
 3
 4
-5
 Cons(1, Cons(2, Cons(3, Cons(4, Nil()))))
 a done
 b{ teeing{a}{ ... } }
 1
 2
 3
 4
-5
 Cons(1, Cons(2, Cons(3, Cons(4, Nil()))))
 a done
+5
 6
 7
 8
 9
-10
 Cons(1, Cons(2, Cons(3, Cons(4, Cons(5, Cons(6, Cons(7, Cons(8, Cons(9, Nil())))))))))
 b done

examples/stdlib/stream/tee.effekt

@@ -11,11 +11,11 @@ def main() = {
     }
   }
   def a{ b: => Unit / emit[Int] }: Unit = {
-    println(collectList[Int]{boundary{limit[Int](5){b}}})
+    println(collectList[Int]{boundary{limit[Int](4){b}}})
     println("a done")
   }
   def b{ b: => Unit / emit[Int] }: Unit = {
-    println(collectList[Int]{boundary{limit[Int](10){b}}})
+    println(collectList[Int]{boundary{limit[Int](9){b}}})
     println("b done")
   }
   println("a{ ... }")

libraries/common/stream.effekt

@@ -30,6 +30,18 @@ effect stop(): Nothing
 def for[A] { stream: () => Unit / emit[A] } { action: A => Unit }: Unit =
   returning::for[A, Unit]{stream}{action}
 
+/// Applies the given function on each emitted element and reemits the result.
+///
+/// e.g. map[Int,String]{ x => x.show } { range(0, 5) }
+def map[A, B] { function: A => B } { stream: () => Unit / emit[A] }: Unit / emit[B] =
+  returning::map[A, B, Unit]{function}{stream}
+
+/// Runs the given action whenever the given reader reads an element.
+///
+/// e.g. var i = -1; with tap { i = i + 1; i }; reader()
+def tap[A] { action: () => A / stop } { reader: () => Unit / read[A] }: Unit =
+  returning::tap[A, Unit]{action}{reader}
+
 /// Runs the stream as a message queue,
 /// handling each of its elements that can produce new elements ("messages"),
 /// until a fixed point is reached.
@@ -162,8 +174,8 @@ def index[A] { stream: () => Unit / emit[A] }: Unit / emit[Indexed[A]] =
 
 
 /// If `number` is zero or negative it does nothing
-def limit[A](number: Int) { stream: () => Unit / emit[A] }: Unit / {emit[A], stop} =
-  returning::limit[A, Unit](number){stream}
+def limit[A](number: Int) { stream: () => Unit / emit[A] }: Unit / emit[A] =
+  boundary { returning::limit[A, Unit](number){stream} }
 
 
 /// If `number` is zero or negative it does nothing
@@ -405,6 +417,12 @@ def encodeUTF8[R] { stream: () => R / emit[Char] }: R / emit[Byte] =
     resume(encodeChar(char))
   }
 
+def writeFileUTF8[R](path: String) { stream: () => R / emit[Char] }: R / Exception[IOError] =
+  writeFile(path) { encodeUTF8 { stream() } }
+
+def readFileUTF8[R](path: String) { reader: () => R / read[Char] }: R / Exception[IOError] =
+  readFile(path) { decodeUTF8 { reader() } }
+
 def feed[R](string: String) { reader: () => R / read[Char] } =
   feed(string.fromString) {
     decodeUTF8 {
@@ -422,6 +440,12 @@ def each(string: String): Unit / emit[Char] =
 def collectString { stream: () => Unit / emit[Char] }: String =
   returning::collectString[Unit]{stream}.second
 
+def writeLine { body: () => Unit / emit[Char] }: Unit / emit[Char] =
+  returning::writeLine[Unit]{body}
+
+def readLine { body: () => Unit / read[Char] }: Unit / {read[Char], stop} =
+  returning::readLine[Unit]{body}
+
 namespace internal {
   effect snapshot(): Unit
 }
@@ -515,6 +539,20 @@ def for[A, R] { stream: () => R / emit[A] } { action: A => Unit }: R =
     resume(action(value))
   }
 
+def map[A, B, R] { function: A => B } { stream: () => R / emit[A] }: R / emit[B] =
+  try {
+    stream()
+  } with emit[A] { value =>
+    resume(do emit(function(value)))
+  }
+
+def tap[A, R] { action: () => A / stop } { reader: () => R / read[A] }: R =
+  try {
+    reader()
+  } with read[A] { () =>
+    resume { action() }
+  }
+
 def boundary[R] { program: () => R / stop }: Option[R] =
   try {
     Some(program())
@@ -542,17 +580,21 @@ def index[A, R] { stream: () => R / emit[A] }: R / emit[Indexed[A]] = {
 
 /// If `number` is zero or negative it does nothing
 def limit[A, R](number: Int) { stream: () => R / emit[A] }: R / { emit[A], stop } = {
-  if (number <= 0) do stop();
-  var i = 1;
-  try {
-    stream()
-  } with emit[A] { v =>
-    if (i < number) {
-      i = i + 1;
-      resume(do emit(v))
-    } else {
-      do stop()
+  if (number > 0) {
+    var i = number;
+    try {
+      stream()
+    } with emit[A] { v =>
+      do emit(v);
+      i = i - 1;
+      if (i > 0) {
+        resume(())
+      } else {
+        do stop()
+      }
     }
+  } else {
+    do stop()
   }
 }
 
@@ -615,6 +657,28 @@ def collectString[R] { stream: () => R / emit[Char] }: (R, String) = {
   (result, bytes.toString)
 }
 
+def writeLine[R] { body: () => R / emit[Char] }: R / emit[Char] = {
+  val result = body()
+  do emit('\n')
+  return result
+}
+
+def readLine[R] { body: () => R / read[Char] }: R / {read[Char], stop} = {
+  var stopped = false
+  try {
+    body()
+  } with read[Char] {
+    if(stopped){
+      resume { do stop() }
+    } else {
+      do read[Char] match {
+        case '\n' => stopped = true; resume { do stop() }
+        case char => resume { return char }
+      }
+    }
+  }
+}
+
 def source[A, R] { stream: () => Unit / emit[A] } { reader: () => R / read[A] }: R = {
   var next = box { None() }
   next = box {