diff --git a/_typos.toml b/_typos.toml index 8bc0f2fa3f..a5d72f6083 100644 --- a/_typos.toml +++ b/_typos.toml @@ -56,3 +56,4 @@ shft = "shft" multline = "multline" sav = "sav" ist = "ist" +Hel = "Hel" diff --git a/src/plugins/system/android/com/foxdebug/system/StreamHttp.java b/src/plugins/system/android/com/foxdebug/system/StreamHttp.java new file mode 100644 index 0000000000..d59bb2f52d --- /dev/null +++ b/src/plugins/system/android/com/foxdebug/system/StreamHttp.java @@ -0,0 +1,400 @@ +package com.foxdebug.system; + +import android.util.Base64; +import android.util.Log; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.net.HttpURLConnection; +import java.net.URL; +import java.nio.charset.StandardCharsets; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +import org.apache.cordova.CallbackContext; +import org.apache.cordova.PluginResult; +import org.json.JSONArray; +import org.json.JSONException; +import org.json.JSONObject; + +/** + * Incrementally streams an HTTP response body to JavaScript. + * + *

The request runs on a dedicated background thread (never the UI thread). + * Response bytes are forwarded to the bridge as they arrive. To survive the + * JSON bridge without a base64 round trip, chunks are sent as raw bytes mapped + * onto ISO-8859-1 characters (byte n == code point n), which the JSON bridge + * round-trips losslessly; only chunks containing control bytes (< 0x20, + * which JSON must escape) fall back to base64. The native layer performs no + * SSE/LLM specific parsing: chunk boundaries are arbitrary and may split + * multi-byte UTF-8 characters or SSE frames. + * + *

Events emitted to JS (JSON object with a {@code type} field): + *

+ * + *

Exactly one terminal event ({@code complete} or {@code error}) is ever + * emitted per stream. 4xx/5xx HTTP statuses are still delivered as a normal + * {@code headers} + body stream and remain distinguishable from network + * failures. + */ +public class StreamHttp implements Runnable { + + private static final String TAG = "SystemStreamHttp"; + private static final int DEFAULT_CHUNK_SIZE = 32 * 1024; + private static final int MAX_CHUNK_SIZE = 100 * 1024; + + private static final int MAX_CONCURRENT_STREAMS = 50; + + private static final int CREDIT_WINDOW_BYTES = 256 * 1024; + + private static final ConcurrentHashMap STREAMS = new ConcurrentHashMap<>(); + + private final String requestId; + private final String url; + private final String method; + private final JSONObject headers; + private final String body; + private final boolean bodyIsBase64; + private final boolean followRedirects; + private final int connectTimeout; + private final int readTimeout; + private final int chunkSize; + private final CallbackContext callback; + + private final Object creditLock = new Object(); + private volatile boolean cancelled; + private volatile boolean finished; + private long pendingBytes; + + private HttpURLConnection connection; + private InputStream inputStream; + + public StreamHttp( + String requestId, + String url, + String method, + JSONObject headers, + String body, + boolean bodyIsBase64, + boolean followRedirects, + int connectTimeout, + int readTimeout, + int chunkSize, + CallbackContext callback + ) { + this.requestId = requestId; + this.url = url; + this.method = method; + this.headers = headers; + this.body = body; + this.bodyIsBase64 = bodyIsBase64; + this.followRedirects = followRedirects; + this.connectTimeout = connectTimeout; + this.readTimeout = readTimeout; + this.chunkSize = chunkSize > 0 + ? Math.min(chunkSize, MAX_CHUNK_SIZE) + : DEFAULT_CHUNK_SIZE; + this.callback = callback; + } + + public static void start( + String requestId, + String url, + String method, + JSONObject headers, + String body, + boolean bodyIsBase64, + boolean followRedirects, + int connectTimeout, + int readTimeout, + int chunkSize, + CallbackContext callback + ) { + StreamHttp stream = new StreamHttp( + requestId, + url, + method, + headers, + body, + bodyIsBase64, + followRedirects, + connectTimeout, + readTimeout, + chunkSize, + callback + ); + if (STREAMS.size() >= MAX_CONCURRENT_STREAMS) { + callback.error( + "Too many concurrent HTTP streams (" + + MAX_CONCURRENT_STREAMS + + "). Cancel an active stream or retry later." + ); + return; + } + STREAMS.put(requestId, stream); + Thread thread = new Thread(stream, "SystemStreamHttp-" + requestId); + thread.setDaemon(true); + thread.start(); + } + + /** Acknowledges that JavaScript has consumed {@code bytes} response bytes. */ + public static void ack(String requestId, int bytes) { + StreamHttp stream = STREAMS.get(requestId); + if (stream != null) stream.ack(bytes); + } + + /** Cancels the request and disconnects the underlying connection. */ + public static void cancel(String requestId) { + StreamHttp stream = STREAMS.get(requestId); + if (stream != null) stream.cancel(); + } + + /** Cancels every active stream. Used during plugin/app teardown. */ + public static void cancelAll() { + for (StreamHttp stream : STREAMS.values()) { + stream.cancel(); + } + } + + private void ack(int bytes) { + if (bytes <= 0) return; + synchronized (creditLock) { + pendingBytes = Math.max(0, pendingBytes - bytes); + creditLock.notifyAll(); + } + } + + private void cancel() { + cancelled = true; + synchronized (creditLock) { + creditLock.notifyAll(); + } + if (connection != null) { + try { + connection.disconnect(); + } catch (Exception e) { + Log.w(TAG, "Failed to disconnect stream " + requestId, e); + } + } + release(); + } + + private void release() { + if (finished) return; + finished = true; + try { + PluginResult result = new PluginResult(PluginResult.Status.NO_RESULT); + result.setKeepCallback(false); + callback.sendPluginResult(result); + } catch (Exception e) { + Log.w(TAG, "Failed to release stream callback " + requestId, e); + } + } + + @Override + public void run() { + try { + if (cancelled) return; + URL url = new URL(this.url); + connection = (HttpURLConnection) url.openConnection(); + connection.setRequestMethod(method); + connection.setInstanceFollowRedirects(followRedirects); + connection.setConnectTimeout(connectTimeout); + connection.setReadTimeout(readTimeout); + connection.setUseCaches(false); + + if (headers != null) { + Iterator keys = headers.keys(); + while (keys.hasNext()) { + String key = keys.next(); + String value = headers.optString(key); + if (key != null && value != null) { + connection.setRequestProperty(key, value); + } + } + } + + if (body != null) { + byte[] bytes = bodyIsBase64 + ? Base64.decode(body, Base64.NO_WRAP) + : body.getBytes(StandardCharsets.UTF_8); + connection.setFixedLengthStreamingMode(bytes.length); + connection.setDoOutput(true); + OutputStream out = connection.getOutputStream(); + out.write(bytes); + out.flush(); + } + + if (cancelled) return; + + int status = connection.getResponseCode(); + if (cancelled) return; + + String statusText = connection.getResponseMessage(); + String finalUrl = connection.getURL().toString(); + sendHeaders(status, statusText, finalUrl, connection.getHeaderFields()); + + InputStream stream = status >= 400 + ? connection.getErrorStream() + : connection.getInputStream(); + inputStream = stream; + + if (stream != null) { + byte[] buffer = new byte[chunkSize]; + int read; + while (!cancelled && (read = stream.read(buffer)) != -1) { + if (read <= 0) continue; + waitForCredit(read); + if (cancelled) break; + byte[] chunk = read == buffer.length + ? buffer + : java.util.Arrays.copyOf(buffer, read); + if (read == buffer.length) { + buffer = new byte[chunkSize]; + } + // Reserve credit before publishing: ack() can only race against + // chunks JavaScript has already received, so incrementing here keeps + // the subtraction in ack() from ever clamping a not-yet-accounted + // reservation to zero and silently exhausting the credit window. + synchronized (creditLock) { + pendingBytes += read; + } + sendData(chunk); + } + } + + if (!cancelled) { + sendComplete(); + } + } catch (Exception e) { + Log.w(TAG, "Stream " + requestId + " failed", e); + if (!cancelled && !finished) { + try { + sendError(e.getMessage() != null ? e.getMessage() : e.toString()); + } catch (JSONException jsonError) { + Log.w(TAG, "Failed to send stream error event", jsonError); + } + } + } finally { + cleanup(); + } + } + + private void waitForCredit(long bytes) { + synchronized (creditLock) { + while (!cancelled && pendingBytes + bytes > CREDIT_WINDOW_BYTES) { + try { + creditLock.wait(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + if (cancelled) return; + } + } + } + } + + private void sendHeaders( + int status, + String statusText, + String finalUrl, + Map> headerFields + ) throws JSONException { + JSONObject event = new JSONObject(); + event.put("type", "headers"); + event.put("status", status); + event.put("statusText", statusText != null ? statusText : ""); + event.put("url", finalUrl != null ? finalUrl : ""); + + // Delivered as ordered [name, value] pairs so repeated headers (e.g. + // Set-Cookie, which cannot be comma joined) survive the JSON bridge. + JSONArray headerPairs = new JSONArray(); + if (headerFields != null) { + for (Map.Entry> entry : headerFields.entrySet()) { + String name = entry.getKey(); + List values = entry.getValue(); + if (name == null || values == null || values.isEmpty()) { + continue; + } + String lower = name.toLowerCase(); + for (String value : values) { + if (value == null) continue; + JSONArray pair = new JSONArray(); + pair.put(lower); + pair.put(value); + headerPairs.put(pair); + } + } + } + event.put("headers", headerPairs); + sendEvent(event, true); + } + + private void sendData(byte[] chunk) throws JSONException { + JSONObject event = new JSONObject(); + event.put("type", "data"); + if (jsonSafe(chunk)) { + // Byte n as code point n survives the JSON bridge unescaped, avoiding + // the ~33% base64 inflation and the base64 decode on the JS side. + event.put("chunk", new String(chunk, StandardCharsets.ISO_8859_1)); + } else { + event.put("b64", true); + event.put("chunk", Base64.encodeToString(chunk, Base64.NO_WRAP)); + } + sendEvent(event, true); + } + + // True when the chunk has no byte below 0x20, i.e. nothing the JSON encoder + // is forced to escape (as a U+XXXX sequence) into a larger payload. + private static boolean jsonSafe(byte[] chunk) { + for (byte b : chunk) { + if ((b & 0xFF) < 0x20) return false; + } + return true; + } + + private void sendComplete() throws JSONException { + JSONObject event = new JSONObject(); + event.put("type", "complete"); + sendEvent(event, false); + } + + private void sendError(String message) throws JSONException { + JSONObject event = new JSONObject(); + event.put("type", "error"); + event.put("message", message); + sendEvent(event, false); + } + + private void sendEvent(JSONObject event, boolean keepCallback) { + if (finished) return; + if (!keepCallback) finished = true; + PluginResult result = new PluginResult(PluginResult.Status.OK, event); + result.setKeepCallback(keepCallback); + callback.sendPluginResult(result); + } + + private void cleanup() { + STREAMS.remove(requestId); + if (inputStream != null) { + try { + inputStream.close(); + } catch (IOException ignored) { + } + } + if (connection != null) { + try { + connection.disconnect(); + } catch (Exception ignored) { + } + } + } +} diff --git a/src/plugins/system/android/com/foxdebug/system/System.java b/src/plugins/system/android/com/foxdebug/system/System.java index 53e0f6a6e7..9628aa8cf3 100644 --- a/src/plugins/system/android/com/foxdebug/system/System.java +++ b/src/plugins/system/android/com/foxdebug/system/System.java @@ -118,6 +118,19 @@ public void run() { ); } + @Override + public void onReset() { + super.onReset(); + StreamHttp.cancelAll(); + intentHandler = null; + } + + @Override + public void onDestroy() { + StreamHttp.cancelAll(); + super.onDestroy(); + } + public boolean execute( String action, final JSONArray args, @@ -159,6 +172,17 @@ public boolean execute( case "get-configuration": getConfiguration(callbackContext); return true; + case "http-stream-start": + httpStreamStart(args, callbackContext); + return true; + case "http-stream-ack": + StreamHttp.ack(arg1, args.optInt(1, 0)); + callbackContext.success(); + return true; + case "http-stream-cancel": + StreamHttp.cancel(arg1); + callbackContext.success(); + return true; case "set-input-type": setInputType(arg1); callbackContext.success(); @@ -656,6 +680,52 @@ private String getMimeTypeFromExtension(String fileName) { return mimeType != null ? mimeType : "application/octet-stream"; } + /** + * Starts a streaming HTTP request. The response body is delivered to + * JavaScript incrementally as raw (or base64 encoded) chunks. + * + *

Args: + *

    + *
  1. requestId - unique id used for pause/resume/cancel
    2. + *
  2. url
    3. + *
  3. options JSON object: + * method, headers, body, bodyIsBase64, followRedirects, + * connectTimeout, readTimeout, chunkSize
    4. + *
+ */ + private void httpStreamStart(JSONArray args, CallbackContext callbackContext) { + try { + final String requestId = args.getString(0); + final String url = args.getString(1); + final JSONObject options = args.getJSONObject(2); + + final String method = options.optString("method", "GET"); + final JSONObject headers = options.optJSONObject("headers"); + final String body = options.isNull("body") ? null : options.optString("body"); + final boolean bodyIsBase64 = options.optBoolean("bodyIsBase64", false); + final boolean followRedirects = options.optBoolean("followRedirects", true); + final int connectTimeout = options.optInt("connectTimeout", 30000); + final int readTimeout = options.optInt("readTimeout", 0); + final int chunkSize = options.optInt("chunkSize", 0); + + StreamHttp.start( + requestId, + url, + method, + headers, + body, + bodyIsBase64, + followRedirects, + connectTimeout, + readTimeout, + chunkSize, + callbackContext + ); + } catch (Exception e) { + callbackContext.error("Failed to start stream: " + e.getMessage()); + } + } + private void getConfiguration(CallbackContext callback) { try { JSONObject result = new JSONObject(); diff --git a/src/plugins/system/plugin.xml b/src/plugins/system/plugin.xml index ed80bf0cb4..bd56bfe372 100644 --- a/src/plugins/system/plugin.xml +++ b/src/plugins/system/plugin.xml @@ -17,6 +17,10 @@ + + + + @@ -42,5 +46,6 @@ + diff --git a/src/plugins/system/readme.md b/src/plugins/system/readme.md deleted file mode 100644 index b0cca56d24..0000000000 --- a/src/plugins/system/readme.md +++ /dev/null @@ -1,11 +0,0 @@ -# Util plugin for cordova apps - -Using this plugin, cordova apps can: - -- Enable/disable full screen -- Share file -- Get webview information -- Send email -- Clear cache - -## Installation diff --git a/src/plugins/system/system.d.ts b/src/plugins/system/system.d.ts index 6131045068..8100a2b99f 100644 --- a/src/plugins/system/system.d.ts +++ b/src/plugins/system/system.d.ts @@ -59,6 +59,18 @@ type FileAction = 'VIEW' | 'EDIT' | 'SEND' | 'RUN'; type OnFail = (err: string) => void; type OnSuccessBool = (res: boolean) => void; +interface HttpStreamOptions { + method?: string; + headers?: Record; + body?: string; + bodyIsBase64?: boolean; + followRedirects?: boolean; + connectTimeout?: number; + readTimeout?: number; + chunkSize?: number; + signal?: AbortSignal; +} + interface System { /** * Get information about current webview @@ -296,6 +308,21 @@ interface System { onFail?: OnFail, ): void; /** + * Perform an HTTP request and stream the response body to JavaScript. + * + * The response body is exposed as a WHATWG `ReadableStream` of `Uint8Array` + * chunks. The native layer does not buffer the whole response and performs + * no SSE/provider specific parsing. A 4xx/5xx status is a normal response; + * only transport failures reject the promise. Cancelling the returned + * stream's reader (or aborting `options.signal`) cancels the underlying + * native HTTP request. + * + * @param url Request URL + * @param options Request options + * @returns A `Response` whose `body` is a `ReadableStream` of `Uint8Array` chunks + */ + httpStream(url: string, options?: HttpStreamOptions): Promise; + /* * Change the app icon at runtime. * @param iconName Icon id, e.g. "midnight_circuit", or "default" to restore the original icon * @param onSuccess diff --git a/src/plugins/system/www/plugin.js b/src/plugins/system/www/plugin.js index c8722e7e4a..e10bc20062 100644 --- a/src/plugins/system/www/plugin.js +++ b/src/plugins/system/www/plugin.js @@ -281,5 +281,231 @@ module.exports = { [text1, text2] ); }); + }, + /** + * Make an HTTP request and receive the response body as a WHATWG + * `ReadableStream` of `Uint8Array` chunks, as bytes arrive from the server. + * + * The native layer does not buffer the whole response and performs no SSE / + * provider specific parsing; it simply forwards raw byte chunks. Chunk + * boundaries are arbitrary and may split multi-byte UTF-8 characters or SSE + * frames. The consumer is responsible for decoding / parsing the stream. + * + * @param {string} url - Request URL + * @param {Object} [options] + * @param {string} [options.method="GET"] - HTTP method + * @param {Object} [options.headers] - Request headers + * @param {string} [options.body] - Request body. Sent as UTF-8 text unless + * `bodyIsBase64` is set, in which case it is decoded from base64. + * @param {boolean} [options.bodyIsBase64=false] + * @param {boolean} [options.followRedirects=true] + * @param {number} [options.connectTimeout=30000] - Connect timeout in ms + * @param {number} [options.readTimeout=0] - Read timeout in ms (0 = none) + * @param {number} [options.chunkSize=32768] - Requested native chunk size in bytes + * @param {AbortSignal} [options.signal] - When aborted, the underlying native + * request is cancelled. If the headers have not yet arrived the returned + * promise rejects with an `AbortError`; otherwise the response stream is + * errored with an `AbortError`. + * @returns {Promise} Resolves with a `Response` whose `body` is a + * `ReadableStream` delivering `Uint8Array` chunks. A 4xx/5xx HTTP status + * is a normal response (not a rejected promise); only transport failures + * reject. Cancelling the returned stream's reader (or aborting + * `options.signal`) cancels the underlying native HTTP request. + */ + httpStream: function (url, options) { + options = options || {}; + var signal = options.signal || null; + + var nativeOptions = {}; + for (var key in options) { + if (key !== 'signal') nativeOptions[key] = options[key]; + } + + return new Promise(function (resolve, reject) { + var requestId = "httpStream_" + Date.now() + "_" + Math.random().toString(36).slice(2, 10); + var HIGH_WATER_MARK = 65536; + var controller = null; + var headersReceived = false; + var started = false; + var cancelSent = false; + var terminal = false; + var receivedBytes = 0; + var ackedBytes = 0; + + function sendCancel() { + if (cancelSent) return; + cancelSent = true; + cordova.exec(null, null, 'System', 'http-stream-cancel', [requestId]); + } + + function teardownSignal() { + if (signal) { + try { + signal.removeEventListener('abort', onAbort); + } catch (e) {} + } + } + + function finish() { + terminal = true; + teardownSignal(); + } + + function fail(err) { + if (terminal) return; + finish(); + if (headersReceived && controller) { + controller.error(err); + } else { + reject(err); + } + } + + function onAbort() { + if (terminal) return; + if (started) sendCancel(); + var err = new Error('The http stream was aborted'); + err.name = 'AbortError'; + fail(err); + } + + function ackConsumed() { + if (terminal || !controller) return; + var desired = controller.desiredSize; + if (desired === null) return; + var buffered = Math.max(0, HIGH_WATER_MARK - desired); + var consumed = receivedBytes - buffered; + var delta = consumed - ackedBytes; + if (delta > 0) { + ackedBytes = consumed; + cordova.exec(null, null, 'System', 'http-stream-ack', [requestId, delta]); + } + } + + function headersFromPairs(pairs) { + var h = new Headers(); + if (!pairs) return h; + if (!Array.isArray(pairs)) { + for (var name in pairs) { + try { + h.append(name, pairs[name]); + } catch (e) {} + } + return h; + } + for (var i = 0; i < pairs.length; i++) { + var pair = pairs[i]; + if (!pair || pair.length < 2) continue; + try { + h.append(pair[0], pair[1]); + } catch (e) {} + } + return h; + } + + var stream = new ReadableStream({ + start: function (c) { + controller = c; + }, + pull: function () { + ackConsumed(); + }, + cancel: function () { + finish(); + if (started) sendCancel(); + } + }, { + highWaterMark: HIGH_WATER_MARK, + size: function (chunk) { + return chunk.byteLength; + } + }); + + if (signal) { + if (signal.aborted) { + onAbort(); + } else { + signal.addEventListener('abort', onAbort); + } + } + if (terminal) return; + + started = true; + cordova.exec( + function (event) { + if (!event || typeof event !== 'object' || terminal) return; + + switch (event.type) { + case 'headers': { + headersReceived = true; + var status = event.status; + var cannotHaveBody = status === 204 || status === 205 || status === 304; + var headers = headersFromPairs(event.headers || []); + var response; + if (cannotHaveBody) { + response = new Response(null, { + status: status, + statusText: event.statusText || '', + headers: headers + }); + } else { + response = new Response(stream, { + status: status, + statusText: event.statusText || '', + headers: headers + }); + } + if (event.url) { + Object.defineProperty(response, 'url', { value: event.url, configurable: true }); + } + resolve(response); + break; + } + case 'data': { + if (controller && event.chunk) { + var bytes = event.b64 + ? base64ToBytes(event.chunk) + : latin1ToBytes(event.chunk); + controller.enqueue(bytes); + receivedBytes += bytes.byteLength; + } + break; + } + case 'complete': { + finish(); + if (controller) controller.close(); + break; + } + case 'error': { + fail(new Error(event.message || 'Stream failed')); + break; + } + } + }, + function (err) { + fail(typeof err === 'string' ? new Error(err) : err); + }, + 'System', + 'http-stream-start', + [requestId, url, nativeOptions] + ); + }); } }; + +function base64ToBytes(base64) { + var binary = atob(base64); + var bytes = new Uint8Array(binary.length); + for (var i = 0; i < binary.length; i++) { + bytes[i] = binary.charCodeAt(i); + } + return bytes; +} + +function latin1ToBytes(text) { + var bytes = new Uint8Array(text.length); + for (var i = 0; i < text.length; i++) { + bytes[i] = text.charCodeAt(i); + } + return bytes; +} diff --git a/tests/unit/systemHttpStream.test.js b/tests/unit/systemHttpStream.test.js new file mode 100644 index 0000000000..fe20d2b5ae --- /dev/null +++ b/tests/unit/systemHttpStream.test.js @@ -0,0 +1,556 @@ +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import http from "node:http"; + +import system from "../../src/plugins/system/www/plugin.js"; + +const CREDIT_WINDOW = 128 * 1024; +const BRIDGE_CHUNK = 8192; + +class TestServer { + constructor() { + this.server = http.createServer((req, res) => this.handle(req, res)); + this.port = 0; + this.requestLog = []; + } + + listen() { + return new Promise((resolve) => { + this.server.listen(0, "127.0.0.1", () => { + this.port = this.server.address().port; + resolve(); + }); + }); + } + + url(path) { + return `http://127.0.0.1:${this.port}${path}`; + } + + handle(req, res) { + const url = new URL(req.url, "http://localhost"); + this.requestLog.push({ method: req.method, path: url.pathname }); + + switch (url.pathname) { + case "/simple": + res.writeHead(200, { "Content-Type": "text/plain" }); + res.end("hello world"); + break; + + case "/genuine-stream": + // chunk 1 -> wait -> chunk 2 -> wait -> chunk 3 -> close + res.writeHead(200, { "Content-Type": "text/event-stream" }); + res.write('data: {"tok', () => { + setTimeout(() => { + res.write('en":"Hel"}\n\n', () => { + setTimeout(() => { + res.end('data: {"token":"lo"}\n\n'); + }, 300); + }); + }, 300); + }); + break; + + case "/utf8-split": + res.writeHead(200, { "Content-Type": "text/plain" }); + const text = "héllo \u00e9\u00e8\u00ea wörld"; + const bytes = Buffer.from(text, "utf8"); + res.write(bytes.subarray(0, 5)); + setTimeout(() => res.end(bytes.subarray(5)), 50); + break; + + case "/large": + res.writeHead(200, { "Content-Type": "application/octet-stream" }); + const block = Buffer.alloc(65536, 0x61); + let sent = 0; + const total = 1024 * 1024; // 1 MiB + const tick = () => { + res.write(block.subarray(0, Math.min(block.length, total - sent))); + sent += block.length; + if (sent < total) { + setImmediate(tick); + } else { + res.end(); + } + }; + tick(); + break; + + case "/status500-empty": + // 4xx/5xx response with NO error body: Java's getErrorStream() + // returns null and the response must be treated as an empty body + // followed by complete, not as a transport failure. + res.writeHead(500, { "Content-Type": "text/plain" }); + res.end(); + break; + + case "/network-error": + res.socket.destroy(); + break; + + case "/keep-open": + res.writeHead(200, { "Content-Type": "text/event-stream" }); + res.write("data: 1\n\n"); + break; + + case "/cookies": + res.writeHead(200, { + "Content-Type": "text/plain", + "Set-Cookie": ["session=abc; Path=/", "theme=dark; Path=/"], + }); + res.end("ok"); + break; + + case "/latin1-text": + // No control bytes (<0x20): must travel on the raw latin1 fast path. + res.writeHead(200, { "Content-Type": "text/plain" }); + res.end('say "hi" \\ path \u00e9\u00ff\u2028 done'); + break; + + case "/binary-ctrl": + // Contains control bytes (0x00/0x08/0x1f): must fall back to base64. + res.writeHead(200, { "Content-Type": "application/octet-stream" }); + res.end(Buffer.from([0x00, 0x08, 0x1f, 0x7f, 0xff, 0x22, 0x00])); + break; + + default: + res.writeHead(404, { "Content-Type": "text/plain" }); + res.end("not found"); + } + } + + close() { + return new Promise((resolve) => this.server.close(resolve)); + } +} + +/** + * An asynchronous fake native bridge that mirrors the real Cordova/Java + * contract: + * + * - The Java reader thread only hands bytes to the Cordova bridge while fewer + * than CREDIT_WINDOW bytes are awaiting a JavaScript ACK (`http-stream-ack`). + * - Delivery to JavaScript is asynchronous (like the Cordova message queue), so + * messages can sit in the bridge while JS catches up. A synchronous fake + * bridge hides real backpressure races, so this one does not deliver inline. + * + * The JS under test must ACK bytes only once the consumer has actually read + * them; if it ACKed ahead of consumption the windowed reader here would keep + * flowing and the bounded-buffer guarantees of the protocol would be broken. + */ +function createNativeBridge() { + const streams = new Map(); + const calls = []; + + const exec = (success, error, service, action, args) => { + calls.push({ action, args }); + + if (action === "http-stream-start") { + const [requestId, url, options] = args; + startStream(requestId, url, options, success); + } else if (action === "http-stream-ack") { + const s = streams.get(args[0]); + if (s) s.ack(args[1]); + if (success) success(); + } else if (action === "http-stream-cancel") { + const s = streams.get(args[0]); + if (s) s.cancel(); + if (success) success(); + } + }; + + function startStream(requestId, url, options, success) { + const u = new URL(url); + const headers = {}; + if (options.headers) { + for (const [k, v] of Object.entries(options.headers)) headers[k] = v; + } + + const req = http.request( + { + hostname: u.hostname, + port: u.port, + path: u.pathname + u.search, + method: options.method || "GET", + headers, + }, + (res) => { + const pairs = []; + for (const [k, v] of Object.entries(res.headers)) { + const values = Array.isArray(v) ? v : [v]; + for (const value of values) { + pairs.push([k, value]); + } + } + success({ + type: "headers", + status: res.statusCode, + statusText: res.statusMessage || "", + url, + headers: pairs, + }); + + const entry = { + pending: 0, + acked: 0, + forwarded: 0, + maxPending: 0, + backlog: [], + delivered: [], + cancelled: false, + ended: false, + completeSent: false, + req, + res, + }; + streams.set(requestId, entry); + + // Mirrors the real Java bridge: PluginResults are JSON encoded on + // the native side and parsed on the JS side. + const deliver = (msg) => + setImmediate(() => success(JSON.parse(JSON.stringify(msg)))); + + // Mirrors StreamHttp.sendData: chunks without control bytes (<0x20) + // travel as raw ISO-8859-1 characters; everything else uses base64. + const encodeChunk = (buf) => { + for (const b of buf) { + if (b < 0x20) { + return { chunk: buf.toString("base64"), b64: true }; + } + } + return { chunk: buf.toString("latin1") }; + }; + + const flush = () => { + while (entry.pending < CREDIT_WINDOW && entry.backlog.length > 0) { + const part = entry.backlog.shift(); + entry.pending += part.length; + entry.forwarded += part.length; + entry.maxPending = Math.max(entry.maxPending, entry.pending); + const msg = { type: "data", ...encodeChunk(part) }; + entry.delivered.push(msg); + deliver(msg); + } + if (entry.backlog.length === 0) { + res.resume(); + if (entry.ended && !entry.completeSent) { + entry.completeSent = true; + deliver({ type: "complete" }); + } + } + }; + + entry.ack = (bytes) => { + entry.pending = Math.max(0, entry.pending - bytes); + entry.acked += bytes; + flush(); + }; + + entry.cancel = () => { + entry.cancelled = true; + req.destroy(); + }; + + // Flowing mode so chunks arrive as the server writes them (the + // paused/readable mode coalesces everything into one event). + res.on("data", (chunk) => { + if (entry.cancelled) return; + for (let off = 0; off < chunk.length; off += BRIDGE_CHUNK) { + entry.backlog.push(chunk.subarray(off, off + BRIDGE_CHUNK)); + } + // Stop draining the socket while the credit window is full so + // the server experiences real backpressure, like StreamHttp. + res.pause(); + flush(); + }); + res.on("end", () => { + entry.ended = true; + flush(); + }); + res.on("error", () => { + if (!entry.cancelled && !entry.completeSent) { + entry.completeSent = true; + deliver({ type: "error", message: "socket error" }); + } + }); + }, + ); + + if (options.body != null) { + req.write(options.body); + } + req.end(); + + req.on("error", (err) => { + if (!streams.has(requestId)) { + success({ type: "error", message: err.message }); + } + }); + } + + return { exec, calls, streams }; +} + +const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms)); + +let server; +let bridge; +let origCordova; + +async function collect(response) { + if (!response.body) return []; + const reader = response.body.getReader(); + const chunks = []; + while (true) { + const { done, value } = await reader.read(); + if (done) break; + chunks.push(value); + } + return chunks; +} + +function concat(chunks) { + const total = chunks.reduce((n, c) => n + c.byteLength, 0); + const out = new Uint8Array(total); + let offset = 0; + for (const chunk of chunks) { + out.set(chunk, offset); + offset += chunk.byteLength; + } + return out; +} + +function decode(chunks) { + return Buffer.from(concat(chunks)).toString("utf8"); +} + +beforeAll(async () => { + server = new TestServer(); + await server.listen(); + + origCordova = globalThis.cordova; +}); + +afterAll(() => { + if (origCordova !== undefined) { + globalThis.cordova = origCordova; + } else { + delete globalThis.cordova; + } + return server.close(); +}); + +describe("system.httpStream", () => { + it("1. delivers a small streamed response", async () => { + bridge = createNativeBridge(); + globalThis.cordova = { exec: (...args) => bridge.exec(...args) }; + const response = await system.httpStream(server.url("/simple")); + expect(response.status).toBe(200); + expect(decode(await collect(response))).toBe("hello world"); + }); + + it("2. streams genuinely: chunk 1 arrives before the later chunks exist", async () => { + bridge = createNativeBridge(); + globalThis.cordova = { exec: (...args) => bridge.exec(...args) }; + + const start = Date.now(); + const response = await system.httpStream(server.url("/genuine-stream")); + const reader = response.body.getReader(); + const received = []; + + while (true) { + const { done, value } = await reader.read(); + if (done) break; + received.push({ time: Date.now() - start, chunk: value }); + } + + const text = received + .map((r) => Buffer.from(r.chunk).toString("utf8")) + .join(""); + expect(text).toBe('data: {"token":"Hel"}\n\ndata: {"token":"lo"}\n\n'); + expect(received.length).toBeGreaterThanOrEqual(2); + expect(received[0].time).toBeLessThan(280); + expect(received[received.length - 1].time).toBeGreaterThanOrEqual(500); + }); + + it("3. handles chunks that split a UTF-8 character", async () => { + bridge = createNativeBridge(); + globalThis.cordova = { exec: (...args) => bridge.exec(...args) }; + const response = await system.httpStream(server.url("/utf8-split")); + const decoder = new TextDecoder(); + let text = ""; + for (const c of await collect(response)) { + text += decoder.decode(c, { stream: true }); + } + text += decoder.decode(); + expect(text).toBe("héllo \u00e9\u00e8\u00ea wörld"); + }); + + it("4. ACKs bytes only as the consumer reads them (bounded in flight)", async () => { + bridge = createNativeBridge(); + globalThis.cordova = { exec: (...args) => bridge.exec(...args) }; + + const response = await system.httpStream(server.url("/large")); + const reader = response.body.getReader(); + + const first = await reader.read(); + expect(first.done).toBe(false); + const firstLen = first.value.byteLength; + + // While the consumer stalls, JS stops granting credit, so the bridge + // reader must stall at its window instead of forwarding the whole 1 MiB. + await sleep(200); + const entry = bridge.streams.values().next().value; + expect(entry.maxPending).toBeLessThanOrEqual(CREDIT_WINDOW + BRIDGE_CHUNK); + expect(entry.forwarded).toBeLessThan(1024 * 1024); + // JS must never ACK more bytes than the consumer has actually read. + expect(entry.acked).toBeLessThanOrEqual(firstLen); + + // Resume consuming; the stream must drain to completion. + let total = firstLen; + while (true) { + const { done, value } = await reader.read(); + if (done) break; + total += value.byteLength; + } + expect(total).toBe(1024 * 1024); + expect(entry.acked).toBe(1024 * 1024); + expect(entry.maxPending).toBeLessThanOrEqual(CREDIT_WINDOW + BRIDGE_CHUNK); + }); + + it("5. keeps a 4xx/5xx with no error body a normal, empty response", async () => { + bridge = createNativeBridge(); + globalThis.cordova = { exec: (...args) => bridge.exec(...args) }; + const response = await system.httpStream(server.url("/status500-empty")); + expect(response.status).toBe(500); + const chunks = await collect(response); + expect(chunks).toHaveLength(0); + }); + + it("6. rejects with a transport error on connection failure before headers", async () => { + bridge = createNativeBridge(); + globalThis.cordova = { exec: (...args) => bridge.exec(...args) }; + const url = server.url("/network-error"); + const err = await system.httpStream(url).then(() => null, (e) => e); + expect(err).toBeInstanceOf(Error); + }); + + it("7. cancelling the reader cancels the underlying request and drops late events", async () => { + bridge = createNativeBridge(); + globalThis.cordova = { exec: (...args) => bridge.exec(...args) }; + + const response = await system.httpStream(server.url("/keep-open")); + const reader = response.body.getReader(); + const first = await reader.read(); + expect(first.done).toBe(false); + + await reader.cancel(); + + expect( + bridge.calls.some( + (c) => + c.action === "http-stream-cancel" && + typeof c.args[0] === "string", + ), + ).toBe(true); + + // The underlying request must have been torn down. + const entry = bridge.streams.values().next().value; + expect(entry.cancelled).toBe(true); + }); + + it("8. preserves repeated headers such as Set-Cookie", async () => { + bridge = createNativeBridge(); + globalThis.cordova = { exec: (...args) => bridge.exec(...args) }; + + const response = await system.httpStream(server.url("/cookies")); + expect(response.status).toBe(200); + const h = response.headers; + if (typeof h.getSetCookie === "function") { + expect(h.getSetCookie()).toEqual([ + "session=abc; Path=/", + "theme=dark; Path=/", + ]); + } else { + expect(h.get("set-cookie")).toBe( + "session=abc; Path=/, theme=dark; Path=/", + ); + } + await collect(response); + }); + + it("9. rejects without starting when the signal is already aborted", async () => { + bridge = createNativeBridge(); + globalThis.cordova = { exec: (...args) => bridge.exec(...args) }; + + const controller = new AbortController(); + controller.abort(); + + const err = await system + .httpStream(server.url("/simple"), { signal: controller.signal }) + .then(() => null, (e) => e); + expect(err).toBeInstanceOf(Error); + expect(err.name).toBe("AbortError"); + expect( + bridge.calls.some((c) => c.action === "http-stream-start"), + ).toBe(false); + }); + + it("10. aborting the signal cancels an in-flight stream and errors the body", async () => { + bridge = createNativeBridge(); + globalThis.cordova = { exec: (...args) => bridge.exec(...args) }; + + const controller = new AbortController(); + const response = await system.httpStream(server.url("/keep-open"), { + signal: controller.signal, + }); + const reader = response.body.getReader(); + const first = await reader.read(); + expect(first.done).toBe(false); + + controller.abort(); + + const err = await reader.read().then(() => null, (e) => e); + expect(err).toBeInstanceOf(Error); + expect(err.name).toBe("AbortError"); + + expect( + bridge.calls.some( + (c) => + c.action === "http-stream-cancel" && + typeof c.args[0] === "string", + ), + ).toBe(true); + const entry = bridge.streams.values().next().value; + expect(entry.cancelled).toBe(true); + }); + + it("11. ships text chunks raw (latin1 fast path, no base64) and decodes them", async () => { + bridge = createNativeBridge(); + globalThis.cordova = { exec: (...args) => bridge.exec(...args) }; + + const response = await system.httpStream(server.url("/latin1-text")); + const text = decode(await collect(response)); + expect(text).toBe('say "hi" \\ path \u00e9\u00ff\u2028 done'); + + const entry = bridge.streams.values().next().value; + const dataMsgs = entry.delivered.filter((m) => m.type === "data"); + expect(dataMsgs.length).toBeGreaterThan(0); + expect(dataMsgs.every((m) => !m.b64)).toBe(true); + }); + + it("12. falls back to base64 for chunks with control bytes and still decodes", async () => { + bridge = createNativeBridge(); + globalThis.cordova = { exec: (...args) => bridge.exec(...args) }; + + const expected = Buffer.from([ + 0x00, 0x08, 0x1f, 0x7f, 0xff, 0x22, 0x00, + ]); + const response = await system.httpStream(server.url("/binary-ctrl")); + const bytes = Buffer.from(concat(await collect(response))); + expect(bytes.equals(expected)).toBe(true); + + const entry = bridge.streams.values().next().value; + const dataMsgs = entry.delivered.filter((m) => m.type === "data"); + expect(dataMsgs.some((m) => m.b64)).toBe(true); + }); +});