The caller writes bytes into the {@link PipedOutputStream} via {@link #write(byte[], int)}
+ * while the DW engine reads from the paired {@link PipedInputStream} on a separate thread.
+ *
+ * Instances are stored in a static registry keyed by a monotonically increasing handle so that
+ * native callers can reference them across {@code @CEntryPoint} invocations.
+ */
+public class InputStreamSession {
+
+ private static final ConcurrentHashMap SESSIONS = new ConcurrentHashMap<>();
+ private static final AtomicLong NEXT_HANDLE = new AtomicLong(1);
+
+ private static final int PIPE_BUFFER_SIZE = 64 * 1024;
+
+ private final PipedInputStream pipedInputStream;
+ private final PipedOutputStream pipedOutputStream;
+ private final String mimeType;
+ private final String charset;
+
+ /**
+ * Creates a new input stream session with the given metadata.
+ *
+ * @param mimeType the MIME type of the input data
+ * @param charset the character set of the input data (may be {@code null}, defaults to UTF-8)
+ */
+ public InputStreamSession(String mimeType, String charset) {
+ try {
+ this.pipedInputStream = new PipedInputStream(PIPE_BUFFER_SIZE);
+ this.pipedOutputStream = new PipedOutputStream(pipedInputStream);
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to create piped streams", e);
+ }
+ this.mimeType = mimeType;
+ this.charset = charset != null ? charset : "UTF-8";
+ }
+
+ /**
+ * Registers this session and returns its handle.
+ *
+ * @return a unique handle that callers use to reference this session
+ */
+ public long register() {
+ long handle = NEXT_HANDLE.getAndIncrement();
+ SESSIONS.put(handle, this);
+ return handle;
+ }
+
+ /**
+ * Looks up a previously registered session.
+ *
+ * @param handle the handle returned by {@link #register()}
+ * @return the session, or {@code null} if not found
+ */
+ public static InputStreamSession get(long handle) {
+ return SESSIONS.get(handle);
+ }
+
+ /**
+ * Removes a session from the registry and closes both ends of the pipe.
+ *
+ * @param handle the session handle
+ */
+ public static void close(long handle) {
+ InputStreamSession session = SESSIONS.remove(handle);
+ if (session != null) {
+ try {
+ session.pipedOutputStream.close();
+ } catch (IOException ignored) {
+ }
+ try {
+ session.pipedInputStream.close();
+ } catch (IOException ignored) {
+ }
+ }
+ }
+
+ /**
+ * Writes bytes into the pipe. The DW engine will read these from the paired
+ * {@link PipedInputStream}.
+ *
+ * @param data the byte array to write from
+ * @param length the number of bytes to write
+ * @throws IOException if an I/O error occurs
+ */
+ public void write(byte[] data, int length) throws IOException {
+ pipedOutputStream.write(data, 0, length);
+ }
+
+ /**
+ * Closes the write end of the pipe, signalling EOF to the reader.
+ *
+ * @throws IOException if an I/O error occurs
+ */
+ public void closeWriter() throws IOException {
+ pipedOutputStream.close();
+ }
+
+ /**
+ * Returns the {@link PipedInputStream} that the DW engine should read from.
+ *
+ * @return the read end of the pipe
+ */
+ public PipedInputStream getInputStream() {
+ return pipedInputStream;
+ }
+
+ public String getMimeType() {
+ return mimeType;
+ }
+
+ public String getCharset() {
+ return charset;
+ }
+}
diff --git a/native-lib/src/main/java/org/mule/weave/lib/NativeCallbacks.java b/native-lib/src/main/java/org/mule/weave/lib/NativeCallbacks.java
new file mode 100644
index 0000000..3872354
--- /dev/null
+++ b/native-lib/src/main/java/org/mule/weave/lib/NativeCallbacks.java
@@ -0,0 +1,49 @@
+package org.mule.weave.lib;
+
+import org.graalvm.nativeimage.c.function.CFunctionPointer;
+import org.graalvm.nativeimage.c.function.InvokeCFunctionPointer;
+import org.graalvm.nativeimage.c.type.CCharPointer;
+import org.graalvm.word.PointerBase;
+
+/**
+ * Function-pointer (callback) interfaces used by the callback-based streaming API in
+ * {@link NativeLib}.
+ *
+ * FFI callers pass C function pointers that conform to these signatures. The Java side
+ * invokes them via {@link InvokeCFunctionPointer} to push/pull data without requiring
+ * the session-based open/read|write/close round-trips.
+ */
+public final class NativeCallbacks {
+
+ private NativeCallbacks() {
+ }
+
+ /**
+ * Callback the native caller provides to receive output data.
+ *
+ * Signature in C:
+ *
{@code int (*WriteCallback)(void *ctx, const char *buffer, int length);}
+ *
+ * The Java side calls this repeatedly with chunks of the script result.
+ * The callback must return {@code 0} on success or a non-zero value to abort.
+ */
+ public interface WriteCallback extends CFunctionPointer {
+ @InvokeCFunctionPointer
+ int invoke(PointerBase ctx, CCharPointer buffer, int length);
+ }
+
+ /**
+ * Callback the native caller provides to supply input data.
+ *
+ * Signature in C:
+ *
{@code int (*ReadCallback)(void *ctx, char *buffer, int bufferSize);}
+ *
+ * The Java side calls this to pull the next chunk of input bytes. The callback must
+ * write up to {@code bufferSize} bytes into {@code buffer} and return the number of
+ * bytes written, {@code 0} on EOF, or {@code -1} on error.
+ */
+ public interface ReadCallback extends CFunctionPointer {
+ @InvokeCFunctionPointer
+ int invoke(PointerBase ctx, CCharPointer buffer, int bufferSize);
+ }
+}
diff --git a/native-lib/src/main/java/org/mule/weave/lib/NativeLib.java b/native-lib/src/main/java/org/mule/weave/lib/NativeLib.java
new file mode 100644
index 0000000..de776e7
--- /dev/null
+++ b/native-lib/src/main/java/org/mule/weave/lib/NativeLib.java
@@ -0,0 +1,333 @@
+package org.mule.weave.lib;
+
+import org.graalvm.nativeimage.IsolateThread;
+import org.graalvm.nativeimage.UnmanagedMemory;
+import org.graalvm.nativeimage.c.function.CEntryPoint;
+import org.graalvm.nativeimage.c.type.CCharPointer;
+import org.graalvm.nativeimage.c.type.CTypeConversion;
+import org.graalvm.word.PointerBase;
+import org.graalvm.word.WordFactory;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+
+/**
+ * GraalVM native entry points exposed for FFI consumers.
+ *
+ * This class provides C-callable functions to execute DataWeave scripts and to free the returned
+ * unmanaged strings.
+ */
+public class NativeLib {
+
+ /**
+ * Native method that executes a DataWeave script with inputs and returns the result.
+ * Can be called from Python via FFI.
+ *
+ * @param thread the isolate thread (automatically provided by GraalVM)
+ * @param script the DataWeave script to execute (C string pointer)
+ * @param inputsJson JSON string containing the inputs map with content (base64 encoded), mimeType, properties and charset for each binding
+ * @return the script execution result base64 encoded (C string pointer)
+ */
+ @CEntryPoint(name = "run_script")
+ public static CCharPointer runDwScriptEncoded(IsolateThread thread, CCharPointer script, CCharPointer inputsJson) {
+ String dwScript = CTypeConversion.toJavaString(script);
+ String inputs = CTypeConversion.toJavaString(inputsJson);
+
+ ScriptRuntime runtime = ScriptRuntime.getInstance();
+ String result = runtime.run(dwScript, inputs);
+ return toUnmanagedCString(result);
+ }
+
+ /**
+ * Frees a C string previously returned by {@link #runDwScriptEncoded(IsolateThread, CCharPointer, CCharPointer)}.
+ *
+ * @param thread the isolate thread (automatically provided by GraalVM)
+ * @param pointer the pointer to the unmanaged C string to free; if null, this is a no-op
+ */
+ @CEntryPoint(name = "free_cstring")
+ public static void freeCString(IsolateThread thread, CCharPointer pointer) {
+ if (pointer.isNull()) {
+ return;
+ }
+ UnmanagedMemory.free(pointer);
+ }
+
+ // ── Callback-based Streaming API ─────────────────────────────────────
+
+ private static final int CALLBACK_BUFFER_SIZE = 8 * 1024;
+
+ /**
+ * Executes a DataWeave script and streams the result to a caller-supplied write callback.
+ *
+ * Instead of the session-based open/read/close cycle, the caller passes a
+ * {@code WriteCallback} function pointer. The Java side reads the output stream in chunks
+ * and invokes the callback for each chunk until the stream is exhausted.
+ *
+ * The returned C string is a JSON object with the execution metadata:
+ *
+ * - On success: {@code {"success":true,"mimeType":"...","charset":"...","binary":true/false}}
+ * - On error: {@code {"success":false,"error":"..."}}
+ *
+ * The caller must free the returned pointer with {@link #freeCString}.
+ *
+ * @param thread the isolate thread
+ * @param script the DataWeave script (C string)
+ * @param inputsJson JSON-encoded inputs map (C string), may be null
+ * @param writeCallback function pointer invoked with each output chunk; must return 0 on success
+ * @param ctx opaque context pointer forwarded to every callback invocation
+ * @return an unmanaged C string with JSON metadata/error
+ */
+ @CEntryPoint(name = "run_script_callback")
+ public static CCharPointer runScriptCallback(
+ IsolateThread thread,
+ CCharPointer script,
+ CCharPointer inputsJson,
+ NativeCallbacks.WriteCallback writeCallback,
+ PointerBase ctx) {
+
+ String dwScript = CTypeConversion.toJavaString(script);
+ String inputs = inputsJson.isNull() ? null : CTypeConversion.toJavaString(inputsJson);
+
+ ScriptRuntime runtime = ScriptRuntime.getInstance();
+ StreamSession session = runtime.runStreaming(dwScript, inputs);
+
+ if (session.isError()) {
+ return toUnmanagedCString("{\"success\":false,\"error\":\""
+ + escapeJsonString(session.getError()) + "\"}");
+ }
+
+ try {
+ byte[] buf = new byte[CALLBACK_BUFFER_SIZE];
+ CCharPointer nativeBuf = UnmanagedMemory.malloc(CALLBACK_BUFFER_SIZE);
+ try {
+ int n;
+ while ((n = session.read(buf, buf.length)) > 0) {
+ for (int i = 0; i < n; i++) {
+ nativeBuf.write(i, buf[i]);
+ }
+ int rc = writeCallback.invoke(ctx, nativeBuf, n);
+ if (rc != 0) {
+ return toUnmanagedCString("{\"success\":false,\"error\":\""
+ + "Write callback returned error: " + rc + "\"}");
+ }
+ }
+ } finally {
+ UnmanagedMemory.free(nativeBuf);
+ }
+ } catch (IOException e) {
+ return toUnmanagedCString("{\"success\":false,\"error\":\""
+ + escapeJsonString(e.getMessage()) + "\"}");
+ } finally {
+ session.closeStream();
+ }
+
+ return toUnmanagedCString("{\"success\":true"
+ + ",\"mimeType\":\"" + session.getMimeType() + "\""
+ + ",\"charset\":\"" + session.getCharset() + "\""
+ + ",\"binary\":" + session.isBinary()
+ + "}");
+ }
+
+ /**
+ * Executes a DataWeave script whose output is streamed via a write callback, and whose
+ * input named {@code inputName} is fed via a read callback.
+ *
+ * The read callback is invoked on a background thread to pull input data while the
+ * output is pushed to the write callback on the calling thread. This allows fully
+ * callback-driven input and output streaming in a single call.
+ *
+ * The returned C string follows the same JSON schema as
+ * {@link #runScriptCallback}.
+ *
+ * @param thread the isolate thread
+ * @param script the DataWeave script (C string)
+ * @param inputsJson JSON-encoded inputs map (C string), may be null; entries for
+ * {@code inputName} are ignored since the read callback supplies that input
+ * @param inputName the binding name for the callback-supplied input (C string)
+ * @param inputMimeType the MIME type of the callback-supplied input (C string)
+ * @param inputCharset the charset of the callback-supplied input (C string), may be null for UTF-8
+ * @param readCallback function pointer invoked to read the next chunk; must return bytes written,
+ * 0 on EOF, or -1 on error
+ * @param writeCallback function pointer invoked with each output chunk; must return 0 on success
+ * @param ctx opaque context pointer forwarded to every callback invocation
+ * @return an unmanaged C string with JSON metadata/error
+ */
+ @CEntryPoint(name = "run_script_input_output_callback")
+ public static CCharPointer runScriptInputOutputCallback(
+ IsolateThread thread,
+ CCharPointer script,
+ CCharPointer inputsJson,
+ CCharPointer inputName,
+ CCharPointer inputMimeType,
+ CCharPointer inputCharset,
+ NativeCallbacks.ReadCallback readCallback,
+ NativeCallbacks.WriteCallback writeCallback,
+ PointerBase ctx) {
+
+ String dwScript = CTypeConversion.toJavaString(script);
+ String inputs = inputsJson.isNull() ? null : CTypeConversion.toJavaString(inputsJson);
+ String inName = CTypeConversion.toJavaString(inputName);
+ String inMime = CTypeConversion.toJavaString(inputMimeType);
+ String inCharset = inputCharset.isNull() ? null : CTypeConversion.toJavaString(inputCharset);
+
+ // Create a piped input stream session for the callback-supplied input
+ InputStreamSession inputSession = new InputStreamSession(inMime, inCharset);
+ long inputHandle = inputSession.register();
+
+ // Merge the stream handle into the inputs JSON
+ String streamEntry = "{\"streamHandle\":\"" + inputHandle + "\",\"mimeType\":\"" + inMime + "\""
+ + (inCharset != null ? ",\"charset\":\"" + inCharset + "\"" : "") + "}";
+ String mergedInputs = mergeInputEntry(inputs, inName, streamEntry);
+
+ // Start a background thread that calls the readCallback and feeds data into the pipe.
+ // Word types (CCharPointer, CFunctionPointer, PointerBase) cannot be captured in
+ // lambdas in GraalVM Native Image, so we use an explicit Runnable that stores their
+ // raw addresses and reconstitutes them via WordFactory.
+ final long readCallbackAddr = readCallback.rawValue();
+ final long ctxAddr = ctx.rawValue();
+ Thread feeder = new Thread(new InputCallbackFeeder(
+ readCallbackAddr, ctxAddr, inputSession), "dw-input-callback-feeder");
+ feeder.setDaemon(true);
+ feeder.start();
+
+ // Execute the script and stream output via the writeCallback
+ ScriptRuntime runtime = ScriptRuntime.getInstance();
+ StreamSession session = runtime.runStreaming(dwScript, mergedInputs);
+
+ if (session.isError()) {
+ cleanupFeeder(feeder, inputHandle);
+ return toUnmanagedCString("{\"success\":false,\"error\":\""
+ + escapeJsonString(session.getError()) + "\"}");
+ }
+
+ try {
+ byte[] buf = new byte[CALLBACK_BUFFER_SIZE];
+ CCharPointer writeBuf = UnmanagedMemory.malloc(CALLBACK_BUFFER_SIZE);
+ try {
+ int n;
+ while ((n = session.read(buf, buf.length)) > 0) {
+ for (int i = 0; i < n; i++) {
+ writeBuf.write(i, buf[i]);
+ }
+ int rc = writeCallback.invoke(ctx, writeBuf, n);
+ if (rc != 0) {
+ cleanupFeeder(feeder, inputHandle);
+ return toUnmanagedCString("{\"success\":false,\"error\":\""
+ + "Write callback returned error: " + rc + "\"}");
+ }
+ }
+ } finally {
+ UnmanagedMemory.free(writeBuf);
+ }
+ } catch (IOException e) {
+ cleanupFeeder(feeder, inputHandle);
+ return toUnmanagedCString("{\"success\":false,\"error\":\""
+ + escapeJsonString(e.getMessage()) + "\"}");
+ } finally {
+ session.closeStream();
+ }
+
+ cleanupFeeder(feeder, inputHandle);
+
+ return toUnmanagedCString("{\"success\":true"
+ + ",\"mimeType\":\"" + session.getMimeType() + "\""
+ + ",\"charset\":\"" + session.getCharset() + "\""
+ + ",\"binary\":" + session.isBinary()
+ + "}");
+ }
+
+ /**
+ * Merges a single input entry into an existing JSON inputs string.
+ */
+ private static String mergeInputEntry(String existingJson, String name, String entryJson) {
+ org.json.JSONObject obj = (existingJson == null || existingJson.trim().isEmpty())
+ ? new org.json.JSONObject()
+ : new org.json.JSONObject(existingJson);
+ obj.put(name, new org.json.JSONObject(entryJson));
+ return obj.toString();
+ }
+
+ /**
+ * Waits for the feeder thread to finish and closes the input session.
+ */
+ private static void cleanupFeeder(Thread feeder, long inputHandle) {
+ try {
+ feeder.join(5000);
+ } catch (InterruptedException ignored) {
+ }
+ InputStreamSession.close(inputHandle);
+ }
+
+ /**
+ * Explicit {@link Runnable} that drives the read-callback loop on a background thread.
+ *
+ * GraalVM Native Image forbids capturing {@code Word} types (such as
+ * {@link CCharPointer} or {@link CFunctionPointer}) inside lambdas. This class stores
+ * the raw addresses as plain {@code long} values and reconstitutes the pointers via
+ * {@link WordFactory#pointer(long)} inside {@link #run()}.
+ *
+ * The feeder allocates its own native read buffer and frees it in its {@code finally}
+ * block, ensuring no shared native memory between threads.
+ */
+ private static final class InputCallbackFeeder implements Runnable {
+ private final long readCallbackAddr;
+ private final long ctxAddr;
+ private final InputStreamSession inputSession;
+
+ InputCallbackFeeder(long readCallbackAddr, long ctxAddr,
+ InputStreamSession inputSession) {
+ this.readCallbackAddr = readCallbackAddr;
+ this.ctxAddr = ctxAddr;
+ this.inputSession = inputSession;
+ }
+
+ @Override
+ public void run() {
+ NativeCallbacks.ReadCallback cb = WordFactory.pointer(readCallbackAddr);
+ PointerBase ctx = WordFactory.pointer(ctxAddr);
+ CCharPointer buf = UnmanagedMemory.malloc(CALLBACK_BUFFER_SIZE);
+ try {
+ while (true) {
+ int n = cb.invoke(ctx, buf, CALLBACK_BUFFER_SIZE);
+ if (n <= 0) {
+ break; // 0 = EOF, negative = error
+ }
+ byte[] tmp = new byte[n];
+ for (int i = 0; i < n; i++) {
+ tmp[i] = buf.read(i);
+ }
+ inputSession.write(tmp, n);
+ }
+ } catch (IOException e) {
+ // pipe broken – DW engine will see the error
+ } finally {
+ UnmanagedMemory.free(buf);
+ try {
+ inputSession.closeWriter();
+ } catch (IOException ignored) {
+ }
+ }
+ }
+ }
+
+ private static String escapeJsonString(String input) {
+ if (input == null) return "";
+ return input
+ .replace("\\", "\\\\")
+ .replace("\"", "\\\"")
+ .replace("\n", "\\n")
+ .replace("\r", "\\r")
+ .replace("\t", "\\t");
+ }
+
+ private static CCharPointer toUnmanagedCString(String value) {
+ byte[] bytes = value.getBytes(StandardCharsets.UTF_8);
+ CCharPointer ptr = UnmanagedMemory.malloc(bytes.length + 1);
+ for (int i = 0; i < bytes.length; i++) {
+ ptr.write(i, bytes[i]);
+ }
+ ptr.write(bytes.length, (byte) 0);
+ return ptr;
+ }
+
+}
diff --git a/native-lib/src/main/java/org/mule/weave/lib/ScriptRuntime.java b/native-lib/src/main/java/org/mule/weave/lib/ScriptRuntime.java
new file mode 100644
index 0000000..9cb4007
--- /dev/null
+++ b/native-lib/src/main/java/org/mule/weave/lib/ScriptRuntime.java
@@ -0,0 +1,234 @@
+package org.mule.weave.lib;
+
+import org.json.JSONObject;
+import org.mule.weave.v2.runtime.BindingValue;
+import org.mule.weave.v2.runtime.DataWeaveResult;
+import org.mule.weave.v2.runtime.ScriptingBindings;
+import org.mule.weave.v2.runtime.api.DWResult;
+import org.mule.weave.v2.runtime.api.DWScript;
+import org.mule.weave.v2.runtime.api.DWScriptingEngine;
+import scala.Option;
+import scala.Tuple2;
+import scala.collection.immutable.Map;
+import scala.collection.immutable.Map$;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.charset.Charset;
+import java.util.Base64;
+
+/**
+ * Singleton wrapper around a {@link DWScriptingEngine} used to compile and execute DataWeave scripts.
+ *
+ * Execution results are returned as a JSON string containing a base64-encoded payload plus metadata
+ * (mime type, charset, and whether the result is binary). Errors are returned as a JSON string with
+ * {@code success=false} and an escaped error message.
+ */
+public class ScriptRuntime {
+
+ private static final ScriptRuntime INSTANCE = new ScriptRuntime();
+
+ /**
+ * Returns the singleton instance.
+ *
+ * @return the shared {@link ScriptRuntime}
+ */
+ public static ScriptRuntime getInstance() {
+ return INSTANCE;
+ }
+
+ private DWScriptingEngine engine;
+
+ private ScriptRuntime() {
+ engine = DWScriptingEngine.builder().build();
+ }
+
+ /**
+ * Executes a DataWeave script with no input bindings.
+ *
+ * @param script the DataWeave script source
+ * @return a JSON string describing either the successful result or an error
+ */
+ public String run(String script) {
+ return run(script, null);
+ }
+
+ /**
+ * Executes a DataWeave script with optional input bindings encoded as JSON.
+ *
+ * The expected JSON structure maps binding names to an object containing {@code content}
+ * (base64), {@code mimeType}, optional {@code charset}, and optional {@code properties}.
+ *
+ * @param script the DataWeave script source
+ * @param inputsJson JSON string encoding the input bindings map, or {@code null}
+ * @return a JSON string describing either the successful result or an error
+ */
+ public String run(String script, String inputsJson) {
+ ScriptingBindings bindings = parseJsonInputsToBindings(inputsJson);
+ String[] inputs = bindings.bindingNames();
+
+ try {
+ DWScript compiled = engine.compileDWScript(script, inputs);
+ DWResult dwResult = compiled.writeDWResult(bindings);
+
+ String encodedResult;
+ if (dwResult.getContent() instanceof InputStream) {
+ try {
+ byte[] ba = ((InputStream) dwResult.getContent()).readAllBytes();
+ encodedResult = Base64.getEncoder().encodeToString(ba);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ } else {
+ throw new RuntimeException("Result is not an InputStream: " + dwResult.getContent().getClass().getName());
+ }
+
+ return "{"
+ + "\"success\":true,"
+ + "\"result\":\"" + encodedResult + "\","
+ + "\"mimeType\":\"" + dwResult.getMimeType() + "\","
+ + "\"charset\":\"" + dwResult.getCharset() + "\","
+ + "\"binary\":" + ((DataWeaveResult) dwResult).isBinary()
+ + "}";
+ } catch (Exception e) {
+ String message = e.getMessage();
+ if (message == null || message.trim().isEmpty()) {
+ message = e.toString();
+ }
+
+ return "{"
+ + "\"success\":false,"
+ + "\"error\":\"" + escapeJsonString(message) + "\""
+ + "}";
+ }
+ }
+
+ /**
+ * Executes a DataWeave script and returns a {@link StreamSession} whose {@link java.io.InputStream}
+ * can be read incrementally, avoiding loading the entire result into memory.
+ *
+ * @param script the DataWeave script source
+ * @param inputsJson JSON string encoding the input bindings map, or {@code null}
+ * @return a {@link StreamSession} with the result stream and metadata, or an error session
+ */
+ public StreamSession runStreaming(String script, String inputsJson) {
+ ScriptingBindings bindings = parseJsonInputsToBindings(inputsJson);
+ String[] inputs = bindings.bindingNames();
+
+ try {
+ DWScript compiled = engine.compileDWScript(script, inputs);
+ DWResult dwResult = compiled.writeDWResult(bindings);
+
+ if (dwResult.getContent() instanceof InputStream) {
+ return new StreamSession(
+ (InputStream) dwResult.getContent(),
+ dwResult.getMimeType(),
+ dwResult.getCharset().name(),
+ ((DataWeaveResult) dwResult).isBinary()
+ );
+ } else {
+ return StreamSession.ofError("Result is not an InputStream: " + dwResult.getContent().getClass().getName());
+ }
+ } catch (Exception e) {
+ String message = e.getMessage();
+ if (message == null || message.trim().isEmpty()) {
+ message = e.toString();
+ }
+ return StreamSession.ofError(message);
+ }
+ }
+
+ private ScriptingBindings parseJsonInputsToBindings(String inputsJson) {
+ ScriptingBindings bindings = new ScriptingBindings();
+
+ if (inputsJson == null || inputsJson.trim().isEmpty()) {
+ return bindings;
+ }
+
+ try {
+ JSONObject root = new JSONObject(inputsJson);
+
+ for (String name : root.keySet()) {
+ JSONObject entry = root.getJSONObject(name);
+
+ if (entry.has("streamHandle")) {
+ long streamHandle = Long.parseLong(entry.getString("streamHandle"));
+ InputStreamSession inputSession = InputStreamSession.get(streamHandle);
+ if (inputSession == null) {
+ throw new RuntimeException("Invalid streamHandle " + streamHandle + " for input '" + name + "'");
+ }
+ String mimeTypeRaw = entry.optString("mimeType", inputSession.getMimeType());
+ String charsetRaw = entry.optString("charset", inputSession.getCharset());
+ Charset charset = Charset.forName(charsetRaw);
+ Option mimeType = Option.apply(mimeTypeRaw);
+
+ BindingValue bindingValue = new BindingValue(inputSession.getInputStream(), mimeType, Map$.MODULE$.empty(), charset);
+ bindings.addBinding(name, bindingValue);
+
+ } else if (entry.has("content")) {
+ String contentRaw = entry.getString("content");
+ String mimeTypeRaw = entry.optString("mimeType", null);
+ String charsetRaw = entry.optString("charset", "UTF-8");
+
+ Map properties = Map$.MODULE$.empty();
+ if (entry.has("properties") && !entry.isNull("properties")) {
+ JSONObject propsObj = entry.getJSONObject("properties");
+ properties = parseJsonProperties(propsObj);
+ }
+
+ Charset charset = Charset.forName(charsetRaw);
+ Option mimeType = Option.apply(mimeTypeRaw);
+
+ byte[] content = Base64.getDecoder().decode(contentRaw);
+ BindingValue bindingValue = new BindingValue(content, mimeType, properties, charset);
+ bindings.addBinding(name, bindingValue);
+ }
+ }
+ } catch (Exception e) {
+ System.err.println("Error parsing JSON inputs: " + e.getMessage());
+ e.printStackTrace();
+ }
+
+ return bindings;
+ }
+
+ @SuppressWarnings("unchecked")
+ private Map parseJsonProperties(JSONObject propsObj) {
+ Map result = Map$.MODULE$.empty();
+
+ for (String key : propsObj.keySet()) {
+ Object val = propsObj.get(key);
+ if (val instanceof String || val instanceof Boolean) {
+ result = (Map) result.$plus(new Tuple2<>(key, val));
+ } else if (val instanceof Number) {
+ Number num = (Number) val;
+ Object boxed;
+ if (val instanceof Double || val instanceof Float) {
+ boxed = num.doubleValue();
+ } else {
+ boxed = num.longValue();
+ }
+ result = (Map) result.$plus(new Tuple2<>(key, boxed));
+ } else if (val == JSONObject.NULL) {
+ throw new IllegalArgumentException("properties values cannot be null (key '" + key + "')");
+ } else {
+ throw new IllegalArgumentException("properties values must be primitive (string/number/boolean) (key '" + key + "')");
+ }
+ }
+
+ return result;
+ }
+
+ private String escapeJsonString(String input) {
+ if (input == null) {
+ return "";
+ }
+
+ return input
+ .replace("\\", "\\\\")
+ .replace("\"", "\\\"")
+ .replace("\n", "\\n")
+ .replace("\r", "\\r")
+ .replace("\t", "\\t");
+ }
+}
diff --git a/native-lib/src/main/java/org/mule/weave/lib/StreamSession.java b/native-lib/src/main/java/org/mule/weave/lib/StreamSession.java
new file mode 100644
index 0000000..b54f299
--- /dev/null
+++ b/native-lib/src/main/java/org/mule/weave/lib/StreamSession.java
@@ -0,0 +1,136 @@
+package org.mule.weave.lib;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * Holds an open {@link InputStream} and associated metadata for a streaming script execution.
+ *
+ * Instances are stored in a static registry keyed by a monotonically increasing handle so that
+ * native callers can reference them across {@code @CEntryPoint} invocations.
+ */
+public class StreamSession {
+
+ private static final ConcurrentHashMap SESSIONS = new ConcurrentHashMap<>();
+ private static final AtomicLong NEXT_HANDLE = new AtomicLong(1);
+
+ private final InputStream inputStream;
+ private final String mimeType;
+ private final String charset;
+ private final boolean binary;
+ private final String error;
+
+ StreamSession(InputStream inputStream, String mimeType, String charset, boolean binary) {
+ this.inputStream = inputStream;
+ this.mimeType = mimeType;
+ this.charset = charset;
+ this.binary = binary;
+ this.error = null;
+ }
+
+ private StreamSession(String error) {
+ this.inputStream = null;
+ this.mimeType = null;
+ this.charset = null;
+ this.binary = false;
+ this.error = error;
+ }
+
+ /**
+ * Creates an error session that carries only an error message.
+ *
+ * @param error the error message
+ * @return an error session
+ */
+ public static StreamSession ofError(String error) {
+ return new StreamSession(error);
+ }
+
+ /**
+ * Registers this session and returns its handle.
+ *
+ * @return a unique handle that callers use to reference this session
+ */
+ public long register() {
+ long handle = NEXT_HANDLE.getAndIncrement();
+ SESSIONS.put(handle, this);
+ return handle;
+ }
+
+ /**
+ * Looks up a previously registered session.
+ *
+ * @param handle the handle returned by {@link #register()}
+ * @return the session, or {@code null} if not found
+ */
+ public static StreamSession get(long handle) {
+ return SESSIONS.get(handle);
+ }
+
+ /**
+ * Removes and closes a session.
+ *
+ * @param handle the session handle
+ */
+ public static void close(long handle) {
+ StreamSession session = SESSIONS.remove(handle);
+ if (session != null && session.inputStream != null) {
+ try {
+ session.inputStream.close();
+ } catch (IOException ignored) {
+ }
+ }
+ }
+
+ /**
+ * Reads up to {@code len} bytes into the provided byte array.
+ *
+ * @param buf destination buffer
+ * @param len maximum number of bytes to read
+ * @return number of bytes actually read, or {@code -1} on EOF
+ * @throws IOException if an I/O error occurs
+ */
+ public int read(byte[] buf, int len) throws IOException {
+ return inputStream.read(buf, 0, len);
+ }
+
+ public String getMimeType() {
+ return mimeType;
+ }
+
+ public String getCharset() {
+ return charset;
+ }
+
+ public boolean isBinary() {
+ return binary;
+ }
+
+ /**
+ * Closes the underlying input stream. Safe to call on error sessions (no-op).
+ */
+ public void closeStream() {
+ if (inputStream != null) {
+ try {
+ inputStream.close();
+ } catch (IOException ignored) {
+ }
+ }
+ }
+
+ /**
+ * Returns the error message if this is an error session, or {@code null} otherwise.
+ */
+ public String getError() {
+ return error;
+ }
+
+ /**
+ * Returns {@code true} if this session represents a failed execution.
+ */
+ public boolean isError() {
+ return error != null;
+ }
+}
diff --git a/native-lib/src/main/resources/META-INF/services/org.mule.weave.v2.module.DataFormat b/native-lib/src/main/resources/META-INF/services/org.mule.weave.v2.module.DataFormat
new file mode 100644
index 0000000..b1a22df
--- /dev/null
+++ b/native-lib/src/main/resources/META-INF/services/org.mule.weave.v2.module.DataFormat
@@ -0,0 +1,9 @@
+org.mule.weave.v2.interpreted.module.WeaveDataFormat
+org.mule.weave.v2.module.core.json.JsonDataFormat
+org.mule.weave.v2.module.core.xml.XmlDataFormat
+org.mule.weave.v2.module.core.csv.CSVDataFormat
+org.mule.weave.v2.module.core.octetstream.OctetStreamDataFormat
+org.mule.weave.v2.module.core.textplain.TextPlainDataFormat
+org.mule.weave.v2.module.core.urlencoded.UrlEncodedDataFormat
+org.mule.weave.v2.module.core.multipart.MultiPartDataFormat
+org.mule.weave.v2.module.core.properties.PropertiesDataFormat
\ No newline at end of file
diff --git a/native-lib/src/main/resources/META-INF/services/org.mule.weave.v2.parser.phase.ModuleLoader b/native-lib/src/main/resources/META-INF/services/org.mule.weave.v2.parser.phase.ModuleLoader
new file mode 100644
index 0000000..ef3215b
--- /dev/null
+++ b/native-lib/src/main/resources/META-INF/services/org.mule.weave.v2.parser.phase.ModuleLoader
@@ -0,0 +1 @@
+org.mule.weave.v2.compilation.loader.WeaveBinaryResourceModuleLoader
\ No newline at end of file
diff --git a/native-lib/src/test/java/org/mule/weave/lib/ScriptRuntimeTest.java b/native-lib/src/test/java/org/mule/weave/lib/ScriptRuntimeTest.java
new file mode 100644
index 0000000..70f8044
--- /dev/null
+++ b/native-lib/src/test/java/org/mule/weave/lib/ScriptRuntimeTest.java
@@ -0,0 +1,613 @@
+package org.mule.weave.lib;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.nio.charset.Charset;
+import java.util.Base64;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicReference;
+
+class ScriptRuntimeTest {
+
+ @Test
+ void runSimpleScript() {
+ ScriptRuntime runtime = ScriptRuntime.getInstance();
+
+ System.out.println("Running sqrt(144) 10 times with timing:");
+ System.out.println("=".repeat(50));
+
+ for (int i = 1; i <= 20; i++) {
+ long startTime = System.nanoTime();
+ String result = runtime.run("sqrt(144)");
+ long endTime = System.nanoTime();
+ double executionTimeMs = (endTime - startTime) / 1_000_000.0;
+
+ assertEquals("12", Result.parse(result).result);
+ System.out.printf("Run %2d: %.3f ms - Result: %s%n", i, executionTimeMs, result);
+ }
+
+ System.out.println("=".repeat(50));
+ }
+
+ @Test
+ void runParseError() {
+ ScriptRuntime runtime = ScriptRuntime.getInstance();
+
+ System.out.println("Running sqrt(144) 10 times with timing:");
+ System.out.println("=".repeat(50));
+
+ String result = runtime.run("invalid syntax here");
+
+ String error = Result.parse(result).error;
+ assertTrue(error.contains("Unable to resolve reference"));
+ System.out.printf("Error: %s%n", result);
+
+ System.out.println("=".repeat(50));
+ }
+
+ @Test
+ void runWithInputs() {
+ ScriptRuntime runtime = ScriptRuntime.getInstance();
+
+ System.out.println("Testing runWithInputs with two integer numbers:");
+ System.out.println("=".repeat(50));
+
+ // Test 1: Sum 25 + 17
+ int num1 = 25;
+ int num2 = 17;
+ int expected = num1 + num2;
+
+ // Create inputs JSON with content and mimeType for each binding
+ String inputsJson = String.format(
+ "{\"num1\": {\"content\": \"%s\", \"mimeType\": \"application/json\"}, " +
+ "\"num2\": {\"content\": \"%s\", \"mimeType\": \"application/json\"}}",
+ encode(num1), encode(num2)
+ );
+
+ String script = "num1 + num2";
+
+ System.out.printf("Test 1: %d + %d%n", num1, num2);
+ System.out.printf("Script: %s%n", script);
+ System.out.printf("Inputs: %s%n", inputsJson);
+
+ long startTime = System.nanoTime();
+ String result = Result.parse(runtime.run(script, inputsJson)).result;
+ long endTime = System.nanoTime();
+ double executionTimeMs = (endTime - startTime) / 1_000_000.0;
+
+ System.out.printf("Result: %s%n", result);
+ System.out.printf("Expected: %d%n", expected);
+ System.out.printf("Execution time: %.3f ms%n", executionTimeMs);
+
+ assertEquals(String.valueOf(expected), result);
+ System.out.println("✓ Test 1 passed!");
+
+ System.out.println("-".repeat(50));
+
+ // Test 2: Sum 100 + 250
+ num1 = 100;
+ num2 = 250;
+ expected = num1 + num2;
+
+ inputsJson = String.format(
+ "{\"num1\": {\"content\": \"%s\", \"mimeType\": \"application/json\"}, " +
+ "\"num2\": {\"content\": \"%s\", \"mimeType\": \"application/json\"}}",
+ encode(num1), encode(num2)
+ );
+
+ System.out.printf("Test 2: %d + %d%n", num1, num2);
+ System.out.printf("Script: %s%n", script);
+
+ startTime = System.nanoTime();
+ result = Result.parse(runtime.run(script, inputsJson)).result;
+ endTime = System.nanoTime();
+ executionTimeMs = (endTime - startTime) / 1_000_000.0;
+
+ System.out.printf("Result: %s%n", result);
+ System.out.printf("Expected: %d%n", expected);
+ System.out.printf("Execution time: %.3f ms%n", executionTimeMs);
+
+ assertEquals(String.valueOf(expected), result);
+ System.out.println("✓ Test 2 passed!");
+
+ System.out.println("=".repeat(50));
+ }
+
+ private String encode(Object value) {
+ byte[] bytes = value instanceof byte[] ? (byte[]) value : String.valueOf(value).getBytes();
+ return Base64.getEncoder().encodeToString(bytes);
+
+ }
+
+ @Test
+ void runWithXmlInput() {
+ ScriptRuntime runtime = ScriptRuntime.getInstance();
+
+ System.out.println("Testing runWithInputs with XML input to calculate average age:");
+ System.out.println("=".repeat(50));
+
+ // XML input with two people
+ String xmlInput = """
+
+
+ 19
+ john
+
+
+ 25
+ jane
+
+
+ """;
+
+ String inputsJson = String.format(
+ "{\"people\": {\"content\": \"%s\", \"mimeType\": \"application/xml\"}}",
+ encode(xmlInput)
+ );
+
+ // DataWeave script to calculate average age
+ String script = """
+ output application/json
+ ---
+ avg(people.people.*person.age)
+ """;
+
+ System.out.printf("XML Input:%n%s%n", xmlInput);
+ System.out.printf("Script:%n%s%n", script);
+
+ long startTime = System.nanoTime();
+ String result = runtime.run(script, inputsJson);
+ long endTime = System.nanoTime();
+ double executionTimeMs = (endTime - startTime) / 1_000_000.0;
+
+ System.out.printf("Result: %s%n", result);
+ System.out.printf("Expected: 22 (average of 19 and 25)%n");
+ System.out.printf("Execution time: %.3f ms%n", executionTimeMs);
+
+ // The average of 19 and 25 is 22
+ assertEquals("22", Result.parse(result).result);
+ System.out.println("✓ Test passed!");
+
+ System.out.println("=".repeat(50));
+ }
+
+ @Test
+ void runWithJsonObjectInput() {
+ ScriptRuntime runtime = ScriptRuntime.getInstance();
+
+ System.out.println("Testing runWithInputs with JSON object input:");
+ System.out.println("=".repeat(50));
+
+ String jsonInput = "{\"name\": \"John\", \"age\": 30}";
+
+ String inputsJson = String.format(
+ "{\"payload\": {\"content\": \"%s\", \"mimeType\": \"application/json\"}}",
+ encode(jsonInput)
+ );
+
+ // DataWeave script to extract name
+ String script = "output application/json\n---\npayload.name";
+
+ System.out.printf("JSON Input: %s%n", jsonInput);
+ System.out.printf("Script: %s%n", script);
+
+ long startTime = System.nanoTime();
+ String result = Result.parse(runtime.run(script, inputsJson)).result;
+ long endTime = System.nanoTime();
+ double executionTimeMs = (endTime - startTime) / 1_000_000.0;
+
+ System.out.printf("Result: %s%n", result);
+ System.out.printf("Expected: \"John\"%n");
+ System.out.printf("Execution time: %.3f ms%n", executionTimeMs);
+
+ assertEquals("\"John\"", result);
+ System.out.println("✓ Test passed!");
+
+ System.out.println("=".repeat(50));
+ }
+
+ @Test
+ void runWithBinaryResult() {
+ ScriptRuntime runtime = ScriptRuntime.getInstance();
+
+ System.out.println("Running fromBase64 10 times with timing:");
+ System.out.println("=".repeat(50));
+
+ for (int i = 1; i <= 1; i++) {
+ long startTime = System.nanoTime();
+ Result result = Result.parse(runtime.run("import fromBase64 from dw::core::Binaries\n" +
+ "output application/octet-stream\n" +
+ "---\n" +
+ "fromBase64(\"12345678\")", ""));
+ long endTime = System.nanoTime();
+ double executionTimeMs = (endTime - startTime) / 1_000_000.0;
+
+ assertEquals("12345678", result.result);
+ System.out.printf("Run %2d: %.3f ms - Result: %s%n", i, executionTimeMs, result.result);
+ }
+
+ System.out.println("=".repeat(50));
+ }
+
+ @Test
+ void runWithInputProperties() {
+ ScriptRuntime runtime = ScriptRuntime.getInstance();
+ String encodedIn0 = Base64.getEncoder().encodeToString("1234567".getBytes());
+ Result result = Result.parse(runtime.run("in0.column_1[0] as Number",
+ "{\"in0\": " +
+ "{\"content\": \"" + encodedIn0 + "\", " +
+ "\"mimeType\": \"application/csv\", " +
+ "\"properties\": {\"header\": false, \"separator\": \"4\"}}}"));
+ assertEquals("567", result.result);
+
+ }
+
+ @Test
+ void streamSimpleScript() throws IOException {
+ ScriptRuntime runtime = ScriptRuntime.getInstance();
+
+ System.out.println("Testing streaming simple script:");
+ System.out.println("=".repeat(50));
+
+ StreamSession session = runtime.runStreaming("sqrt(144)", null);
+ assertFalse(session.isError(), "Expected successful session");
+ assertNull(session.getError());
+ assertNotNull(session.getMimeType());
+
+ byte[] buf = new byte[64];
+ ByteArrayOutputStream bos = new ByteArrayOutputStream();
+ int n;
+ while ((n = session.read(buf, buf.length)) > 0) {
+ bos.write(buf, 0, n);
+ }
+ String result = bos.toString(session.getCharset());
+ assertEquals("12", result);
+ StreamSession.close(session.register()); // clean up handle
+
+ System.out.println("Result: " + result);
+ System.out.println("✓ Streaming simple script passed!");
+ System.out.println("=".repeat(50));
+ }
+
+ @Test
+ void streamWithInputs() throws IOException {
+ ScriptRuntime runtime = ScriptRuntime.getInstance();
+
+ System.out.println("Testing streaming with inputs:");
+ System.out.println("=".repeat(50));
+
+ String inputsJson = String.format(
+ "{\"num1\": {\"content\": \"%s\", \"mimeType\": \"application/json\"}, " +
+ "\"num2\": {\"content\": \"%s\", \"mimeType\": \"application/json\"}}",
+ encode(25), encode(17)
+ );
+
+ StreamSession session = runtime.runStreaming("num1 + num2", inputsJson);
+ assertFalse(session.isError());
+
+ byte[] buf = new byte[64];
+ ByteArrayOutputStream bos = new ByteArrayOutputStream();
+ int n;
+ while ((n = session.read(buf, buf.length)) > 0) {
+ bos.write(buf, 0, n);
+ }
+ String result = bos.toString(session.getCharset());
+ assertEquals("42", result);
+ StreamSession.close(session.register());
+
+ System.out.println("Result: " + result);
+ System.out.println("✓ Streaming with inputs passed!");
+ System.out.println("=".repeat(50));
+ }
+
+ @Test
+ void streamChunkedRead() throws IOException {
+ ScriptRuntime runtime = ScriptRuntime.getInstance();
+
+ System.out.println("Testing streaming chunked read:");
+ System.out.println("=".repeat(50));
+
+ String script = "output application/json\n---\n{items: (1 to 100) map {id: $, name: \"item_\" ++ $}}";
+
+ StreamSession session = runtime.runStreaming(script, null);
+ assertFalse(session.isError());
+
+ byte[] smallBuf = new byte[32];
+ ByteArrayOutputStream bos = new ByteArrayOutputStream();
+ int n;
+ int chunkCount = 0;
+ while ((n = session.read(smallBuf, smallBuf.length)) > 0) {
+ bos.write(smallBuf, 0, n);
+ chunkCount++;
+ }
+ String result = bos.toString(session.getCharset());
+ assertTrue(chunkCount > 1, "Expected multiple chunks, got " + chunkCount);
+ assertTrue(result.contains("item_1"));
+ assertTrue(result.contains("item_100"));
+ StreamSession.close(session.register());
+
+ System.out.printf("Read %d chunks, total %d bytes%n", chunkCount, bos.size());
+ System.out.println("✓ Streaming chunked read passed!");
+ System.out.println("=".repeat(50));
+ }
+
+ @Test
+ void streamWithStreamingInput() throws Exception {
+ ScriptRuntime runtime = ScriptRuntime.getInstance();
+
+ System.out.println("Testing streaming with streaming input:");
+ System.out.println("=".repeat(50));
+
+ // Create an input stream session for JSON data
+ InputStreamSession inputSession = new InputStreamSession("application/json", "UTF-8");
+ long inputHandle = inputSession.register();
+
+ // Build inputs JSON referencing the streamHandle
+ String inputsJson = "{\"payload\": {\"streamHandle\": \"" + inputHandle + "\", \"mimeType\": \"application/json\"}}";
+
+ // The DW engine will read from the PipedInputStream on the main thread,
+ // so we must feed data from a separate thread.
+ CountDownLatch started = new CountDownLatch(1);
+ AtomicReference feedError = new AtomicReference<>();
+
+ Thread feeder = new Thread(() -> {
+ try {
+ started.countDown();
+ String jsonData = "{\"name\": \"Alice\", \"age\": 30}";
+ byte[] bytes = jsonData.getBytes("UTF-8");
+ inputSession.write(bytes, bytes.length);
+ inputSession.closeWriter();
+ } catch (Exception e) {
+ feedError.set(e);
+ }
+ });
+ feeder.start();
+ started.await();
+
+ // Run streaming with the piped input
+ StreamSession session = runtime.runStreaming("output application/json\n---\npayload.name", inputsJson);
+ assertFalse(session.isError(), "Expected successful session but got: " + session.getError());
+
+ byte[] buf = new byte[64];
+ ByteArrayOutputStream bos = new ByteArrayOutputStream();
+ int n;
+ while ((n = session.read(buf, buf.length)) > 0) {
+ bos.write(buf, 0, n);
+ }
+ String result = bos.toString(session.getCharset());
+ assertEquals("\"Alice\"", result);
+ StreamSession.close(session.register());
+ InputStreamSession.close(inputHandle);
+ feeder.join(5000);
+ assertNull(feedError.get(), "Feeder thread threw: " + feedError.get());
+
+ System.out.println("Result: " + result);
+ System.out.println("✓ Streaming with streaming input passed!");
+ System.out.println("=".repeat(50));
+ }
+
+ @Test
+ void streamWithLargeStreamingInput() throws Exception {
+ ScriptRuntime runtime = ScriptRuntime.getInstance();
+
+ System.out.println("Testing streaming with large streaming input:");
+ System.out.println("=".repeat(50));
+
+ InputStreamSession inputSession = new InputStreamSession("application/json", "UTF-8");
+ long inputHandle = inputSession.register();
+
+ String inputsJson = "{\"payload\": {\"streamHandle\": \"" + inputHandle + "\", \"mimeType\": \"application/json\"}}";
+
+ // Feed a large JSON array from a separate thread
+ AtomicReference feedError = new AtomicReference<>();
+ Thread feeder = new Thread(() -> {
+ try {
+ StringBuilder sb = new StringBuilder("[");
+ for (int i = 1; i <= 1000; i++) {
+ if (i > 1) sb.append(",");
+ sb.append("{\"id\":").append(i).append(",\"val\":\"item_").append(i).append("\"}");
+ }
+ sb.append("]");
+ byte[] bytes = sb.toString().getBytes("UTF-8");
+ // Write in chunks to simulate streaming
+ int chunkSize = 4096;
+ for (int off = 0; off < bytes.length; off += chunkSize) {
+ int len = Math.min(chunkSize, bytes.length - off);
+ byte[] chunk = new byte[len];
+ System.arraycopy(bytes, off, chunk, 0, len);
+ inputSession.write(chunk, len);
+ }
+ inputSession.closeWriter();
+ } catch (Exception e) {
+ feedError.set(e);
+ }
+ });
+ feeder.start();
+
+ StreamSession session = runtime.runStreaming("output application/json\n---\nsizeOf(payload)", inputsJson);
+ assertFalse(session.isError(), "Expected successful session but got: " + session.getError());
+
+ byte[] buf = new byte[256];
+ ByteArrayOutputStream bos = new ByteArrayOutputStream();
+ int n;
+ while ((n = session.read(buf, buf.length)) > 0) {
+ bos.write(buf, 0, n);
+ }
+ String result = bos.toString(session.getCharset());
+ assertEquals("1000", result);
+ StreamSession.close(session.register());
+ InputStreamSession.close(inputHandle);
+ feeder.join(10000);
+ assertNull(feedError.get(), "Feeder thread threw: " + feedError.get());
+
+ System.out.println("Result: " + result);
+ System.out.println("✓ Streaming with large streaming input passed!");
+ System.out.println("=".repeat(50));
+ }
+
+ @Test
+ void streamErrorSession() {
+ ScriptRuntime runtime = ScriptRuntime.getInstance();
+
+ System.out.println("Testing streaming error session:");
+ System.out.println("=".repeat(50));
+
+ StreamSession session = runtime.runStreaming("invalid syntax here", null);
+ assertTrue(session.isError());
+ assertNotNull(session.getError());
+ assertTrue(session.getError().contains("Unable to resolve reference"));
+
+ System.out.println("Error: " + session.getError());
+ System.out.println("✓ Streaming error session passed!");
+ System.out.println("=".repeat(50));
+ }
+
+ // ── Callback-based streaming pattern tests ──────────────────────────
+
+ @Test
+ void callbackOutputStreaming() throws IOException {
+ ScriptRuntime runtime = ScriptRuntime.getInstance();
+
+ System.out.println("Testing callback-based output streaming:");
+ System.out.println("=".repeat(50));
+
+ StreamSession session = runtime.runStreaming("output application/json\n---\n{items: (1 to 50) map {id: $}}", null);
+ assertFalse(session.isError());
+
+ // Simulate the write-callback pattern: read chunks and collect them
+ ByteArrayOutputStream collected = new ByteArrayOutputStream();
+ byte[] buf = new byte[64];
+ int callbackCount = 0;
+ int n;
+ while ((n = session.read(buf, buf.length)) > 0) {
+ // This is what the write callback would receive
+ collected.write(buf, 0, n);
+ callbackCount++;
+ }
+ String result = collected.toString(session.getCharset());
+ assertTrue(result.contains("\"id\": 1"), "Expected id 1 in result");
+ assertTrue(result.contains("\"id\": 50"), "Expected id 50 in result");
+ assertTrue(callbackCount > 0, "Expected at least one callback invocation");
+ StreamSession.close(session.register());
+
+ System.out.printf("Callback invoked %d times, total %d bytes%n", callbackCount, collected.size());
+ System.out.println("✓ Callback output streaming passed!");
+ System.out.println("=".repeat(50));
+ }
+
+ @Test
+ void callbackInputOutputStreaming() throws Exception {
+ ScriptRuntime runtime = ScriptRuntime.getInstance();
+
+ System.out.println("Testing callback-based input+output streaming:");
+ System.out.println("=".repeat(50));
+
+ // Simulate the read-callback pattern: a feeder thread pulls from a data source
+ // and pushes into an InputStreamSession, while the main thread reads the output.
+ InputStreamSession inputSession = new InputStreamSession("application/json", "UTF-8");
+ long inputHandle = inputSession.register();
+
+ String inputsJson = "{\"payload\": {\"streamHandle\": \"" + inputHandle + "\", \"mimeType\": \"application/json\"}}";
+
+ // Simulate read callback: feeds a JSON array in chunks
+ byte[] sourceData = "[10, 20, 30, 40, 50]".getBytes("UTF-8");
+ AtomicReference feedError = new AtomicReference<>();
+
+ Thread feeder = new Thread(() -> {
+ try {
+ int chunkSize = 8;
+ for (int off = 0; off < sourceData.length; off += chunkSize) {
+ int len = Math.min(chunkSize, sourceData.length - off);
+ byte[] chunk = new byte[len];
+ System.arraycopy(sourceData, off, chunk, 0, len);
+ inputSession.write(chunk, len);
+ }
+ inputSession.closeWriter();
+ } catch (Exception e) {
+ feedError.set(e);
+ }
+ }, "test-read-callback-feeder");
+ feeder.start();
+
+ StreamSession session = runtime.runStreaming("output application/json\n---\npayload map ($ * 2)", inputsJson);
+ assertFalse(session.isError(), "Expected successful session but got: " + session.getError());
+
+ // Simulate write callback: collect output chunks
+ ByteArrayOutputStream collected = new ByteArrayOutputStream();
+ byte[] buf = new byte[32];
+ int callbackCount = 0;
+ int n;
+ while ((n = session.read(buf, buf.length)) > 0) {
+ collected.write(buf, 0, n);
+ callbackCount++;
+ }
+ String result = collected.toString(session.getCharset());
+ assertTrue(result.contains("20"), "Expected 20 in result (10*2)");
+ assertTrue(result.contains("100"), "Expected 100 in result (50*2)");
+
+ StreamSession.close(session.register());
+ InputStreamSession.close(inputHandle);
+ feeder.join(5000);
+ assertNull(feedError.get(), "Feeder thread threw: " + feedError.get());
+
+ System.out.printf("Read callback fed %d bytes, write callback invoked %d times, output: %s%n",
+ sourceData.length, callbackCount, result.trim());
+ System.out.println("✓ Callback input+output streaming passed!");
+ System.out.println("=".repeat(50));
+ }
+
+ @Test
+ void callbackOutputStreamingError() {
+ ScriptRuntime runtime = ScriptRuntime.getInstance();
+
+ System.out.println("Testing callback-based output streaming with error:");
+ System.out.println("=".repeat(50));
+
+ StreamSession session = runtime.runStreaming("invalid syntax here", null);
+ assertTrue(session.isError());
+ assertNotNull(session.getError());
+
+ System.out.println("Error correctly returned: " + session.getError());
+ System.out.println("✓ Callback output streaming error passed!");
+ System.out.println("=".repeat(50));
+ }
+
+ static class Result {
+ boolean success;
+ String result;
+ String error;
+ boolean binary;
+ String mimeType;
+ String charset;
+
+ static Result parse(String json) {
+ Result result = new Result();
+ org.json.JSONObject obj = new org.json.JSONObject(json);
+
+ result.success = obj.getBoolean("success");
+ if (result.success) {
+ result.binary = obj.getBoolean("binary");
+ result.mimeType = obj.getString("mimeType");
+ result.charset = obj.getString("charset");
+ String encoded = obj.getString("result");
+ if (result.binary) {
+ result.result = encoded;
+ } else {
+ result.result = new String(Base64.getDecoder().decode(encoded), Charset.forName(result.charset));
+ }
+ } else {
+ result.error = obj.getString("error");
+ }
+ return result;
+ }
+ }
+
+}
diff --git a/settings.gradle b/settings.gradle
index a47c02c..befbb96 100644
--- a/settings.gradle
+++ b/settings.gradle
@@ -1,3 +1,3 @@
include 'native-cli'
include 'native-cli-integration-tests'
-
+include 'native-lib'