feat(graalvm): cleanup webstreams api surface

Signed-off-by: Dario Valdespino <[email protected]>

Author: darvld

packages/graalvm/src/main/kotlin/elide/runtime/gvm/internals/intrinsics/js/JsPromiseImpl.kt

@@ -64,7 +64,7 @@ internal class JsPromiseImpl<T> private constructor(token: Token) : CompletableJ
   private val onReject: MutableList<(Any?) -> Unit> = mutableListOf()
 
   /** The promise is considered as 'closed' once the [token] is no longer [Pending]. */
-  override val isClosed: Boolean get() = token.get() != Pending
+  override val isDone: Boolean get() = token.get() != Pending
 
   @Suppress("UNCHECKED_CAST")
   override fun then(onFulfilled: (T) -> Unit, onCatch: ((Any?) -> Unit)?): JsPromise<T> = apply {

packages/graalvm/src/main/kotlin/elide/runtime/gvm/internals/intrinsics/js/webstreams/AbstractReadableStream.kt

@@ -0,0 +1,69 @@
+package elide.runtime.gvm.internals.intrinsics.js.webstreams
+
+import java.util.concurrent.atomic.AtomicInteger
+import elide.runtime.intrinsics.js.ReadableStream
+import elide.runtime.intrinsics.js.stream.QueueingStrategy
+import elide.runtime.intrinsics.js.stream.ReadableStreamSource
+
+
+/**
+ * Base abstract class for the native `ReadableStream` standard implementation.
+ *
+ * Unlike in the WHATWG reference, the default and BYOB stream components are separated into specialized classes to
+ * remove the need for casting and type checks on every operation.
+ *
+ * This class defines the shared public API for streams, according to the
+ * [spec](https://streams.spec.whatwg.org/#rs-class-definition).
+ */
+public sealed class AbstractReadableStream : ReadableStream {
+  /** Thread-safe state flag for the stream. */
+  protected val streamState: AtomicInteger = AtomicInteger(STREAM_READABLE)
+
+  /** The source this stream pulls data from. */
+  protected abstract val source: ReadableStreamSource
+
+  /** The queuing strategy used by the stream's controller to pull data from the [source]. */
+  protected abstract val strategy: QueueingStrategy
+
+  public companion object {
+    // Stream state values
+
+    /** The stream is readable and fully functional. */
+    internal const val STREAM_READABLE = 0
+
+    /** The stream has been closed normally without an exception. */
+    internal const val STREAM_CLOSED = 1
+
+    /** The stream has been closed exceptionally. */
+    internal const val STREAM_ERRORED = 2
+
+    // Controller state values
+
+    /** The underlying source has not been initialized yet.*/
+    internal const val CONTROLLER_UNINITIALIZED = 0
+
+    /** The controller is ready, the source has completed startup. */
+    internal const val CONTROLLER_STARTED = 1
+
+    /** The controller is pulling a new value from the source. */
+    internal const val CONTROLLER_PULLING = 2
+
+    /** The controller is pulling and will pull again when the current pull completes. */
+    internal const val CONTROLLER_PULL_AGAIN = 3
+
+    /** The controller is being closed, but must finish other operations before finishing. */
+    internal const val CONTROLLER_CLOSING = 4
+
+    /** The controller is closed and can no longer be used. */
+    internal const val CONTROLLER_CLOSED = 5
+
+    public fun from(
+      source: ReadableStreamSource,
+      strategy: QueueingStrategy = QueueingStrategy.Default
+    ): ReadableStream = when (source.type) {
+      ReadableStream.Type.Default -> ReadableDefaultStream(source, strategy)
+      ReadableStream.Type.BYOB -> TODO("BYOB streams are not yet supported")
+    }
+  }
+}
+

…treams/readable/ReadableDefaultStream.kt …s/js/webstreams/ReadableDefaultStream.ktpackages/graalvm/src/main/kotlin/elide/runtime/gvm/internals/intrinsics/js/webstreams/readable/ReadableDefaultStream.kt renamed to packages/graalvm/src/main/kotlin/elide/runtime/gvm/internals/intrinsics/js/webstreams/ReadableDefaultStream.kt packages/graalvm/src/main/kotlin/elide/runtime/gvm/internals/intrinsics/js/webstreams/readable/ReadableDefaultStream.kt renamed to packages/graalvm/src/main/kotlin/elide/runtime/gvm/internals/intrinsics/js/webstreams/ReadableDefaultStream.kt

@@ -1,13 +1,13 @@
-package elide.runtime.gvm.internals.intrinsics.js.webstreams.readable
+package elide.runtime.gvm.internals.intrinsics.js.webstreams
 
 import com.google.common.util.concurrent.AtomicDouble
 import java.util.concurrent.ConcurrentLinkedQueue
 import java.util.concurrent.atomic.AtomicInteger
 import java.util.concurrent.atomic.AtomicReference
-import elide.runtime.gvm.internals.intrinsics.js.webstreams.readable.ReadableDefaultStream.ReadResult
-import elide.runtime.intrinsics.js.CompletableJsPromise
-import elide.runtime.intrinsics.js.JsPromise
+import elide.runtime.intrinsics.js.*
+import elide.runtime.intrinsics.js.ReadableStream.ReadResult
 import elide.runtime.intrinsics.js.err.TypeError
+import elide.runtime.intrinsics.js.stream.*
 import elide.vm.annotations.Polyglot
 
 /** Shorthand for a completable promise used to define read queues. */
@@ -19,39 +19,34 @@ private typealias DefaultReadRequest = CompletableJsPromise<ReadResult>
  */
 internal class ReadableDefaultStream internal constructor(
   override val source: ReadableStreamSource,
-  override val strategy: ReadableStreamQueuingStrategy,
-) : ReadableStream() {
-  /**
-   * Encapsulates the result of a read; [done] indicates whether the [chunk] is the final value that will be available
-   * from the source.
-   */
-  data class ReadResult(val chunk: Any?, val done: Boolean)
-
+  override val strategy: QueueingStrategy = QueueingStrategy.Default,
+) : AbstractReadableStream() {
   /** An arbitrary [chunk] value with an attached [size], as calculated by the stream's queueing strategy. */
   data class SizedChunk(val chunk: Any?, val size: Double)
 
   /**
    * [`ReadableStreamDefaultController`](https://streams.spec.whatwg.org/#rs-default-controller-class) spec
    * implementation, including all the exposed fields and methods in the spec.
    */
-  private inner class ReadableStreamDefaultController : ReadableStreamController {
+  private inner class ControllerImpl : ReadableStreamDefaultController {
     /** Atomic controller state flag. */
     private val controllerState = AtomicInteger(CONTROLLER_STARTED)
 
-    /** Desired total size for the inbound queue, used for backpressure control. */
-    val desiredSize: Double?
+    override val desiredSize: Double?
       @Polyglot get() = when (streamState.get()) {
         STREAM_CLOSED -> 0.0
         STREAM_READABLE -> strategy.highWaterMark() - sourceChunksSize.get()
         else -> null
       }
 
-    init {
-      // setup the source
+    fun setup() {
+      // set up the source
       runCatching { source.start(this) }
-        .onFailure(::error)
+        .onFailure {
+          error(it)
+        }
         .onSuccess {
-          controllerState.set(CONTROLLER_STARTED)
+          controllerState.compareAndSet(CONTROLLER_UNINITIALIZED, CONTROLLER_STARTED)
           if (shouldPull()) pull()
         }
     }
@@ -75,32 +70,20 @@ internal class ReadableDefaultStream internal constructor(
       return source.cancel(reason)
     }
 
-    /**
-     * Close the controller and the associated stream. If there are undelivered chunks, the stream will not be closed
-     * until they are claimed.
-     */
-    @Polyglot fun close() {
+    @Polyglot override fun close() {
       if (controllerState.getAndSet(CONTROLLER_CLOSING) == CONTROLLER_CLOSING) return
       // don't close the stream if there are undelivered elements
       if (sourceChunks.isEmpty()) this@ReadableDefaultStream.close()
     }
 
-    /**
-     * Shut down the controller and associated stream with the given error [reason]. Unlike [close], this method does
-     * not wait for undelivered elements to be claimed, all unread data will be lost.
-     */
-    @Polyglot fun error(reason: Any? = null) {
+    @Polyglot override fun error(reason: Any?) {
       if (streamState.get() != STREAM_READABLE) return
       sourceChunks.clear()
       sourceChunksSize.set(0.0)
       this@ReadableDefaultStream.error(reason)
     }
 
-    /**
-     * Enqueue a new chunk, making it available immediately for readers. If any pending read requests are found, the
-     * chunk will be delivered directly without using the queue.
-     */
-    @Polyglot fun enqueue(chunk: Any? = null) {
+    @Polyglot override fun enqueue(chunk: Any?) {
       // check the controller isn't closing or closed
       if (controllerState.get() == CONTROLLER_CLOSING || streamState.get() != STREAM_READABLE) {
         throw TypeError.create("Controller is closing or stream is not readable")
@@ -166,16 +149,14 @@ internal class ReadableDefaultStream internal constructor(
     }
   }
 
-
   /**
    * [`ReadableStreamDefaultReader`](https://streams.spec.whatwg.org/#default-reader-class) spec
    * implementation, including all the exposed fields and methods in the spec.
    */
-  private inner class ReadableStreamDefaultReader : ReadableStreamReader {
-    /** A promised that completes when the reader is closed, and rejects when the reader errors. */
-    @Polyglot val closed: CompletableJsPromise<Unit> = CompletableJsPromise()
+  private inner class ReadableStreamDefaultReaderImpl : ReadableStreamDefaultReader {
+    @Polyglot override val closed: CompletableJsPromise<Unit> = JsPromise()
 
-    init {
+    fun setup() {
       // early closure
       when (streamState.get()) {
         STREAM_CLOSED -> closed.resolve(Unit)
@@ -184,16 +165,17 @@ internal class ReadableDefaultStream internal constructor(
       }
     }
 
-    /** Read a chunk from the stream, returning a promise that is fulfilled with the result. */
-    @Polyglot fun read(): JsPromise<ReadResult> {
-      if (closed.isDone) throw TypeError.create("Reader has been closed")
+    @Polyglot override fun read(): JsPromise<ReadResult> {
+      if (reader.get() !== this) throw TypeError.create("Reader is detached")
       return when (streamState.get()) {
         STREAM_READABLE -> {
           // consume queued chunks if available; if the stream was closed after polling, mark as 'done'
-          controller.poll()?.let { return JsPromise.resolved(ReadResult(it, streamState.get() != STREAM_READABLE)) }
+          controller.poll()?.let {
+            return JsPromise.resolved(ReadResult(it.chunk, streamState.get() != STREAM_READABLE))
+          }
 
           // enqueue read request and pull
-          CompletableJsPromise<ReadResult>().also { request ->
+          JsPromise<ReadResult>().also { request ->
             readQueue.add(request)
             controller.pullIfNeeded()
           }
@@ -204,18 +186,13 @@ internal class ReadableDefaultStream internal constructor(
       }
     }
 
-    /**
-     * Release this reader's lock on the stream, allowing a new reader to be acquired and invalidating this instance.
-     * Pending reads will still complete normally.
-     */
-    @Polyglot fun releaseLock() {
+    @Polyglot override fun releaseLock() {
       if (closed.isDone) throw TypeError.create("Reader has already been released")
       closed.reject(TypeError.create("Reader lock was released"))
       reader.set(null)
     }
 
-    /** Cancel the stream for this reader. */
-    @Polyglot fun cancel() {
+    @Polyglot override fun cancel() {
       this@ReadableDefaultStream.cancel()
     }
   }
@@ -235,17 +212,17 @@ internal class ReadableDefaultStream internal constructor(
    */
   private val readQueue: ConcurrentLinkedQueue<DefaultReadRequest> = ConcurrentLinkedQueue()
 
-  /** Handle used by the [source] to push new chunks and control the stream. */
-  private val controller = ReadableStreamDefaultController()
+  /** Stored cause for the stream's failure. */
+  private val errorCause = AtomicReference<Any>()
 
   /**
    * Handle used by consumers to read chunks; if not `null`, the stream is considered 'locked' and must be released
    * before a new reader can be acquired.
    */
-  private val reader = AtomicReference<ReadableStreamDefaultReader>()
+  private val reader = AtomicReference<ReadableStreamDefaultReaderImpl>()
 
-  /** Stored cause for the stream's failure. */
-  private val errorCause = AtomicReference<Any>()
+  /** Handle used by the [source] to push new chunks and control the stream. */
+  private val controller = ControllerImpl().also { it.setup() }
 
   @get:Polyglot override val locked: Boolean get() = reader.get() != null
 
@@ -282,7 +259,13 @@ internal class ReadableDefaultStream internal constructor(
   }
 
   @Polyglot override fun getReader(options: Any?): ReadableStreamReader {
-    if (reader.getAndUpdate { it ?: ReadableStreamDefaultReader() } != null) throw TypeError.create("Stream is locked")
+    if (reader.getAndUpdate { current ->
+        current ?: ReadableStreamDefaultReaderImpl().also { it.setup() }
+      } != null) throw TypeError.create("Stream is locked")
     return reader.get()
   }
+
+  override fun pipeThrough(transform: TransformStream, options: Any?): ReadableStream = TODO("Not yet implemented")
+  override fun pipeTo(destination: WritableStream, options: Any?): JsPromise<Unit> = TODO("Not yet implemented")
+  override fun tee(): Array<ReadableStream> = TODO("Not yet implemented")
 }

packages/graalvm/src/main/kotlin/elide/runtime/gvm/internals/intrinsics/js/webstreams/ReadableStreamIntrinsic.kt

@@ -12,64 +12,12 @@
  */
 package elide.runtime.gvm.internals.intrinsics.js.webstreams
 
-import java.io.InputStream
-import java.io.Reader
-import java.nio.ByteBuffer
 import elide.runtime.gvm.api.Intrinsic
 import elide.runtime.gvm.internals.intrinsics.js.AbstractJsIntrinsic
 import elide.runtime.intrinsics.GuestIntrinsic
-import elide.runtime.intrinsics.js.ReadableStream
 
 /** Implementation of readable streams (via the Web Streams standard). */
 @Intrinsic(global = "ReadableStream") internal class ReadableStreamIntrinsic : AbstractJsIntrinsic() {
-  /**
-   * TBD.
-   */
-  internal class ReadableStreamImpl : ReadableStream {
-
-  }
-
-  /**
-   * TBD.
-   */
-  internal companion object Factory : ReadableStream.Factory<ReadableStreamImpl> {
-    /**
-     * TBD.
-     */
-    override fun empty(): ReadableStreamImpl {
-      TODO("Not yet implemented")
-    }
-
-    /**
-     * TBD.
-     */
-    override fun wrap(input: InputStream): ReadableStreamImpl {
-      TODO("Not yet implemented")
-    }
-
-    /**
-     * TBD.
-     */
-    override fun wrap(reader: Reader): ReadableStreamImpl {
-      TODO("Not yet implemented")
-    }
-
-    /**
-     * TBD.
-     */
-    override fun wrap(bytes: ByteArray): ReadableStreamImpl {
-      // @TODO
-      return ReadableStreamImpl()
-    }
-
-    /**
-     * TBD.
-     */
-    override fun wrap(buffer: ByteBuffer): ReadableStreamImpl {
-      TODO("Not yet implemented")
-    }
-  }
-
   override fun install(bindings: GuestIntrinsic.MutableIntrinsicBindings) {
     // not yet implemented
   }