Move stream module to prelude

Author: phischu

effekt/jvm/src/main/scala/effekt/Runner.scala

@@ -43,7 +43,7 @@ trait Runner[Executable] {
    * if module A depends on module B, then B should come before A.
    * - Furthermore, each module mentioned here must import the `effekt` module as its first import.
    */
-  def prelude: List[String] = List("effekt", "option", "list", "result", "exception", "array", "char", "string", "ref")
+  def prelude: List[String] = List("effekt", "option", "stream", "list", "result", "exception", "array", "char", "bytearray", "string", "ref")
 
   /**
    * Creates a OS-specific script file that will execute the command when executed,

libraries/common/array.effekt

@@ -4,6 +4,7 @@ import effekt
 import exception
 import list
 import option
+import stream
 
 /// A mutable 0-indexed fixed-sized array.
 extern type Array[T]
@@ -641,3 +642,55 @@ def println(l: Array[Int]): Unit = println(show(l))
 def println(l: Array[Double]): Unit = println(show(l))
 def println(l: Array[Bool]): Unit = println(show(l))
 def println(l: Array[String]): Unit = println(show(l))
+
+
+// Streaming
+// ---------
+
+/// Turns an `array` into a producer of a push stream
+/// by emitting each contained value from 0 to length - 1.
+def each[T](array: Array[T]): Unit / emit[T] = {
+  val n = array.size
+  def go(i: Int): Unit = {
+    if (i < n) {
+      do emit(array.unsafeGet(i))
+      go(i + 1)
+    }
+  }
+  go(0)
+}
+
+def feed[T, R](array: Array[T]) { reader: () => R / read[T] }: R = {
+  var i = 0
+  try {
+    reader()
+  } with read[T] {
+    resume {
+      if (i < array.size) {
+        val c = i
+        i = c + 1
+        array.unsafeGet(c)
+      } else {
+        do stop()
+      }
+    }
+  }
+}
+
+def collectArray[A] { stream: () => Unit / emit[A] }: Array[A] =
+  returning::collectArray[A, Unit]{stream}.second
+
+namespace returning {
+  def collectArray[A, R] { stream: () => R / emit[A] }: (R, Array[A]) = {
+    var i = 0
+    var a = allocate(1)
+    try {
+      (stream(), a.resize(i))
+    } with emit[A] { (v) =>
+      if (i >= a.size) { a = a.resize(2 * a.size) }
+      a.unsafeSet(i, v)
+      i = i + 1
+      resume(())
+    }
+  }
+}

libraries/common/bytearray.effekt

@@ -1,5 +1,8 @@
 module bytearray
 
+import effekt
+import stream
+
 /**
  * A memory managed, mutable, fixed-length array of bytes.
  *
@@ -200,3 +203,55 @@ def compareStringBytes(left: String, right: String): Ordering = {
   val r = right.fromString
   compareByteArray(l, r)
 }
+
+
+// Streaming
+// ---------
+
+/// Turns `bytes` into a producer of a push stream
+/// by emitting each contained value from 0 to length - 1.
+def each(bytes: ByteArray): Unit / emit[Byte] = {
+  val n = bytes.size
+  def go(i: Int): Unit = {
+    if (i < n) {
+      do emit(bytes.unsafeGet(i))
+      go(i + 1)
+    }
+  }
+  go(0)
+}
+
+def feed[R](bytes: ByteArray) { reader: () => R / read[Byte] }: R = {
+  var i = 0
+  try {
+    reader()
+  } with read[Byte] {
+    resume {
+      if (i < bytes.size) {
+        val c = i
+        i = c + 1
+        bytes.unsafeGet(c)
+      } else {
+        do stop()
+      }
+    }
+  }
+}
+
+def collectBytes { stream: () => Unit / emit[Byte] }: ByteArray =
+  returning::collectBytes[Unit]{stream}.second
+
+namespace returning {
+  def collectBytes[R] { stream: () => R / emit[Byte] }: (R, ByteArray) = {
+    var i = 0
+    var a = allocate(1)
+    try {
+      (stream(), a.resize(i))
+    } with emit[Byte] { (v) =>
+      if (i >= a.size) { a = a.resize(2 * a.size) }
+      a.unsafeSet(i, v)
+      i = i + 1
+      resume(())
+    }
+  }
+}

libraries/common/char.effekt

@@ -5,6 +5,7 @@ import effekt
 import exception
 import option
 import result
+import stream
 
 
 /// Checks if the given character is an ASCII whitespace
@@ -175,3 +176,89 @@ def utf16UnitCount(codepoint: Char): Int = codepoint match {
 extern def charWidth(c: Char) at {}: Int =
   // JavaScript strings are UTF-16 where every unicode character after 0xffff takes two units
   js "(${c} > 0xffff) ? 2 : 1"
+
+
+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 returning {
+  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 decodeChar(): Char / {read[Byte], stop} = {
+  val b = do read().toInt
+  if (b < 128) {
+    b.toChar
+  } else if (b < 224) {
+    val part1 = bitwiseShl(bitwiseAnd(b, 31), 6)
+    val part2 = bitwiseAnd(do read().toInt, 63)
+    bitwiseOr(part1, part2).toChar
+  } else if (b < 240) {
+    val part1 = bitwiseShl(bitwiseAnd(b, 15), 12)
+    val part2 = bitwiseShl(bitwiseAnd(do read().toInt, 63), 6)
+    val part3 = bitwiseAnd(do read().toInt, 63)
+    bitwiseOr(bitwiseOr(part1, part2), part3).toChar
+  } else {
+    val part1 = bitwiseShl(bitwiseAnd(b, 7), 18)
+    val part2 = bitwiseShl(bitwiseAnd(do read().toInt, 63), 12)
+    val part3 = bitwiseShl(bitwiseAnd(do read().toInt, 63), 6)
+    val part4 = bitwiseAnd(do read().toInt, 63)
+    bitwiseOr(bitwiseOr(bitwiseOr(part1, part2), part3), part4).toChar
+  }
+}
+
+def encodeChar(char: Char): Unit / emit[Byte] = {
+  val code = char.toInt
+  if (code < 128) {
+    do emit(code.toByte)
+  } else if (code < 2048) {
+    do emit(bitwiseOr(192, bitwiseShr(code, 6)).toByte)
+    do emit(bitwiseOr(128, bitwiseAnd(code, 63)).toByte)
+  } else if (code < 65536) {
+    do emit(bitwiseOr(224, bitwiseShr(code, 12)).toByte)
+    do emit(bitwiseOr(128, bitwiseAnd(bitwiseShr(code, 6), 63)).toByte)
+    do emit(bitwiseOr(128, bitwiseAnd(code, 63)).toByte)
+  } else {
+    do emit(bitwiseOr(240, bitwiseShr(code, 18)).toByte)
+    do emit(bitwiseOr(128, bitwiseAnd(bitwiseShr(code, 12), 63)).toByte)
+    do emit(bitwiseOr(128, bitwiseAnd(bitwiseShr(code, 6), 63)).toByte)
+    do emit(bitwiseOr(128, bitwiseAnd(code, 63)).toByte)
+  }
+}
+
+def decodeUTF8[R] { reader: () => R / read[Char] }: R / read[Byte] =
+  try {
+    reader()
+  } with read[Char] {
+    resume { decodeChar() }
+  }
+
+def encodeUTF8[R] { stream: () => R / emit[Char] }: R / emit[Byte] =
+  try {
+    stream()
+  } with emit[Char] { char =>
+    resume(encodeChar(char))
+  }

libraries/common/io/filesystem.effekt

@@ -71,6 +71,77 @@ def appendFile(path: String, contents: String): Unit / Exception[IOError] = {
   go()
 }
 
+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 writeFile[R](path: String) { stream: () => R / emit[Byte] }: R / Exception[IOError] = {
+
+  val file = openForWriting(path);
+  with on[IOError].finalize { close(file) }
+
+  val chunkSize = 1048576 // 1MB
+  val buffer = bytearray::allocate(chunkSize)
+  var offset = 0
+
+  def push(i: Int, n: Int): Unit = {
+    val r = write(file, buffer, i, n, -1)
+    if (r < n) {
+      push(i + r, n - r)
+    }
+  }
+
+  try {
+    val r = stream()
+    push(0, offset)
+    return r
+  } with emit[Byte] { (byte) =>
+    if (offset >= buffer.size) {
+      push(0, buffer.size)
+      offset = 0
+    }
+    buffer.unsafeSet(offset, byte)
+    offset = offset + 1
+    resume(())
+  }
+}
+
+def readFile[R](path: String) { reader: () => R / read[Byte] }: R / Exception[IOError] = {
+
+  val file = openForReading(path);
+  with on[IOError].finalize { close(file) }
+
+  val chunkSize = 1048576 // 1MB
+  val buffer = bytearray::allocate(chunkSize)
+  var offset = 0
+  var length = 0
+
+  def pull(): Unit / stop = {
+    read(file, buffer, 0, chunkSize, -1) match {
+      case 0 =>
+        do stop()
+      case n =>
+        length = n
+    }
+  }
+
+  try {
+    reader()
+  } with read[Byte] {
+    resume {
+      if (offset >= length) {
+        pull()
+        offset = 0
+      }
+      val byte = buffer.unsafeGet(offset)
+      offset = offset + 1
+      return byte
+    }
+  }
+}
+
 /// An abstract interface applications can program against.
 ///
 /// Can be interpreted with the `filesystem` handler, or virtualized etc.

libraries/common/list.effekt

@@ -3,6 +3,7 @@ module list
 import effekt
 import option
 import exception
+import stream
 
 /// Immutable linked list for finite sequences of elements.
 type List[A] {
@@ -836,3 +837,46 @@ def println(l: List[Int]): Unit = println(show(l))
 def println(l: List[Double]): Unit = println(show(l))
 def println(l: List[Bool]): Unit = println(show(l))
 def println(l: List[String]): Unit = println(show(l))
+
+
+// Streaming
+// ---------
+
+/// Turns a `list` into a producer of a push stream
+/// by emitting each contained value left-to-right.
+def each[A](list: List[A]): Unit / emit[A] =
+  list match {
+    case Nil() => ()
+    case Cons(head, tail) =>
+      do emit(head)
+      each(tail)
+  }
+
+def feed[T, R](list: List[T]) { reader: () => R / read[T] }: R = {
+  var l = list
+  try {
+    reader()
+  } with read[T] {
+    resume {
+      l match {
+        case Nil() => do stop()
+        case Cons(value, rest) =>
+          l = rest
+          return value
+      }
+    }
+  }
+}
+
+def collectList[A] { stream: () => Unit / emit[A] }: List[A] =
+  returning::collectList[A, Unit]{stream}.second
+
+namespace returning {
+  def collectList[A, R] { stream: () => R / emit[A] }: (R, List[A]) =
+    try {
+      (stream(), Nil())
+    } with emit[A] { (v) =>
+      val (r, vs) = resume(());
+      (r, Cons(v, vs))
+    }
+}