test(graalvm): webstream tests and polyglot proxies

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

Author: darvld

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

@@ -13,11 +13,14 @@
 package elide.runtime.gvm.internals.intrinsics.js.webstreams
 
 import org.graalvm.polyglot.Value
+import org.graalvm.polyglot.proxy.ProxyExecutable
 import java.util.*
 import java.util.concurrent.atomic.AtomicInteger
 import java.util.concurrent.atomic.AtomicReference
 import java.util.concurrent.locks.ReentrantLock
 import elide.runtime.exec.GuestExecutor
+import elide.runtime.gvm.internals.intrinsics.js.abort.AbortSignal
+import elide.runtime.interop.ReadOnlyProxyObject
 import elide.runtime.intrinsics.js.*
 import elide.runtime.intrinsics.js.ReadableStream.ReadResult
 import elide.runtime.intrinsics.js.err.TypeError
@@ -30,7 +33,7 @@ import elide.vm.annotations.Polyglot
  * methods used by other components. This contract allows readers and controllers to be decoupled from the internal
  * state of the stream itself.
  */
-internal abstract class ReadableStreamBase : ReadableStream {
+internal abstract class ReadableStreamBase : ReadableStream, ReadOnlyProxyObject {
   /** Executor used to schedule stream operations. */
   protected abstract val executor: GuestExecutor
 
@@ -106,7 +109,7 @@ internal abstract class ReadableStreamBase : ReadableStream {
 
     source.pull(controller).then(
       onFulfilled = {
-        if (sourceState.compareAndSet(SOURCE_PULL_AGAIN, SOURCE_PULLING)) maybePull()
+        if (sourceState.compareAndSet(SOURCE_PULL_AGAIN, SOURCE_READY)) maybePull()
         else sourceState.compareAndSet(SOURCE_PULLING, SOURCE_READY)
       },
       onCatch = ::error,
@@ -151,7 +154,7 @@ internal abstract class ReadableStreamBase : ReadableStream {
     if (destination.locked) throw TypeError.create("Destination is locked, cannot pipe")
 
     // unpack options
-    val signal = options?.getMember("signal")?.takeIf { it.isHostObject }?.asHostObject<AbortSignal>()
+    val signal = options?.getMember("signal")?.takeIf { it.isProxyObject }?.asProxyObject<AbortSignal>()
     val preventClose = options?.getMember("preventClose")?.takeIf { it.isBoolean }?.asBoolean() ?: false
     val preventAbort = options?.getMember("preventAbort")?.takeIf { it.isBoolean }?.asBoolean() ?: false
     val preventCancel = options?.getMember("preventCancel")?.takeIf { it.isBoolean }?.asBoolean() ?: false
@@ -187,6 +190,31 @@ internal abstract class ReadableStreamBase : ReadableStream {
     }
   }
 
+  override fun getMemberKeys(): Array<String> = MEMBERS
+  override fun getMember(key: String?): Any? = when (key) {
+    MEMBER_LOCKED -> locked
+    MEMBER_CANCEL -> ProxyExecutable { cancel(it.firstOrNull()) }
+    MEMBER_GET_READER -> ProxyExecutable { getReader(it.firstOrNull()) }
+    MEMBER_TEE -> ProxyExecutable { tee() }
+    MEMBER_PIPE_TO -> ProxyExecutable { args ->
+      val destination = args.firstOrNull() ?: throw TypeError.create("A pipe destination must be specified")
+      val destinationStream = runCatching { destination.asProxyObject<WritableStream>() }
+        .getOrElse { throw TypeError.create("The specified destination stream is not a writable stream", it) }
+
+      pipeTo(destinationStream, args.getOrNull(1))
+    }
+
+    MEMBER_PIPE_THROUGH -> ProxyExecutable { args ->
+      val transform = args.firstOrNull() ?: throw TypeError.create("A transform stream must be specified")
+      val transformStream = runCatching { transform.asProxyObject<TransformStream>() }
+        .getOrElse { throw TypeError.create("The specified destination stream is not a transform stream", it) }
+
+      pipeThrough(transformStream, args.getOrNull(1))
+    }
+
+    else -> null
+  }
+
   internal companion object {
     // stream state
     internal const val READABLE_STREAM_READABLE = 0
@@ -200,6 +228,22 @@ internal abstract class ReadableStreamBase : ReadableStream {
     internal const val SOURCE_PULL_AGAIN = 3
     internal const val SOURCE_CLOSING = 4
     internal const val SOURCE_CLOSED = 5
+
+    private const val MEMBER_LOCKED = "locked"
+    private const val MEMBER_CANCEL = "cancel"
+    private const val MEMBER_GET_READER = "getReader"
+    private const val MEMBER_TEE = "tee"
+    private const val MEMBER_PIPE_TO = "pipeTo"
+    private const val MEMBER_PIPE_THROUGH = "pipeThrough"
+
+    private val MEMBERS = arrayOf(
+      MEMBER_LOCKED,
+      MEMBER_CANCEL,
+      MEMBER_GET_READER,
+      MEMBER_TEE,
+      MEMBER_PIPE_TO,
+      MEMBER_PIPE_THROUGH,
+    )
   }
 }
 

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

@@ -12,6 +12,7 @@
  */
 package elide.runtime.gvm.internals.intrinsics.js.webstreams
 
+import elide.annotations.Singleton
 import elide.runtime.core.DelicateElideApi
 import elide.runtime.gvm.api.Intrinsic
 import elide.runtime.gvm.internals.intrinsics.js.AbstractJsIntrinsic
@@ -24,7 +25,7 @@ import elide.runtime.intrinsics.js.stream.ReadableStreamBYOBReader
 import elide.runtime.intrinsics.js.stream.ReadableStreamDefaultReader
 
 /** Implementation of readable streams (via the Web Streams standard). */
-@Intrinsic(global = "ReadableStream") internal class ReadableStreamIntrinsic : AbstractJsIntrinsic() {
+@Intrinsic(global = "ReadableStream") @Singleton internal class ReadableStreamIntrinsic : AbstractJsIntrinsic() {
   @OptIn(DelicateElideApi::class)
   override fun install(bindings: GuestIntrinsic.MutableIntrinsicBindings) {
     bindings[READABLE_STREAM_SYMBOL.asPublicJsSymbol()] = ReadableStream

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

@@ -18,6 +18,7 @@ import java.util.concurrent.atomic.AtomicReference
 import elide.runtime.exec.GuestExecutor
 import elide.runtime.gvm.internals.intrinsics.js.webstreams.ReadableStreamBase.Companion.READABLE_STREAM_ERRORED
 import elide.runtime.gvm.internals.intrinsics.js.webstreams.WritableDefaultStream.Companion.WRITABLE_STREAM_ERRORED
+import elide.runtime.interop.ReadOnlyProxyObject
 import elide.runtime.intrinsics.js.CompletableJsPromise
 import elide.runtime.intrinsics.js.JsPromise
 import elide.runtime.intrinsics.js.TransformStream
@@ -30,8 +31,7 @@ internal class TransformDefaultStream(
   executor: GuestExecutor,
   writableStrategy: QueueingStrategy = QueueingStrategy.DefaultReadStrategy,
   readableStrategy: QueueingStrategy = QueueingStrategy.DefaultReadStrategy,
-) : TransformStream {
-
+) : TransformStream, ReadOnlyProxyObject {
   @JvmInline private value class DelegatingSource(private val stream: TransformDefaultStream) : ReadableStreamSource {
     override fun pull(controller: ReadableStreamController): JsPromise<Unit> {
       check(stream.backpressure.get())
@@ -161,13 +161,13 @@ internal class TransformDefaultStream(
 
   private val controller = TransformStreamDefaultControllerToken(this)
 
-  @get:Polyglot override val readable: ReadableDefaultStream = ReadableDefaultStream(
-      source = DelegatingSource(this),
-      strategy = readableStrategy,
-      executor
+  @Polyglot override val readable: ReadableDefaultStream = ReadableDefaultStream(
+    source = DelegatingSource(this),
+    strategy = readableStrategy,
+    executor,
   )
 
-  @get:Polyglot override val writable: WritableDefaultStream = WritableDefaultStream(
+  @Polyglot override val writable: WritableDefaultStream = WritableDefaultStream(
     sink = DelegatingSink(this),
     strategy = writableStrategy,
   )
@@ -204,4 +204,17 @@ internal class TransformDefaultStream(
 
     return promise
   }
+
+  override fun getMemberKeys(): Array<String> = MEMBERS
+  override fun getMember(key: String?): Any? = when (key) {
+    MEMBER_READABLE -> readable
+    MEMBER_WRITABLE -> writable
+    else -> null
+  }
+
+  private companion object {
+    private const val MEMBER_READABLE = "readable"
+    private const val MEMBER_WRITABLE = "writable"
+    private val MEMBERS = arrayOf(MEMBER_READABLE, MEMBER_WRITABLE)
+  }
 }

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

@@ -12,6 +12,7 @@
  */
 package elide.runtime.gvm.internals.intrinsics.js.webstreams
 
+import elide.annotations.Singleton
 import elide.runtime.core.DelicateElideApi
 import elide.runtime.gvm.api.Intrinsic
 import elide.runtime.gvm.internals.intrinsics.js.AbstractJsIntrinsic
@@ -21,7 +22,7 @@ import elide.runtime.intrinsics.js.TransformStream
 import elide.runtime.intrinsics.js.WritableStream
 
 /** Implementation of transform streams (via the Web Streams standard). */
-@Intrinsic(global = "TransformStream") internal class TransformStreamIntrinsic : AbstractJsIntrinsic() {
+@Intrinsic(global = "TransformStream") @Singleton internal class TransformStreamIntrinsic : AbstractJsIntrinsic() {
   @OptIn(DelicateElideApi::class)
   override fun install(bindings: GuestIntrinsic.MutableIntrinsicBindings) {
     bindings[TRANSFORM_STREAM_SYMBOL.asPublicJsSymbol()] = TransformStream

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

@@ -14,13 +14,15 @@ package elide.runtime.gvm.internals.intrinsics.js.webstreams
 
 import com.google.common.util.concurrent.AtomicDouble
 import org.graalvm.polyglot.Value
+import org.graalvm.polyglot.proxy.ProxyExecutable
 import java.util.concurrent.ConcurrentLinkedQueue
 import java.util.concurrent.atomic.AtomicBoolean
 import java.util.concurrent.atomic.AtomicInteger
 import java.util.concurrent.atomic.AtomicReference
 import elide.runtime.gvm.internals.intrinsics.js.abort.AbortController
 import elide.runtime.gvm.internals.intrinsics.js.webstreams.WritableDefaultStream.QueueElement.Chunk
 import elide.runtime.gvm.internals.intrinsics.js.webstreams.WritableDefaultStream.QueueElement.CloseToken
+import elide.runtime.interop.ReadOnlyProxyObject
 import elide.runtime.intrinsics.js.AbortSignal
 import elide.runtime.intrinsics.js.CompletableJsPromise
 import elide.runtime.intrinsics.js.JsPromise
@@ -41,16 +43,31 @@ internal class WritableDefaultStream(
   private val sink: WritableStreamSink,
   /** Queueing strategy used to control backpressure. */
   private val strategy: QueueingStrategy = QueueingStrategy.DefaultWriteStrategy,
-) : WritableStream {
+) : WritableStream, ReadOnlyProxyObject {
   /** Inline wrapper for the stream providing the controller API by delegation. */
-  @JvmInline private value class ControllerToken(val stream: WritableDefaultStream) : WritableStreamDefaultController {
+  @JvmInline private value class ControllerToken(
+    val stream: WritableDefaultStream
+  ) : WritableStreamDefaultController, ReadOnlyProxyObject {
     override val signal: AbortSignal
       get() = stream.abortController.signal
 
     override fun error(e: Any?) {
       if (stream.state.get() != WRITABLE_STREAM_WRITABLE) return
       stream.startErroring(e)
     }
+
+    override fun getMemberKeys(): Array<String> = MEMBERS
+    override fun getMember(key: String?): Any? = when (key) {
+      MEMBER_SIGNAL -> signal
+      MEMBER_ERROR -> ProxyExecutable { error(it.firstOrNull()) }
+      else -> null
+    }
+
+    private companion object {
+      private const val MEMBER_SIGNAL = "signal"
+      private const val MEMBER_ERROR = "error"
+      private val MEMBERS = arrayOf(MEMBER_SIGNAL, MEMBER_ERROR)
+    }
   }
 
   /**
@@ -528,10 +545,25 @@ internal class WritableDefaultStream(
     else closeStream()
   }
 
+  override fun getMemberKeys(): Array<String> = MEMBERS
+  override fun getMember(key: String?): Any? = when (key) {
+    MEMBER_CLOSE -> ProxyExecutable { close() }
+    MEMBER_ABORT -> ProxyExecutable { abort(it.firstOrNull()) }
+    MEMBER_GET_WRITER -> ProxyExecutable { getWriter() }
+    MEMBER_LOCKED -> locked
+    else -> null
+  }
+
   internal companion object {
     internal const val WRITABLE_STREAM_WRITABLE: Int = 0
     internal const val WRITABLE_STREAM_CLOSED: Int = 1
     internal const val WRITABLE_STREAM_ERRORING: Int = 2
     internal const val WRITABLE_STREAM_ERRORED: Int = 3
+
+    private const val MEMBER_CLOSE = "close"
+    private const val MEMBER_ABORT = "abort"
+    private const val MEMBER_GET_WRITER = "getWriter"
+    private const val MEMBER_LOCKED = "locked"
+    private val MEMBERS = arrayOf(MEMBER_CLOSE, MEMBER_ABORT, MEMBER_GET_WRITER, MEMBER_LOCKED)
   }
 }

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

@@ -13,10 +13,12 @@
 package elide.runtime.gvm.internals.intrinsics.js.webstreams
 
 import org.graalvm.polyglot.Value
+import org.graalvm.polyglot.proxy.ProxyExecutable
 import java.util.concurrent.atomic.AtomicBoolean
 import java.util.concurrent.atomic.AtomicReference
 import elide.runtime.gvm.internals.intrinsics.js.webstreams.WritableDefaultStream.Companion.WRITABLE_STREAM_CLOSED
 import elide.runtime.gvm.internals.intrinsics.js.webstreams.WritableDefaultStream.Companion.WRITABLE_STREAM_ERRORED
+import elide.runtime.interop.ReadOnlyProxyObject
 import elide.runtime.intrinsics.js.CompletableJsPromise
 import elide.runtime.intrinsics.js.JsPromise
 import elide.runtime.intrinsics.js.err.TypeError
@@ -34,10 +36,10 @@ internal class WritableStreamDefaultWriterToken(
   private val detached = AtomicBoolean()
 
   /** A promise used to communicate changes in backpressure to data producers. */
-  private val mutableReadyPromise = AtomicReference<CompletableJsPromise<Unit>>()
+  private val mutableReadyPromise = AtomicReference(JsPromise<Unit>().also { it.resolve(Unit) })
 
   /** A promise used to signal the closing of the stream to data producers. */
-  private val mutableClosePromise = AtomicReference<CompletableJsPromise<Unit>>()
+  private val mutableClosePromise = AtomicReference(JsPromise<Unit>())
 
   @get:Polyglot override val ready: CompletableJsPromise<Unit> get() = mutableReadyPromise.get()
   @get:Polyglot override val closed: CompletableJsPromise<Unit> get() = mutableClosePromise.get()
@@ -56,8 +58,8 @@ internal class WritableStreamDefaultWriterToken(
   }
 
   /**
-   * Ensures that the writer's [ready] promise is rejected, by replacing it with a new rejected promise if it the
-   * current already.instance is already settled, or rejecting it if it is pending.
+   * Ensures that the writer's [ready] promise is rejected, by replacing it with a new rejected promise if the current
+   * ready promise is already settled, or rejecting it if it is pending.
    */
   internal fun ensureReadyPromiseRejected(reason: Any?) {
     val readyPromise = ready
@@ -66,8 +68,8 @@ internal class WritableStreamDefaultWriterToken(
   }
 
   /**
-   * Ensures that the writer's [closed] promise is rejected, by replacing it with a new rejected promise if it the
-   * current already.instance is already settled, or rejecting it if it is pending.
+   * Ensures that the writer's [closed] promise is rejected, by replacing it with a new rejected promise if the current
+   * promise is already settled, or rejecting it if it is pending.
    */
   internal fun ensureClosedPromiseRejected(reason: Any?) {
     val closePromise = ready
@@ -106,4 +108,36 @@ internal class WritableStreamDefaultWriterToken(
     stream.closeQueuedOrInflight -> JsPromise.rejected(TypeError.create("Stream is already closing"))
     else -> stream.closeStream()
   }
+
+  override fun getMemberKeys(): Array<String> = MEMBERS
+  override fun getMember(key: String?): Any? = when (key) {
+    MEMBER_READY -> ready
+    MEMBER_CLOSED -> closed
+    MEMBER_DESIRED_SIZE -> desiredSize
+    MEMBER_CLOSE -> ProxyExecutable { close() }
+    MEMBER_ABORT -> ProxyExecutable { abort(it.firstOrNull()) }
+    MEMBER_WRITE -> ProxyExecutable { write(it.firstOrNull() ?: throw TypeError.create("Chunk must not be null")) }
+    MEMBER_RELEASE_LOCK -> ProxyExecutable { releaseLock() }
+    else -> null
+  }
+
+  private companion object {
+    private const val MEMBER_WRITE = "write"
+    private const val MEMBER_RELEASE_LOCK = "releaseLock"
+    private const val MEMBER_ABORT = "abort"
+    private const val MEMBER_CLOSE = "close"
+    private const val MEMBER_READY = "ready"
+    private const val MEMBER_CLOSED = "closed"
+    private const val MEMBER_DESIRED_SIZE = "desiredSize"
+
+    private val MEMBERS = arrayOf(
+      MEMBER_READY,
+      MEMBER_CLOSED,
+      MEMBER_DESIRED_SIZE,
+      MEMBER_CLOSE,
+      MEMBER_ABORT,
+      MEMBER_WRITE,
+      MEMBER_RELEASE_LOCK,
+    )
+  }
 }

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

@@ -12,6 +12,7 @@
  */
 package elide.runtime.gvm.internals.intrinsics.js.webstreams
 
+import elide.annotations.Singleton
 import elide.runtime.core.DelicateElideApi
 import elide.runtime.gvm.api.Intrinsic
 import elide.runtime.gvm.internals.intrinsics.js.AbstractJsIntrinsic
@@ -21,7 +22,7 @@ import elide.runtime.intrinsics.js.WritableStream
 import elide.runtime.intrinsics.js.stream.WritableStreamDefaultWriter
 
 /** Implementation of writable streams (via the Web Streams standard). */
-@Intrinsic(global = "WritableStream") internal class WritableStreamIntrinsic : AbstractJsIntrinsic() {
+@Intrinsic(global = "WritableStream") @Singleton internal class WritableStreamIntrinsic : AbstractJsIntrinsic() {
   @OptIn(DelicateElideApi::class)
   override fun install(bindings: GuestIntrinsic.MutableIntrinsicBindings) {
     bindings[WRITABLE_STREAM_SYMBOL.asPublicJsSymbol()] = WritableStream

packages/graalvm/src/main/kotlin/elide/runtime/intrinsics/js/JsPromise.kt

@@ -144,7 +144,7 @@ public interface JsPromise<out T> : ProxyObject {
       unwrapFulfilled: (Value?) -> T,
       unwrapRejected: (Value?) -> Any? = { it },
     ): JsPromise<T>? {
-      return if (value.isHostObject) runCatching { value.asHostObject<JsPromise<T>>() }.getOrNull()
+      return if (value.isProxyObject) runCatching { value.asProxyObject<JsPromise<T>>() }.getOrNull()
       else if (!value.canInvokeMember(THEN_SYMBOL)) null
       else object : JsPromise<T> {
         override val isDone: Boolean get() = false

packages/graalvm/src/main/kotlin/elide/runtime/intrinsics/js/ReadableStream.kt

@@ -16,12 +16,14 @@ import org.graalvm.polyglot.HostAccess
 import org.graalvm.polyglot.Value
 import org.graalvm.polyglot.proxy.ProxyInstantiable
 import org.graalvm.polyglot.proxy.ProxyIterable
+import org.graalvm.polyglot.proxy.ProxyObject
 import java.io.InputStream
 import java.io.Reader
 import java.nio.ByteBuffer
 import elide.annotations.API
 import elide.runtime.exec.GuestExecution
 import elide.runtime.gvm.internals.intrinsics.js.webstreams.*
+import elide.runtime.interop.ReadOnlyProxyObject
 import elide.runtime.intrinsics.js.ReadableStream.ReaderMode.BYOB
 import elide.runtime.intrinsics.js.ReadableStream.ReaderMode.Default
 import elide.runtime.intrinsics.js.stream.*
@@ -34,7 +36,7 @@ import elide.vm.annotations.Polyglot
  * Readable streams implement streams of arbitrary data which can be consumed by an interested developer, and form part
  * of the wider Web Streams API.
  */
-@API @HostAccess.Implementable public interface ReadableStream : Stream, ProxyIterable {
+@API @HostAccess.Implementable public interface ReadableStream : Stream, ProxyIterable, ProxyObject {
   /**
    * Encapsulates the result of a read; [done] indicates whether the [value] is the final value that will be available
    * from the source.