Merge pull request #20 from maxm123/explicit-cont-listener

This bundles the foundations for a platform-independent implementation of Gears, as well as the listener API and channels with read and write operation sources, capable of racing (selecting). It is the first of a couple upcoming steps to carve out the interface that will be provided by Gears.

.github/workflows/ci.yml

@@ -15,6 +15,6 @@ jobs:
     - uses: actions/setup-java@v3
       with:
         distribution: temurin
-        java-version: 20
+        java-version: 21
     - name: Test
       run: sbt test

.jvmopts

@@ -1 +0,0 @@
---enable-preview

build.sbt

@@ -6,6 +6,5 @@ lazy val root = project
     name := "Gears",
     organization := "ch.epfl.lamp",
     version := "0.1.0-SNAPSHOT",
-    javaOptions += "--enable-preview --version 19",
     libraryDependencies += "org.scalameta" %% "munit" % "0.7.29" % Test
   )

project/build.properties

@@ -1 +1 @@
-sbt.version=1.8.2
+sbt.version=1.9.6

src/main/scala/PosixLikeIO/PIO.scala

@@ -0,0 +1,140 @@
+package PosixLikeIO
+
+import gears.async.{Async, Future, given}
+import Future.Promise
+
+import java.net.{DatagramPacket, DatagramSocket, InetAddress, InetSocketAddress, ServerSocket, Socket}
+import java.nio.ByteBuffer
+import java.nio.channels.{AsynchronousFileChannel, CompletionHandler, SocketChannel}
+import java.nio.charset.{Charset, StandardCharsets}
+import java.nio.file.{Path, StandardOpenOption}
+import scala.Tuple.Union
+import scala.concurrent.ExecutionContext
+import scala.util.{Failure, Success, Try}
+
+
+class File(val path: String) {
+  private var channel: Option[AsynchronousFileChannel] = None
+
+  def isOpened: Boolean = channel.isDefined && channel.get.isOpen
+
+  def open(options: StandardOpenOption*): File =
+    assert(channel.isEmpty)
+    val options1 = if (options.isEmpty) Seq(StandardOpenOption.READ) else options
+    channel = Some(
+      AsynchronousFileChannel.open(Path.of(path), options1*))
+    this
+
+  def close(): Unit =
+    if (channel.isDefined)
+      channel.get.close()
+      channel = None
+
+  def read(buffer: ByteBuffer): Future[Int] =
+    assert(channel.isDefined)
+
+    val p = Promise[Int]()
+    channel.get.read(buffer, 0, buffer, new CompletionHandler[Integer, ByteBuffer] {
+      override def completed(result: Integer, attachment: ByteBuffer): Unit = p.complete(Success(result))
+      override def failed(e: Throwable, attachment: ByteBuffer): Unit = p.complete(Failure(e))
+    })
+    p.future
+
+  def readString(size: Int, charset: Charset = StandardCharsets.UTF_8): Future[String] =
+    assert(channel.isDefined)
+    assert(size >= 0)
+
+    val buffer = ByteBuffer.allocate(size)
+    val p = Promise[String]()
+    channel.get.read(buffer, 0, buffer, new CompletionHandler[Integer, ByteBuffer] {
+      override def completed(result: Integer, attachment: ByteBuffer): Unit =
+        p.complete(Success(charset.decode(attachment.slice(0, result)).toString()))
+      override def failed(e: Throwable, attachment: ByteBuffer): Unit = p.complete(Failure(e))
+    })
+    p.future
+
+  def write(buffer: ByteBuffer): Future[Int] =
+    assert(channel.isDefined)
+
+    val p = Promise[Int]()
+    channel.get.write(buffer, 0, buffer, new CompletionHandler[Integer, ByteBuffer] {
+      override def completed(result: Integer, attachment: ByteBuffer): Unit = p.complete(Success(result))
+      override def failed(e: Throwable, attachment: ByteBuffer): Unit = p.complete(Failure(e))
+    })
+    p.future
+
+
+  def writeString(s: String, charset: Charset = StandardCharsets.UTF_8): Future[Int] =
+    write(ByteBuffer.wrap(s.getBytes(charset)))
+
+  override def finalize(): Unit = {
+    super.finalize()
+    if (channel.isDefined)
+      channel.get.close()
+  }
+}
+
+class SocketUDP() {
+  private var socket: Option[DatagramSocket] = None
+
+  def isOpened: Boolean = socket.isDefined && !socket.get.isClosed
+
+  def bindAndOpen(port: Int): SocketUDP =
+    assert(socket.isEmpty)
+    socket = Some(DatagramSocket(port))
+    this
+
+  def open(): SocketUDP =
+    assert(socket.isEmpty)
+    socket = Some(DatagramSocket())
+    this
+
+  def close(): Unit =
+    if (socket.isDefined)
+      socket.get.close()
+      socket = None
+
+  def send(data: ByteBuffer, address: String, port: Int): Future[Unit] =
+    assert(socket.isDefined)
+
+    Async.blocking:
+      Future:
+        val packet: DatagramPacket = new DatagramPacket(data.array(), data.limit(), InetAddress.getByName(address), port)
+        socket.get.send(packet)
+
+  def receive(): Future[DatagramPacket] =
+    assert(socket.isDefined)
+
+    Async.blocking:
+      Future[DatagramPacket]:
+        val buffer = Array.fill[Byte](10 * 1024)(0)
+        val packet: DatagramPacket = DatagramPacket(buffer, 10 * 1024)
+        socket.get.receive(packet)
+        packet
+
+  override def finalize(): Unit = {
+    super.finalize()
+    if (socket.isDefined)
+      socket.get.close()
+  }
+}
+
+object PIOHelper {
+  def withFile[T](path: String, options: StandardOpenOption*)(f: File => T): T =
+    val file = File(path).open(options*)
+    val ret = f(file)
+    file.close()
+    ret
+
+  def withSocketUDP[T]()(f: SocketUDP => T): T =
+    val s = SocketUDP().open()
+    val ret = f(s)
+    s.close()
+    ret
+
+  def withSocketUDP[T](port: Int)(f: SocketUDP => T): T =
+    val s = SocketUDP().bindAndOpen(port)
+    val ret = f(s)
+    s.close()
+    ret
+}

