From 0f15d847d0bd8bff23f052f89e90e933d890865e Mon Sep 17 00:00:00 2001 From: duobei Date: Thu, 12 Mar 2026 11:19:27 +0800 Subject: [PATCH 1/8] feat: add wasm-gc support for timer, event loop, and time MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add wasm-gc implementations that delegate to the JS host via WASM imports: - timer.wasm-gc.mbt: Timer using host-provided setTimeout/clearTimeout - event_loop.wasm-gc.mbt: cooperative scheduling via setTimeout(0) - time.mbt: ms_since_epoch via host-provided Date.now() - wasm-gc-imports.js: JS import object for the host to provide This enables @async.sleep(), @async.Timer, and @async.now() on the wasm-gc target when running in a JavaScript host (browser or Node.js). The WASM import convention uses "moonbitlang_async_timer" and "moonbitlang_async_time" as module names. Partial fix for #233 — timer/event-loop/time only; fs/process/socket remain unimplemented on wasm-gc. --- .../event_loop/event_loop.wasm-gc.mbt | 36 +++++++++++++++++++ src/internal/event_loop/moon.pkg | 5 ++- src/internal/event_loop/timer.wasm-gc.mbt | 36 +++++++++++++++++++ src/internal/event_loop/wasm-gc-imports.js | 14 ++++++++ src/internal/time/time.mbt | 29 ++++++++++++++- 5 files changed, 118 insertions(+), 2 deletions(-) create mode 100644 src/internal/event_loop/event_loop.wasm-gc.mbt create mode 100644 src/internal/event_loop/timer.wasm-gc.mbt create mode 100644 src/internal/event_loop/wasm-gc-imports.js diff --git a/src/internal/event_loop/event_loop.wasm-gc.mbt b/src/internal/event_loop/event_loop.wasm-gc.mbt new file mode 100644 index 000000000..703527ab7 --- /dev/null +++ b/src/internal/event_loop/event_loop.wasm-gc.mbt @@ -0,0 +1,36 @@ +// Copyright 2025 International Digital Economy Academy +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +///| +let _ignore_unused_import : Unit = { + ignore(@c_buffer.unimplemented) + ignore(@fd_util.unimplemented) + ignore(@os_string.unimplemented) + ignore(@os_error.unimplemented) + ignore(@time.ms_since_epoch) +} + +///| +pub fn reschedule() -> Unit { + guard !@coroutine.no_more_work() else { } + @coroutine.reschedule() + // Instead of looping blockingly until there is no ready task, + // we only perform one round of scheduling here + // (i.e., only those tasks already ready when `reschedule` is called are run). + // Remaining tasks are delayed until the next event loop iteration, + // so that those blocking jobs got a chance to execute instead of starving. + if @coroutine.has_immediately_ready_task() { + ignore(wasm_set_timeout(0, reschedule)) + } +} diff --git a/src/internal/event_loop/moon.pkg b/src/internal/event_loop/moon.pkg index a3745c747..f9f2bc9a5 100644 --- a/src/internal/event_loop/moon.pkg +++ b/src/internal/event_loop/moon.pkg @@ -36,6 +36,7 @@ options( "cancel_before_submit_test.mbt": [ "native" ], "event_loop.js.mbt": [ "js" ], "event_loop.mbt": [ "native" ], + "event_loop.wasm-gc.mbt": [ "wasm-gc" ], "fs.mbt": [ "native" ], "io.mbt": [ "native" ], "io_unix.mbt": [ "native" ], @@ -49,8 +50,10 @@ options( "process_windows.mbt": [ "native" ], "stdio.mbt": [ "native" ], "thread_pool.mbt": [ "native" ], + "timer.js.mbt": [ "js" ], "timer.mbt": [ "native" ], - "unimplemented.mbt": [ "wasm", "wasm-gc" ], + "timer.wasm-gc.mbt": [ "wasm-gc" ], + "unimplemented.mbt": [ "wasm" ], "worker_wbtest.mbt": [ "native", "js" ], }, ) diff --git a/src/internal/event_loop/timer.wasm-gc.mbt b/src/internal/event_loop/timer.wasm-gc.mbt new file mode 100644 index 000000000..73895dce1 --- /dev/null +++ b/src/internal/event_loop/timer.wasm-gc.mbt @@ -0,0 +1,36 @@ +// Copyright 2025 International Digital Economy Academy +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +///| +#external +type Timer + +///| +fn wasm_set_timeout(duration : Int, f : () -> Unit) -> Timer = "moonbitlang_async_timer" "set_timeout" + +///| +fn wasm_clear_timeout(timer : Timer) = "moonbitlang_async_timer" "clear_timeout" + +///| +pub fn Timer::new(duration : Int, f : () -> Unit) -> Timer { + wasm_set_timeout(duration, () => { + f() + reschedule() + }) +} + +///| +pub fn Timer::cancel(self : Timer) -> Unit { + wasm_clear_timeout(self) +} diff --git a/src/internal/event_loop/wasm-gc-imports.js b/src/internal/event_loop/wasm-gc-imports.js new file mode 100644 index 000000000..a2405c918 --- /dev/null +++ b/src/internal/event_loop/wasm-gc-imports.js @@ -0,0 +1,14 @@ +// Import object for moonbitlang/async wasm-gc timer and event loop. +// When loading the compiled wasm-gc module, merge this into the import object: +// +// const imports = { ...otherImports, ...asyncImports }; +// const { instance } = await WebAssembly.instantiateStreaming(fetch("module.wasm"), imports); + +export const moonbitlang_async_timer = { + set_timeout: (duration, f) => setTimeout(f, duration), + clear_timeout: (timer) => clearTimeout(timer), +}; + +export const moonbitlang_async_time = { + date_now: () => Date.now(), +}; diff --git a/src/internal/time/time.mbt b/src/internal/time/time.mbt index 74a4d4da3..652dd110b 100644 --- a/src/internal/time/time.mbt +++ b/src/internal/time/time.mbt @@ -62,7 +62,34 @@ pub fn ms_since_epoch() -> Int64 { } ///| -#cfg(any(target="wasm", target="wasm-gc")) +#cfg(target="wasm-gc") +fn wasm_gc_date_now() -> Double = "moonbitlang_async_time" "date_now" + +///| +/// Get the current wall-clock time in milliseconds. +/// +/// This is the time source used by the async runtime. It is intended for +/// computing elapsed time by subtraction. The value can jump forwards or +/// backwards if the system clock is adjusted, so do not assume monotonicity. +/// +/// Platform notes: +/// - Unix/macOS: `gettimeofday()` (Unix epoch) +/// - Windows: `GetSystemTimeAsFileTime()` (FILETIME epoch, 1601) +/// - JavaScript/wasm-gc: `Date.now()` (Unix epoch) +/// +/// # Example +/// ```mbt check +/// test { +/// let _ : Int64 = ms_since_epoch() +/// } +/// ``` +#cfg(target="wasm-gc") +pub fn ms_since_epoch() -> Int64 { + wasm_gc_date_now().to_int64() +} + +///| +#cfg(target="wasm") pub fn ms_since_epoch() -> Int64 { abort("unimplemented") } From a7a6d9ada9b1490e9e5d82403674d9ee534f4be2 Mon Sep 17 00:00:00 2001 From: duobei Date: Thu, 12 Mar 2026 12:00:41 +0800 Subject: [PATCH 2/8] docs: fix wasm-gc-imports.js header comment - Mention time module alongside timer and event loop - Show correct import usage with named exports --- src/internal/event_loop/wasm-gc-imports.js | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/src/internal/event_loop/wasm-gc-imports.js b/src/internal/event_loop/wasm-gc-imports.js index a2405c918..391300a29 100644 --- a/src/internal/event_loop/wasm-gc-imports.js +++ b/src/internal/event_loop/wasm-gc-imports.js @@ -1,7 +1,8 @@ -// Import object for moonbitlang/async wasm-gc timer and event loop. -// When loading the compiled wasm-gc module, merge this into the import object: +// Import objects for moonbitlang/async wasm-gc timer, event loop, and time modules. +// When loading the compiled wasm-gc module, merge these into the import object: // -// const imports = { ...otherImports, ...asyncImports }; +// import { moonbitlang_async_timer, moonbitlang_async_time } from "./wasm-gc-imports.js"; +// const imports = { ...otherImports, moonbitlang_async_timer, moonbitlang_async_time }; // const { instance } = await WebAssembly.instantiateStreaming(fetch("module.wasm"), imports); export const moonbitlang_async_timer = { From c4416f890dd799df5cac6f46c6fe515d8194211d Mon Sep 17 00:00:00 2001 From: duobei Date: Thu, 12 Mar 2026 11:55:40 +0800 Subject: [PATCH 3/8] feat: add wasm-gc support for Promise bridging, ReadableStream, and HTTP client Port js_async and http packages to wasm-gc target using WASM imports instead of extern "js". Key design decisions: - Deferred pattern replaces new Promise(executor) since wasm-gc cannot directly call JS functions received as parameters - Callback-based byte copying for Bytes/Uint8Array boundary crossing (one FFI call per chunk, JS loops internally) - RawExternRef + %identity for cross-package #external types in WASM import signatures (compiler restriction workaround) - JsHeaders::for_each replaces to_array (wasm-gc Array != JS Array) --- src/http/client.wasm-gc.mbt | 298 +++++++++++++++++++++++ src/http/moon.pkg | 3 +- src/http/wasm-gc-imports.js | 27 ++ src/js_async/js_async.wasm-gc.mbt | 216 ++++++++++++++++ src/js_async/moon.pkg | 4 +- src/js_async/readable_stream.wasm-gc.mbt | 179 ++++++++++++++ src/js_async/wasm-gc-imports.js | 53 ++++ 7 files changed, 778 insertions(+), 2 deletions(-) create mode 100644 src/http/client.wasm-gc.mbt create mode 100644 src/http/wasm-gc-imports.js create mode 100644 src/js_async/js_async.wasm-gc.mbt create mode 100644 src/js_async/readable_stream.wasm-gc.mbt create mode 100644 src/js_async/wasm-gc-imports.js diff --git a/src/http/client.wasm-gc.mbt b/src/http/client.wasm-gc.mbt new file mode 100644 index 000000000..c2de8c674 --- /dev/null +++ b/src/http/client.wasm-gc.mbt @@ -0,0 +1,298 @@ +// Copyright 2025 International Digital Economy Academy +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +///| +#external +priv type JsHeaders + +///| +fn JsHeaders::new() -> JsHeaders = "moonbitlang_async_http" "new_headers" + +///| +fn JsHeaders::append(headers : JsHeaders, name : String, value : String) = "moonbitlang_async_http" "headers_append" + +///| +fn JsHeaders::for_each( + headers : JsHeaders, + callback : (String, String) -> Unit, +) -> Unit = "moonbitlang_async_http" "headers_for_each" + +///| +#external +priv type JsResponse + +///| +fn JsResponse::status(response : JsResponse) -> Int = "moonbitlang_async_http" "response_status" + +///| +fn JsResponse::status_text(response : JsResponse) -> String = "moonbitlang_async_http" "response_status_text" + +///| +fn JsResponse::headers(response : JsResponse) -> JsHeaders = "moonbitlang_async_http" "response_headers" + +///| +/// Local externref aliases for cross-package types used in WASM import +/// signatures. The wasm-gc backend requires WASM import parameters and +/// return types to be defined in the same compilation unit. +#external +priv type RawExternRef + +///| +fn[X, Y] raw_cast(x : X) -> Y = "%identity" + +///| +fn wasm_response_body(response : JsResponse) -> RawExternRef = "moonbitlang_async_http" "response_body" + +///| +fn JsResponse::body(response : JsResponse) -> @js_async.JsReadableStream { + raw_cast(wasm_response_body(response)) +} + +///| +priv struct OngoingRequest { + body_writer : @io.PipeWrite + abort_controller : @js_async.AbortController + response : @js_async.Promise[JsResponse] +} + +///| +struct Client { + host : String + port : Int + protocol : Protocol + headers : Map[String, String] + mut request : OngoingRequest? + mut response_body : @js_async.ReadableStream? +} + +///| +pub fn Client::close(self : Client) -> Unit { + if self.request is Some(request) { + request.body_writer.close() + self.request = None + } + if self.response_body is Some(response_body) { + response_body.close() + } +} + +///| +fn Client::connect( + host : String, + headers? : Map[String, String] = {}, + protocol? : Protocol = Https, + port? : Int = protocol.default_port(), + proxy? : Client, +) -> Client { + ignore(proxy) + { host, headers, protocol, port, request: None, response_body: None } +} + +///| +/// Create a new HTTP client by connecting to a remote host. +/// Host should be specified via `protocol://host[:port]`, +/// where `protocol` is one of `http` or `https`. +/// +/// `headers` can be used to specify persistent headers for the client, +/// i.e. all requests made from this client will share these headers. +/// The ownership of `headers` will be transferred to the new client, +/// so `headers` should not be used by the caller later. +/// The headers mentioned in +/// https://developer.mozilla.org/en-US/docs/Glossary/Forbidden_request_header +/// must not be set in `headers`. +/// +/// The HTTP client make requests using native fetch API. +/// +/// The `proxy` argument is not supported on wasm-gc backend and has no effect. +#warnings("-unused_async") +pub async fn Client::new( + uri : String, + headers? : Map[String, String] = {}, + proxy? : Client, +) -> Client { + ignore(proxy) + let (protocol, port, host, path) = resolve_url(uri) + guard path is "/" else { raise InvalidFormat } + Client::connect(host, protocol~, port~, headers~, proxy?) +} + +///| +pub impl @io.Writer for Client with write_once(self, buf, offset~, len~) { + guard self.request is Some(request) + request.body_writer.write_once(buf, offset~, len~) +} + +///| +#warnings("-unused_async") +pub async fn Client::flush(_ : Client) -> Unit { + () +} + +///| +pub impl @io.Reader for Client with _get_internal_buffer(self) { + guard self.response_body is Some(stream) + stream._get_internal_buffer() +} + +///| +pub impl @io.Reader for Client with _direct_read(self, buf, offset~, max_len~) { + guard self.response_body is Some(stream) + stream._direct_read(buf, offset~, max_len~) +} + +///| +pub async fn Client::end_request(self : Client) -> Response { + guard self.request is Some(request) + self.request = None + request.body_writer.close() + let js_response : JsResponse = request.response.wait( + abort_controller=request.abort_controller, + ) + self.response_body = Some( + @js_async.ReadableStream::from_js(js_response.body()), + ) + let headers = {} + js_response + .headers() + .for_each(fn(name, value) { headers[name.to_lower()] = value }) + { code: js_response.status(), reason: js_response.status_text(), headers } +} + +///| +fn wasm_fetch_request( + uri : String, + meth : String, + headers : JsHeaders, + body : RawExternRef, + signal : RawExternRef, +) -> RawExternRef = "moonbitlang_async_http" "fetch_request" + +///| +fn request_ffi( + uri : String, + meth : String, + headers : JsHeaders, + body : @js_async.JsReadableStream, + signal : @js_async.AbortSignal, +) -> @js_async.Promise[JsResponse] { + raw_cast( + wasm_fetch_request(uri, meth, headers, raw_cast(body), raw_cast(signal)), + ) +} + +///| +/// Send a HTTP request to the server. +/// Only the header of the request will be sent, +/// request body can be sent by using `Client` as a `@io.Writer`. +/// Once request body has been sent, +/// `end_request` must be called to complete the request and obtain response from the server. +/// +/// After performing a request, +/// the next request MUST NOT be made before the request is completed via `end_request`. +/// +/// In addition to headers in `Client::new`, +/// extra HTTP headers can be passed via `extra_headers`. +/// The headers mentioned in +/// https://developer.mozilla.org/en-US/docs/Glossary/Forbidden_request_header +/// must not be set in `extra_headers`. +#warnings("-unused_async") +pub async fn Client::request( + self : Client, + meth : RequestMethod, + path : String, + extra_headers? : Map[String, String] = {}, +) -> Unit { + guard self.request is None + let protocol = match self.protocol { + Http => "http://" + Https => "https://" + } + let path = if path is ['/', ..] { path } else { "/\{path}" } + let port = if self.port == self.protocol.default_port() { + "" + } else { + ":\{self.port}" + } + let uri = "\{protocol}\{self.host}\{port}\{path}" + let meth = match meth { + Get => "GET" + Head => "HEAD" + Post => "POST" + Put => "PUT" + Delete => "DELETE" + Connect => "CONNECT" + Options => "OPTIONS" + Trace => "TRACE" + Patch => "PATCH" + } + let headers = JsHeaders::new() + for k, v in self.headers { + headers.append(k, v) + } + for k, v in extra_headers { + headers.append(k, v) + } + let abort_controller = @js_async.AbortController::new() + let (body, we_write) = @js_async.JsReadableStream::new_pipe() + let response_promise = request_ffi( + uri, + meth, + headers, + body, + abort_controller.signal(), + ) + let request : OngoingRequest = { + abort_controller, + body_writer: we_write, + response: response_promise, + } + self.request = Some(request) +} + +///| +/// Perform a `GET` request to the server, see `Client::request` for more details. +pub async fn Client::get( + self : Client, + path : String, + extra_headers? : Map[String, String] = {}, + body? : &@io.Data, +) -> Response { + self.request(Get, path, extra_headers~) + if body is Some(body) { + self.write(body) + } + self.end_request() +} + +///| +/// Perform a `PUT` request to the server, see `Client::request` for more details. +pub async fn Client::put( + self : Client, + path : String, + body : &@io.Data, + extra_headers? : Map[String, String] = {}, +) -> Response { + self..request(Put, path, extra_headers~)..write(body).end_request() +} + +///| +/// Perform a `POST` request to the server, see `Client::request` for more details. +pub async fn Client::post( + self : Client, + path : String, + body : &@io.Data, + extra_headers? : Map[String, String] = {}, +) -> Response { + self..request(Post, path, extra_headers~)..write(body).end_request() +} diff --git a/src/http/moon.pkg b/src/http/moon.pkg index d733ccb42..6e5896047 100644 --- a/src/http/moon.pkg +++ b/src/http/moon.pkg @@ -25,11 +25,12 @@ options( targets: { "client.js.mbt": [ "js" ], "client.mbt": [ "native" ], + "client.wasm-gc.mbt": [ "wasm-gc" ], "client_test.mbt": [ "native", "js" ], "parser.mbt": [ "native" ], "parser_wbtest.mbt": [ "native" ], "proxy_test.mbt": [ "native" ], - "request.mbt": [ "native", "js" ], + "request.mbt": [ "native", "js", "wasm-gc" ], "request_test.mbt": [ "native", "js" ], "resolve_url_wbtest.mbt": [ "native" ], "send.mbt": [ "native" ], diff --git a/src/http/wasm-gc-imports.js b/src/http/wasm-gc-imports.js new file mode 100644 index 000000000..c0744d0f9 --- /dev/null +++ b/src/http/wasm-gc-imports.js @@ -0,0 +1,27 @@ +// Import object for moonbitlang/async http wasm-gc fetch client. +// When loading the compiled wasm-gc module, merge this into the import object: +// +// const imports = { ...otherImports, ...httpImports }; +// const { instance } = await WebAssembly.instantiateStreaming(fetch("module.wasm"), imports); + +export const moonbitlang_async_http = { + new_headers: () => new Headers(), + headers_append: (h, name, value) => h.append(name, value), + headers_for_each: (h, callback) => { + for (const [name, value] of h.entries()) callback(name, value); + }, + response_status: (r) => r.status, + response_status_text: (r) => r.statusText, + response_headers: (r) => r.headers, + response_body: (r) => r.body, + fetch_request: (uri, method, headers, body, signal) => { + const fixed_body = (method === "GET" || method === "HEAD") ? null : body; + return fetch(uri, { + body: fixed_body, + method: method, + headers: headers, + signal: signal, + duplex: 'half', + }); + }, +}; diff --git a/src/js_async/js_async.wasm-gc.mbt b/src/js_async/js_async.wasm-gc.mbt new file mode 100644 index 000000000..15252e67e --- /dev/null +++ b/src/js_async/js_async.wasm-gc.mbt @@ -0,0 +1,216 @@ +// Copyright 2025 International Digital Economy Academy +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +///| +/// A JavaScript promise that resolves to a value of type `X` +#external +pub type Promise[X] + +///| +#external +priv type JsValue + +///| +fn[X] JsValue::make(v : X) -> JsValue = "%identity" + +///| +fn[X] JsValue::cast(v : JsValue) -> X = "%identity" + +///| +fn JsValue::abort_error() -> JsValue = "moonbitlang_async_js" "abort_error" + +///| +fn JsValue::to_string(v : JsValue) -> String = "moonbitlang_async_js" "jsvalue_to_string" + +///| +/// A JavaScript exception wrapped in MoonBit error +suberror JsError { + JsError(JsValue) +} + +///| +pub impl Show for JsError with output(self, logger) { + let JsError(value) = self + logger.write_string(value.to_string()) +} + +///| +/// A JavaScript `AbortController` used to cancel promises +#external +pub type AbortController + +///| +/// A JavaScript `AbortSignal` used to cancel promises +#external +pub type AbortSignal + +///| +/// Create a new abort controller +pub fn AbortController::new() -> AbortController = "moonbitlang_async_js" "new_abort_controller" + +///| +/// Get the abort signal of an abort controller. +/// The abort signal will be triggered when `.abort()` is called on the controller. +pub fn AbortController::signal(self : Self) -> AbortSignal = "moonbitlang_async_js" "abort_controller_signal" + +///| +/// Abort the abort controller and activate its abort signal. +/// All promises tied to the signal will be cancelled. +pub fn AbortController::abort(self : Self) = "moonbitlang_async_js" "abort_controller_abort" + +///| +fn AbortSignal::on_abort(signal : AbortSignal, f : () -> Unit) -> Unit = "moonbitlang_async_js" "abort_signal_on_abort" + +///| +fn JsValue::then( + promise : JsValue, + resolve : (JsValue) -> Unit, + reject : (JsValue) -> Unit, +) = "moonbitlang_async_js" "promise_then" + +///| +/// A deferred promise with exposed resolve/reject handles. +/// Used instead of `new Promise(executor)` because wasm-gc cannot +/// directly call JS functions received as parameters. +#external +priv type Deferred + +///| +fn Deferred::new() -> Deferred = "moonbitlang_async_js" "create_deferred" + +///| +fn Deferred::resolve(self : Deferred, value : JsValue) -> Unit = "moonbitlang_async_js" "resolve_deferred" + +///| +fn Deferred::reject(self : Deferred, err : JsValue) -> Unit = "moonbitlang_async_js" "reject_deferred" + +///| +fn Deferred::promise(self : Deferred) -> JsValue = "moonbitlang_async_js" "get_deferred_promise" + +///| +/// Wait for a JavaScript promise to resolve and return the fulfilled value, +/// or raise an error if the promise is rejected. +/// +/// If `abort_controller` is provided, +/// it should be an abort controller that controls the promise, +/// and `.abort()` will be called on the controller automatically when `wait` is cancelled, +/// cancelling the promise automatically. +/// +/// If `abort_controller` is absent, `wait` is non-cancellable. +pub async fn[X] Promise::wait( + promise : Promise[X], + abort_controller? : AbortController, +) -> X { + struct Waiter { + mut coro : @coroutine.Coroutine? + mut ret : X? + mut err : Error? + } + let waiter = { + coro: Some(@coroutine.current_coroutine()), + ret: None, + err: None, + } + fn resolve(value : JsValue) { + waiter.ret = Some(value.cast()) + if waiter.coro is Some(coro) { + coro.wake() + @event_loop.reschedule() + } + } + + fn reject(err) { + waiter.err = Some(JsError(err)) + if waiter.coro is Some(coro) { + coro.wake() + @event_loop.reschedule() + } + } + + JsValue::make(promise).then(resolve, reject) + defer { + waiter.coro = None + } + if abort_controller is Some(controller) { + @coroutine.suspend() catch { + err => { + controller.abort() + raise err + } + } + } else { + @coroutine.protect_from_cancel(@coroutine.suspend) + } + if waiter.err is Some(err) { + raise err + } else { + waiter.ret.unwrap() + } +} + +///| +/// Convenient helper for calling async JavaScript code from MoonBit. +/// `run_promise(f)` create a fresh abort signal, passed it to `f`, +/// and wait for the promise that `f` returns. +/// If `f` resolve to a value, `run_promise` return that value. +/// If `f` is rejected with an error, `run_promise` raise that error. +/// If `run_promise` is cancelled, the abort signal passed to `f` is activated, +/// and the promise returned by `f` should be cancelled automatically. +/// +/// `run_promise` should only be used for cancellable JavaScript code +/// (i.e. the promised returned by `f` should properly handle the passed-in abort signal). +/// For non-cancellable JavaScript code, use `Promise::wait()` directly. +pub async fn[X] run_promise(f : (AbortSignal) -> Promise[X]) -> X { + let controller = AbortController::new() + let promise = f(controller.signal()) + promise.wait(abort_controller=controller) +} + +///| +/// Convert a MoonBit async function into a JavaScript promise. +/// `Promise::from_async(f)` returns a promise that: +/// +/// - resolve to the result of `f` when `f` return +/// - reject with the error that `f` raise if `f` fail +/// +/// If `abort_signal` is present, +/// `f` will be automatically cancelled when the signal is activated, +/// and the returned promise will reject with JavaScript `AbortError`. +/// +/// The async function `f` will be run in a global context, +/// so there is no structured concurrency support for `Promise::from_async`, +/// and this function should only be used for exporting MoonBit code to JavaScript. +/// +/// It is undefined whether `f` will actually start running immediately, +/// or start at the next JavaScript event loop. +pub fn[X] Promise::from_async( + f : async () -> X, + abort_signal? : AbortSignal, +) -> Promise[X] { + let deferred = Deferred::new() + let coro = @coroutine.spawn(() => { + try f() catch { + @coroutine.Cancelled if @coroutine.is_being_cancelled() => + deferred.reject(JsValue::abort_error()) + err => deferred.reject(JsValue::make(err.to_string())) + } noraise { + ret => deferred.resolve(JsValue::make(ret)) + } + }) + if abort_signal is Some(signal) { + signal.on_abort(() => coro.cancel()) + } + @coroutine.reschedule() + deferred.promise().cast() +} diff --git a/src/js_async/moon.pkg b/src/js_async/moon.pkg index 2222dc02c..d28cdd3d1 100644 --- a/src/js_async/moon.pkg +++ b/src/js_async/moon.pkg @@ -12,9 +12,11 @@ import { options( targets: { "js_async.mbt": [ "js" ], + "js_async.wasm-gc.mbt": [ "wasm-gc" ], "js_async_test.mbt": [ "js" ], "readable_stream.mbt": [ "js" ], + "readable_stream.wasm-gc.mbt": [ "wasm-gc" ], "readable_stream_test.mbt": [ "js" ], - "unimplemented.mbt": [ "native", "wasm", "wasm-gc" ], + "unimplemented.mbt": [ "native", "wasm" ], }, ) diff --git a/src/js_async/readable_stream.wasm-gc.mbt b/src/js_async/readable_stream.wasm-gc.mbt new file mode 100644 index 000000000..23fbebdbb --- /dev/null +++ b/src/js_async/readable_stream.wasm-gc.mbt @@ -0,0 +1,179 @@ +// Copyright 2025 International Digital Economy Academy +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +///| +#external +priv type StreamReader + +///| +/// Opaque result from StreamReader.read() — a JS {value, done} object. +#external +priv type ReadResult + +///| +fn StreamReader::read_ffi(reader : Self) -> Promise[ReadResult] = "moonbitlang_async_stream" "reader_read" + +///| +fn ReadResult::done(self : ReadResult) -> Bool = "moonbitlang_async_stream" "read_result_done" + +///| +fn ReadResult::value_length(self : ReadResult) -> Int = "moonbitlang_async_stream" "read_result_value_length" + +///| +fn ReadResult::copy_value( + self : ReadResult, + set_byte : (Int, Int) -> Unit, +) -> Unit = "moonbitlang_async_stream" "read_result_copy_value" + +///| +async fn StreamReader::read(reader : StreamReader) -> Bytes? { + let result : ReadResult = reader.read_ffi().wait() + if result.done() { + None + } else { + let len = result.value_length() + let buf = Array::make(len, b'\x00') + result.copy_value(fn(i, b) { buf[i] = b.to_byte() }) + Some(Bytes::from_array(buf[:])) + } +} + +///| +fn StreamReader::release_lock(reader : Self) -> Unit = "moonbitlang_async_stream" "reader_release_lock" + +///| +#external +priv type ReadableStreamController + +///| +fn ReadableStreamController::close(controller : Self) = "moonbitlang_async_stream" "controller_close" + +///| +fn wasm_controller_enqueue( + controller : ReadableStreamController, + len : Int, + get_byte : (Int) -> Int, +) -> Unit = "moonbitlang_async_stream" "controller_enqueue" + +///| +fn ReadableStreamController::enqueue(controller : Self, chunk : Bytes) -> Unit { + let len = chunk.length() + wasm_controller_enqueue(controller, len, fn(i) { chunk.at(i).to_int() }) +} + +///| +/// The `ReadableStream` type in Web API, see +/// https://developer.mozilla.org/en-US/docs/Web/API/ReadableStream +/// for more details. +/// +/// Use this type on JavaScript FFI boundary to interact with +/// API that make use of `ReadableStream`, such as `fetch`. +#external +type JsReadableStream + +///| +fn JsReadableStream::wasm_new( + pull : (ReadableStreamController) -> Promise[Unit], + cancel : (JsValue) -> Unit, +) -> JsReadableStream = "moonbitlang_async_stream" "new_readable_stream" + +///| +fn JsReadableStream::cancel(stream : Self) -> Unit = "moonbitlang_async_stream" "cancel_stream" + +///| +fn JsReadableStream::get_reader(stream : Self) -> StreamReader = "moonbitlang_async_stream" "get_reader" + +///| +/// A wrapper around `JsReadableStream` +/// that allows reading the content of the stream via the `@io.Reader` interface. +/// +/// This type SHOULD NOT be used on FFI directly: +/// its ABI is NOT the same as `JsReadableStream`, +/// and most JavaScript types won't understand it. +struct ReadableStream { + read_end : @io.PipeRead + worker : @coroutine.Coroutine +} + +///| +/// Create a `ReadableStream` wrapper from a `JsReadableStream` object, +/// in order to read the content of the stream from the MoonBit side. +/// The ownership of the JS stream will be transferred to the `ReadableStream`: +/// other code cannot read the stream anymore. +pub fn ReadableStream::from_js(stream : JsReadableStream) -> ReadableStream { + let (r, w) = @io.pipe() + let worker = @coroutine.spawn(() => { + defer w.close() + defer stream.cancel() + let reader = stream.get_reader() + defer reader.release_lock() + while reader.read() is Some(chunk) { + w.write(chunk) + } + }) + { read_end: r, worker } +} + +///| +/// Close a `ReadableStream`. The underlying JS stream will be cancelled. +pub fn ReadableStream::close(self : ReadableStream) -> Unit { + self.worker.cancel() + self.read_end.close() +} + +///| +pub impl @io.Reader for ReadableStream with _get_internal_buffer(self) { + self.read_end._get_internal_buffer() +} + +///| +pub impl @io.Reader for ReadableStream with _direct_read( + self, + buf, + offset~, + max_len~, +) { + self.read_end._direct_read(buf, offset~, max_len~) +} + +///| +/// Create an anonymous pipe and wrap the read end into a `JsReadableStream`. +/// This allows writing data from MoonBit code to JavaScript. +/// Usually users should pass the read end of the pipe (`JsReadableStream`) +/// to JavaScript code, and write to the write end of the pipe (`@io.PipeWrite`) +/// from MoonBit code. +/// +/// Closing the write end of the pipe will signal EOF on the read end. +/// If the read end of the stream is cancelled from the JavaScript side, +/// further write operations on the write end will fail with error. +pub fn JsReadableStream::new_pipe() -> (JsReadableStream, @io.PipeWrite) { + let abort_controller = AbortController::new() + let abort_signal = abort_controller.signal() + let (pipe_r, pipe_w) = @io.pipe() + fn pull(controller : ReadableStreamController) { + Promise::from_async(abort_signal~, () => { + guard pipe_r.read_some(max_len=1024) is Some(chunk) else { + controller.close() + } + controller.enqueue(chunk) + }) + } + fn cancel(_reason) { + pipe_r.close() + abort_controller.abort() + @coroutine.reschedule() + } + let stream = JsReadableStream::wasm_new(pull, cancel) + (stream, pipe_w) +} diff --git a/src/js_async/wasm-gc-imports.js b/src/js_async/wasm-gc-imports.js new file mode 100644 index 000000000..2f67ef17b --- /dev/null +++ b/src/js_async/wasm-gc-imports.js @@ -0,0 +1,53 @@ +// Import object for moonbitlang/async js_async wasm-gc Promise and stream bridging. +// When loading the compiled wasm-gc module, merge this into the import object: +// +// const imports = { ...otherImports, ...jsAsyncImports }; +// const { instance } = await WebAssembly.instantiateStreaming(fetch("module.wasm"), imports); + +export const moonbitlang_async_js = { + abort_error: () => { + const err = new Error(); + err.name = 'AbortError'; + return err; + }, + jsvalue_to_string: (v) => v.toString(), + new_abort_controller: () => new AbortController(), + abort_controller_signal: (ctrl) => ctrl.signal, + abort_controller_abort: (ctrl) => ctrl.abort(), + abort_signal_on_abort: (signal, f) => + signal.addEventListener('abort', f, { once: true }), + promise_then: (p, resolve, reject) => p.then(resolve, reject), + create_deferred: () => { + let resolve, reject; + const promise = new Promise((res, rej) => { + resolve = res; + reject = rej; + }); + return { promise, resolve, reject }; + }, + resolve_deferred: (d, value) => d.resolve(value), + reject_deferred: (d, err) => d.reject(err), + get_deferred_promise: (d) => d.promise, +}; + +export const moonbitlang_async_stream = { + reader_read: (reader) => reader.read(), + reader_release_lock: (reader) => reader.releaseLock(), + read_result_done: (result) => result.done, + read_result_value_length: (result) => + result.value ? result.value.byteLength : 0, + read_result_copy_value: (result, set_byte) => { + const arr = result.value; + for (let i = 0; i < arr.byteLength; i++) set_byte(i, arr[i]); + }, + controller_close: (ctrl) => ctrl.close(), + controller_enqueue: (ctrl, len, get_byte) => { + const arr = new Uint8Array(len); + for (let i = 0; i < len; i++) arr[i] = get_byte(i); + ctrl.enqueue(arr); + }, + new_readable_stream: (pull, cancel) => + new ReadableStream({ start: pull, pull: pull, cancel: cancel }), + cancel_stream: (stream) => stream.cancel(), + get_reader: (stream) => stream.getReader(), +}; From da6b6b8a9c96a00f23aecf395c03d46350b31223 Mon Sep 17 00:00:00 2001 From: duobei Date: Thu, 12 Mar 2026 12:07:52 +0800 Subject: [PATCH 4/8] docs: fix wasm-gc-imports.js usage examples to match actual exports --- src/http/wasm-gc-imports.js | 3 ++- src/js_async/wasm-gc-imports.js | 7 ++++--- 2 files changed, 6 insertions(+), 4 deletions(-) diff --git a/src/http/wasm-gc-imports.js b/src/http/wasm-gc-imports.js index c0744d0f9..e2d906dae 100644 --- a/src/http/wasm-gc-imports.js +++ b/src/http/wasm-gc-imports.js @@ -1,7 +1,8 @@ // Import object for moonbitlang/async http wasm-gc fetch client. // When loading the compiled wasm-gc module, merge this into the import object: // -// const imports = { ...otherImports, ...httpImports }; +// import { moonbitlang_async_http } from "./wasm-gc-imports.js"; +// const imports = { ...otherImports, moonbitlang_async_http }; // const { instance } = await WebAssembly.instantiateStreaming(fetch("module.wasm"), imports); export const moonbitlang_async_http = { diff --git a/src/js_async/wasm-gc-imports.js b/src/js_async/wasm-gc-imports.js index 2f67ef17b..9f58e73d8 100644 --- a/src/js_async/wasm-gc-imports.js +++ b/src/js_async/wasm-gc-imports.js @@ -1,7 +1,8 @@ -// Import object for moonbitlang/async js_async wasm-gc Promise and stream bridging. -// When loading the compiled wasm-gc module, merge this into the import object: +// Import objects for moonbitlang/async js_async wasm-gc Promise and stream bridging. +// When loading the compiled wasm-gc module, merge these into the import object: // -// const imports = { ...otherImports, ...jsAsyncImports }; +// import { moonbitlang_async_js, moonbitlang_async_stream } from "./wasm-gc-imports.js"; +// const imports = { ...otherImports, moonbitlang_async_js, moonbitlang_async_stream }; // const { instance } = await WebAssembly.instantiateStreaming(fetch("module.wasm"), imports); export const moonbitlang_async_js = { From da10060aa7b44f90db4871d5841641b41b95eb2b Mon Sep 17 00:00:00 2001 From: duobei Date: Thu, 12 Mar 2026 12:22:14 +0800 Subject: [PATCH 5/8] ci: re-trigger CI From 845498a24b9038191e80ae042323b44797a8230c Mon Sep 17 00:00:00 2001 From: duobei Date: Thu, 12 Mar 2026 12:28:32 +0800 Subject: [PATCH 6/8] test: update readdir snapshot to include wasm-gc files --- src/js_async/js_async_test.mbt | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/js_async/js_async_test.mbt b/src/js_async/js_async_test.mbt index d271ef914..787db555e 100644 --- a/src/js_async/js_async_test.mbt +++ b/src/js_async/js_async_test.mbt @@ -67,8 +67,10 @@ async test "read dir" { recursive: false, }).wait() json_inspect(files, content=[ - "js_async.mbt", "js_async_test.mbt", "moon.pkg", "pkg.generated.mbti", "readable_stream.mbt", + "js_async.mbt", "js_async.wasm-gc.mbt", "js_async_test.mbt", "moon.pkg", + "pkg.generated.mbti", "readable_stream.mbt", "readable_stream.wasm-gc.mbt", "readable_stream_test.mbt", "unimplemented.mbt", "unimplemented_test.mbt", + "wasm-gc-imports.js", ]) } From 249894d74df147edc89f233e023a5834079013c3 Mon Sep 17 00:00:00 2001 From: duobei Date: Thu, 12 Mar 2026 12:51:38 +0800 Subject: [PATCH 7/8] style: fix moon fmt formatting --- src/js_async/js_async_test.mbt | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/src/js_async/js_async_test.mbt b/src/js_async/js_async_test.mbt index 787db555e..ca371e7d7 100644 --- a/src/js_async/js_async_test.mbt +++ b/src/js_async/js_async_test.mbt @@ -67,10 +67,9 @@ async test "read dir" { recursive: false, }).wait() json_inspect(files, content=[ - "js_async.mbt", "js_async.wasm-gc.mbt", "js_async_test.mbt", "moon.pkg", - "pkg.generated.mbti", "readable_stream.mbt", "readable_stream.wasm-gc.mbt", - "readable_stream_test.mbt", "unimplemented.mbt", "unimplemented_test.mbt", - "wasm-gc-imports.js", + "js_async.mbt", "js_async.wasm-gc.mbt", "js_async_test.mbt", "moon.pkg", "pkg.generated.mbti", + "readable_stream.mbt", "readable_stream.wasm-gc.mbt", "readable_stream_test.mbt", + "unimplemented.mbt", "unimplemented_test.mbt", "wasm-gc-imports.js", ]) } From 5fd8987c175c1af40e6d5735988378f7e7c3ce8c Mon Sep 17 00:00:00 2001 From: duobei Date: Thu, 12 Mar 2026 13:10:57 +0800 Subject: [PATCH 8/8] ci: retry flaky request cancel test