feat(graalvm): readable web streams implementation

Signed-off-by: Dario Valdespino <dvaldespino00@gmail.com>

Author: darvld

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

@@ -23,6 +23,7 @@ import elide.core.api.Symbolic.Unresolved
 import elide.runtime.core.DelicateElideApi
 import elide.runtime.gvm.api.Intrinsic
 import elide.runtime.gvm.internals.intrinsics.js.AbstractJsIntrinsic
+import elide.runtime.gvm.internals.intrinsics.js.webstreams.readable.ReadableStreamSource
 import elide.runtime.gvm.js.JsError
 import elide.runtime.gvm.js.JsSymbol.JsSymbols.asPublicJsSymbol
 import elide.runtime.intrinsics.GuestIntrinsic

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

@@ -0,0 +1,288 @@
+package elide.runtime.gvm.internals.intrinsics.js.webstreams.readable
+
+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.err.TypeError
+import elide.vm.annotations.Polyglot
+
+/** Shorthand for a completable promise used to define read queues. */
+private typealias DefaultReadRequest = CompletableJsPromise<ReadResult>
+
+/**
+ * Specialized implementation for a 'default' stream, i.e. a [ReadableStream] with a `ReadableStreamDefaultController`
+ * and `ReadableStreamDefaultReader`.
+ */
+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)
+
+  /** 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 {
+    /** Atomic controller state flag. */
+    private val controllerState = AtomicInteger(CONTROLLER_STARTED)
+
+    /** Desired total size for the inbound queue, used for backpressure control. */
+    val desiredSize: Double?
+      @Polyglot get() = when (streamState.get()) {
+        STREAM_CLOSED -> 0.0
+        STREAM_READABLE -> strategy.highWaterMark() - sourceChunksSize.get()
+        else -> null
+      }
+
+    init {
+      // setup the source
+      runCatching { source.start(this) }
+        .onFailure(::error)
+        .onSuccess {
+          controllerState.set(CONTROLLER_STARTED)
+          if (shouldPull()) pull()
+        }
+    }
+
+    /** Whether the controller should request new chunks from the underlying [source]. */
+    private fun shouldPull(): Boolean {
+      val state = controllerState.get()
+      return when {
+        state == CONTROLLER_CLOSING || streamState.get() != STREAM_READABLE -> false
+        state == CONTROLLER_UNINITIALIZED -> false
+        locked && readQueue.isNotEmpty() -> true
+        (desiredSize ?: 0.0) > 0.0 -> true
+        else -> false
+      }
+    }
+
+    /** Cancel the controller and the underlying source. Called by the stream during [ReadableDefaultStream.cancel]. */
+    fun cancel(reason: Any? = null): JsPromise<Unit> {
+      sourceChunks.clear()
+      sourceChunksSize.set(0.0)
+      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() {
+      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) {
+      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) {
+      // 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")
+      }
+
+      // consume queued reads if possible
+      readQueue.poll()?.let {
+        it.resolve(ReadResult(chunk, false))
+        return
+      }
+
+      try {
+        // enqueue the sized unread chunk
+        val size = strategy.size(chunk)
+        sourceChunks.add(SizedChunk(chunk, size))
+        sourceChunksSize.addAndGet(size)
+        if (shouldPull()) pull()
+      } catch (cause: Throwable) {
+        error(cause)
+      }
+    }
+
+    /** Pull from the underlying source if the backpressure controls allow it. */
+    fun pullIfNeeded() {
+      if (shouldPull()) pull()
+    }
+
+    /**
+     * Pull from the underlying source to obtain new chunks. If a pull is already in progress, a 'pull again' flag is
+     * set to trigger a new pull after the current one.
+     */
+    fun pull() {
+      // if pulling, flag for retry, the queued read will be fulfilled
+      if (controllerState.compareAndSet(CONTROLLER_PULLING, CONTROLLER_PULL_AGAIN)) return
+
+      // if idle, flag as pulling and pull from the source
+      if (controllerState.compareAndSet(CONTROLLER_STARTED, CONTROLLER_PULLING)) {
+        source.pull(this).then(
+          onFulfilled = {
+            if (controllerState.compareAndSet(CONTROLLER_PULL_AGAIN, CONTROLLER_PULLING)) pullIfNeeded()
+            else controllerState.compareAndSet(CONTROLLER_PULLING, CONTROLLER_STARTED)
+          },
+          onCatch = ::error,
+        )
+        return
+      }
+    }
+
+    /**
+     * Poll the inbound queue for available chunks, returning `null` if the queue is empty.
+     *
+     * If the controller is currently closing and the queue is emptied as a result of this operation, the controller
+     * stream will be closed.
+     */
+    fun poll(): SizedChunk? {
+      val chunk = sourceChunks.poll()?.also { sourceChunksSize.addAndGet(-it.size) }
+
+      // finish closing if we removed the last chunk from the queue
+      if (sourceChunks.isEmpty() && controllerState.compareAndSet(CONTROLLER_CLOSING, CONTROLLER_CLOSED))
+        this@ReadableDefaultStream.close()
+
+      return chunk
+    }
+  }
+
+
+  /**
+   * [`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()
+
+    init {
+      // early closure
+      when (streamState.get()) {
+        STREAM_CLOSED -> closed.resolve(Unit)
+        STREAM_ERRORED -> closed.reject(errorCause.get())
+        else -> Unit
+      }
+    }
+
+    /** 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")
+      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)) }
+
+          // enqueue read request and pull
+          CompletableJsPromise<ReadResult>().also { request ->
+            readQueue.add(request)
+            controller.pullIfNeeded()
+          }
+        }
+
+        STREAM_ERRORED -> JsPromise.rejected(TypeError.create(errorCause.get().toString()))
+        else -> JsPromise.resolved(ReadResult(null, true))
+      }
+    }
+
+    /**
+     * 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() {
+      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() {
+      this@ReadableDefaultStream.cancel()
+    }
+  }
+
+  /**
+   * Chunks are pushed by the source using the controller, and pulled by consumers using the reader; the queue can be
+   * skipped if there are pending reads when a new chunk is pushed.
+   */
+  private val sourceChunks: ConcurrentLinkedQueue<SizedChunk> = ConcurrentLinkedQueue()
+
+  /** Total size of the chunks in [sourceChunks], as calculated by the stream's queueing [strategy]. */
+  private val sourceChunksSize = AtomicDouble(0.0)
+
+  /**
+   * Reads are enqueued by consumers using the reader, and fulfilled by the source by pushing chunks with the
+   * controller; the queue can be skipped if there are queued chunks when a read is requested.
+   */
+  private val readQueue: ConcurrentLinkedQueue<DefaultReadRequest> = ConcurrentLinkedQueue()
+
+  /** Handle used by the [source] to push new chunks and control the stream. */
+  private val controller = ReadableStreamDefaultController()
+
+  /**
+   * 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>()
+
+  /** Stored cause for the stream's failure. */
+  private val errorCause = AtomicReference<Any>()
+
+  @get:Polyglot override val locked: Boolean get() = reader.get() != null
+
+  /** Close the stream, preventing new chunks from being enqueued by the source or read by consumers. */
+  private fun close() {
+    if (!streamState.compareAndSet(STREAM_READABLE, STREAM_CLOSED)) throw TypeError.create("Stream is not readable")
+    reader.getAndSet(null)?.closed?.resolve(Unit) ?: return
+
+    // close unfulfilled requests
+    while (true) readQueue.poll()?.resolve(ReadResult(null, true)) ?: break
+  }
+
+  /**
+   * Cancel the stream with an error; if locked, the reader will also be closed with the same cause. All pending read
+   * requests will be rejected with [reason].
+   */
+  private fun error(reason: Any? = null) {
+    if (!streamState.compareAndSet(STREAM_READABLE, STREAM_ERRORED)) throw TypeError.create("Stream is not readable")
+    errorCause.set(reason)
+
+    reader.get()?.closed?.reject(reason) ?: return
+    while (true) readQueue.poll()?.reject(reason) ?: break
+  }
+
+  @Polyglot override fun cancel(reason: Any?): JsPromise<Unit> {
+    return when (streamState.get()) {
+      STREAM_CLOSED -> JsPromise.resolved(Unit)
+      STREAM_ERRORED -> JsPromise.rejected(TypeError.create(errorCause.get().toString()))
+      else -> {
+        close()
+        controller.cancel(reason)
+      }
+    }
+  }
+
+  @Polyglot override fun getReader(options: Any?): ReadableStreamReader {
+    if (reader.getAndUpdate { it ?: ReadableStreamDefaultReader() } != null) throw TypeError.create("Stream is locked")
+    return reader.get()
+  }
+}