src/main/scala/PosixLikeIO/examples/clientAndServerUDP.scala

@@ -0,0 +1,35 @@
+package PosixLikeIO.examples
+
+import gears.async.{Async, Future, given}
+import gears.async.AsyncOperations.*
+import PosixLikeIO.{PIOHelper, SocketUDP}
+
+import java.net.DatagramPacket
+import java.nio.ByteBuffer
+import java.nio.file.StandardOpenOption
+import scala.concurrent.ExecutionContext
+
+
+@main def clientAndServerUDP(): Unit =
+  given ExecutionContext = ExecutionContext.global
+  Async.blocking:
+    val server = Future:
+      PIOHelper.withSocketUDP(8134): serverSocket =>
+        val got: DatagramPacket = serverSocket.receive().result.get
+        val messageReceived = String(got.getData.slice(0, got.getLength), "UTF-8")
+        val responseMessage = (messageReceived.toInt + 1).toString.getBytes
+        serverSocket.send(ByteBuffer.wrap(responseMessage), got.getAddress.toString.substring(1), got.getPort)
+        sleep(50)
+
+    def client(value: Int): Future[Unit] =
+      Future:
+        PIOHelper.withSocketUDP(): clientSocket =>
+          val data: Array[Byte] = value.toString.getBytes
+          clientSocket.send(ByteBuffer.wrap(data), "localhost", 8134).result.get
+          val responseDatagram = clientSocket.receive().result.get
+          val messageReceived = String(responseDatagram.getData.slice(0, responseDatagram.getLength), "UTF-8").toInt
+          println("Sent " + value.toString + " and got " + messageReceived.toString + " in return.")
+
+
+    Async.await(client(100))
+    Async.await(server)

src/main/scala/PosixLikeIO/examples/readAndWriteFile.scala

@@ -0,0 +1,19 @@
+package PosixLikeIO.examples
+
+import gears.async.{Async, given}
+import PosixLikeIO.PIOHelper
+
+import java.nio.ByteBuffer
+import java.nio.charset.StandardCharsets
+import java.nio.file.StandardOpenOption
+import scala.concurrent.ExecutionContext
+
+
+@main def readAndWriteFile(): Unit =
+  given ExecutionContext = ExecutionContext.global
+  Async.blocking:
+    PIOHelper.withFile("/home/julian/Desktop/x.txt", StandardOpenOption.READ, StandardOpenOption.WRITE): f =>
+      Async.await(f.writeString("Hello world! (1)"))
+      println(Async.await(f.readString(1024)).get)
+      Async.await(f.writeString("Hello world! (2)"))
+      println(Async.await(f.readString(1024)).get)

src/main/scala/PosixLikeIO/examples/readWholeFile.scala

@@ -0,0 +1,25 @@
+package PosixLikeIO.examples
+
+import gears.async.{Async, given}
+import PosixLikeIO.PIOHelper
+
+import java.nio.ByteBuffer
+import java.nio.charset.StandardCharsets
+import java.nio.file.StandardOpenOption
+import scala.concurrent.ExecutionContext
+
+
+@main def readWholeFile(): Unit =
+  given ExecutionContext = ExecutionContext.global
+  Async.blocking:
+      PIOHelper.withFile("/home/julian/Desktop/x.txt", StandardOpenOption.READ): f =>
+        val b = ByteBuffer.allocate(1024)
+        val retCode = f.read(b).result.get
+        assert(retCode >= 0)
+        val s = StandardCharsets.UTF_8.decode(b.slice(0, retCode)).toString()
+        println("Read size with read(): " + retCode.toString())
+        println("Data: " + s)
+
+
+        println("Read with readString():")
+        println(Async.await(f.readString(1000)).get)