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..e4f259e
--- /dev/null
+++ b/native-lib/src/main/java/org/mule/weave/lib/NativeLib.java
@@ -0,0 +1,524 @@
+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);
+ }
+
+ // ── Streaming Input API ─────────────────────────────────────────────
+
+ /**
+ * Creates a new input stream session that the caller can write data into.
+ * The returned handle references a {@link java.io.PipedInputStream}/{@link java.io.PipedOutputStream}
+ * pair. The caller writes chunks via {@link #inputStreamWrite} and signals EOF
+ * with {@link #inputStreamClose}. The handle can then be referenced in the inputs
+ * JSON passed to {@link #runScriptOpen} using a {@code "streamHandle"} key.
+ *
+ * @param thread the isolate thread
+ * @param mimeType the MIME type of the data being streamed (C string)
+ * @param charset the charset of the data (C string), may be null for UTF-8
+ * @return a positive handle for the input stream session
+ */
+ @CEntryPoint(name = "input_stream_open")
+ public static long inputStreamOpen(IsolateThread thread, CCharPointer mimeType, CCharPointer charset) {
+ String mime = CTypeConversion.toJavaString(mimeType);
+ String cs = charset.isNull() ? null : CTypeConversion.toJavaString(charset);
+ InputStreamSession session = new InputStreamSession(mime, cs);
+ return session.register();
+ }
+
+ /**
+ * Writes a chunk of bytes into an open input stream session.
+ *
+ * @param thread the isolate thread
+ * @param handle the input stream session handle
+ * @param buffer the data to write
+ * @param bufferSize number of bytes to write from the buffer
+ * @return {@code 0} on success, {@code -1} on error (invalid handle or I/O failure)
+ */
+ @CEntryPoint(name = "input_stream_write")
+ public static int inputStreamWrite(IsolateThread thread, long handle, CCharPointer buffer, int bufferSize) {
+ InputStreamSession session = InputStreamSession.get(handle);
+ if (session == null) {
+ return -1;
+ }
+ try {
+ byte[] tmp = new byte[bufferSize];
+ for (int i = 0; i < bufferSize; i++) {
+ tmp[i] = buffer.read(i);
+ }
+ session.write(tmp, bufferSize);
+ return 0;
+ } catch (IOException e) {
+ return -1;
+ }
+ }
+
+ /**
+ * Closes the write end of an input stream session, signalling EOF to the reader.
+ * The session remains in the registry so the DW engine can finish reading.
+ * It will be fully cleaned up when the output streaming session is closed.
+ *
+ * @param thread the isolate thread
+ * @param handle the input stream session handle
+ * @return {@code 0} on success, {@code -1} on error
+ */
+ @CEntryPoint(name = "input_stream_close")
+ public static int inputStreamClose(IsolateThread thread, long handle) {
+ InputStreamSession session = InputStreamSession.get(handle);
+ if (session == null) {
+ return -1;
+ }
+ try {
+ session.closeWriter();
+ return 0;
+ } catch (IOException e) {
+ return -1;
+ }
+ }
+
+ // ── Streaming Output API ──────────────────────────────────────────────
+
+ /**
+ * Executes a DataWeave script and returns an opaque handle to a streaming session.
+ * The result can then be read incrementally via {@link #runScriptRead} and must be
+ * closed with {@link #runScriptClose} when done.
+ *
+ * A handle value of {@code -1} indicates an error during compilation or execution.
+ * In that case call {@link #runScriptStreamError} to retrieve the error message.
+ *
+ * @param thread the isolate thread
+ * @param script the DataWeave script (C string)
+ * @param inputsJson JSON-encoded inputs map (C string), may be null
+ * @return a positive session handle, or {@code -1} on error
+ */
+ @CEntryPoint(name = "run_script_open")
+ public static long runScriptOpen(IsolateThread thread, CCharPointer script, CCharPointer inputsJson) {
+ String dwScript = CTypeConversion.toJavaString(script);
+ String inputs = inputsJson.isNull() ? null : CTypeConversion.toJavaString(inputsJson);
+
+ ScriptRuntime runtime = ScriptRuntime.getInstance();
+ StreamSession session = runtime.runStreaming(dwScript, inputs);
+ return session.register();
+ }
+
+ /**
+ * Reads the next chunk of bytes from an open streaming session.
+ *
+ * @param thread the isolate thread
+ * @param handle the session handle returned by {@link #runScriptOpen}
+ * @param buffer caller-allocated buffer to write into
+ * @param bufferSize size of the buffer in bytes
+ * @return the number of bytes written into {@code buffer}, {@code 0} on EOF,
+ * or {@code -1} on error (invalid handle or I/O failure)
+ */
+ @CEntryPoint(name = "run_script_read")
+ public static int runScriptRead(IsolateThread thread, long handle, CCharPointer buffer, int bufferSize) {
+ StreamSession session = StreamSession.get(handle);
+ if (session == null) {
+ return -1;
+ }
+ try {
+ byte[] tmp = new byte[bufferSize];
+ int n = session.read(tmp, bufferSize);
+ if (n <= 0) {
+ return 0;
+ }
+ for (int i = 0; i < n; i++) {
+ buffer.write(i, tmp[i]);
+ }
+ return n;
+ } catch (IOException e) {
+ return -1;
+ }
+ }
+
+ /**
+ * Returns JSON metadata for an open streaming session:
+ * {@code {"mimeType":"...","charset":"...","binary":true/false}}.
+ *
+ * The caller must free the returned pointer with {@link #freeCString}.
+ *
+ * @param thread the isolate thread
+ * @param handle the session handle
+ * @return an unmanaged C string with JSON metadata, or a null pointer if the handle is invalid
+ */
+ @CEntryPoint(name = "run_script_metadata")
+ public static CCharPointer runScriptMetadata(IsolateThread thread, long handle) {
+ StreamSession session = StreamSession.get(handle);
+ if (session == null) {
+ return CTypeConversion.toCString("").get();
+ }
+ String json = "{"
+ + "\"mimeType\":\"" + session.getMimeType() + "\","
+ + "\"charset\":\"" + session.getCharset() + "\","
+ + "\"binary\":" + session.isBinary()
+ + "}";
+ return toUnmanagedCString(json);
+ }
+
+ /**
+ * Retrieves the error message for a failed streaming session.
+ *
+ * When a session was created from an execution failure its handle is still valid
+ * and the error message can be obtained here. The caller must free the returned pointer
+ * with {@link #freeCString}.
+ *
+ * @param thread the isolate thread
+ * @param handle the session handle
+ * @return an unmanaged C string with the error message, or empty string if not an error session
+ */
+ @CEntryPoint(name = "run_script_stream_error")
+ public static CCharPointer runScriptStreamError(IsolateThread thread, long handle) {
+ StreamSession session = StreamSession.get(handle);
+ if (session == null || session.getError() == null) {
+ return toUnmanagedCString("");
+ }
+ return toUnmanagedCString(session.getError());
+ }
+
+ /**
+ * Closes a streaming session, releasing the underlying {@link java.io.InputStream} and
+ * removing the session from the registry.
+ *
+ * @param thread the isolate thread
+ * @param handle the session handle
+ */
+ @CEntryPoint(name = "run_script_close")
+ public static void runScriptClose(IsolateThread thread, long handle) {
+ StreamSession.close(handle);
+ }
+
+ // ── 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()) + "\"}");
+ }
+
+ 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.
+ CCharPointer readBuf = UnmanagedMemory.malloc(CALLBACK_BUFFER_SIZE);
+ final long readCallbackAddr = readCallback.rawValue();
+ final long ctxAddr = ctx.rawValue();
+ final long readBufAddr = readBuf.rawValue();
+ Thread feeder = new Thread(new InputCallbackFeeder(
+ readCallbackAddr, ctxAddr, readBufAddr, 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, readBuf, 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, readBuf, inputHandle);
+ return toUnmanagedCString("{\"success\":false,\"error\":\""
+ + "Write callback returned error: " + rc + "\"}");
+ }
+ }
+ } finally {
+ UnmanagedMemory.free(writeBuf);
+ }
+ } catch (IOException e) {
+ cleanupFeeder(feeder, readBuf, inputHandle);
+ return toUnmanagedCString("{\"success\":false,\"error\":\""
+ + escapeJsonString(e.getMessage()) + "\"}");
+ }
+
+ cleanupFeeder(feeder, readBuf, 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) {
+ if (existingJson == null || existingJson.trim().isEmpty() || existingJson.trim().equals("{}")) {
+ return "{\"" + name + "\":" + entryJson + "}";
+ }
+ // Insert before the final closing brace
+ String trimmed = existingJson.trim();
+ int lastBrace = trimmed.lastIndexOf('}');
+ if (lastBrace <= 0) {
+ return "{\"" + name + "\":" + entryJson + "}";
+ }
+ String prefix = trimmed.substring(0, lastBrace).trim();
+ // Add comma if there is already content
+ if (prefix.length() > 1) {
+ return prefix + ",\"" + name + "\":" + entryJson + "}";
+ }
+ return "{\"" + name + "\":" + entryJson + "}";
+ }
+
+ /**
+ * Waits for the feeder thread to finish and frees native resources.
+ */
+ private static void cleanupFeeder(Thread feeder, CCharPointer readBuf, long inputHandle) {
+ try {
+ feeder.join(5000);
+ } catch (InterruptedException ignored) {
+ }
+ UnmanagedMemory.free(readBuf);
+ 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()}.
+ */
+ private static final class InputCallbackFeeder implements Runnable {
+ private final long readCallbackAddr;
+ private final long ctxAddr;
+ private final long readBufAddr;
+ private final InputStreamSession inputSession;
+
+ InputCallbackFeeder(long readCallbackAddr, long ctxAddr, long readBufAddr,
+ InputStreamSession inputSession) {
+ this.readCallbackAddr = readCallbackAddr;
+ this.ctxAddr = ctxAddr;
+ this.readBufAddr = readBufAddr;
+ this.inputSession = inputSession;
+ }
+
+ @Override
+ public void run() {
+ NativeCallbacks.ReadCallback cb = WordFactory.pointer(readCallbackAddr);
+ PointerBase ctx = WordFactory.pointer(ctxAddr);
+ CCharPointer buf = WordFactory.pointer(readBufAddr);
+ 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 {
+ 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..849e63f
--- /dev/null
+++ b/native-lib/src/main/java/org/mule/weave/lib/ScriptRuntime.java
@@ -0,0 +1,472 @@
+package org.mule.weave.lib;
+
+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 {
+ String json = inputsJson.trim();
+
+ // Parse top-level entries: "name": { ... }
+ int pos = 1; // Skip opening brace
+
+ while (pos < json.length()) {
+ // Skip whitespace, commas
+ while (pos < json.length() && (Character.isWhitespace(json.charAt(pos)) || json.charAt(pos) == ',')) {
+ pos++;
+ }
+
+ if (pos >= json.length() || json.charAt(pos) == '}') break;
+
+ // Expect a quoted string (binding name)
+ if (json.charAt(pos) != '"') break;
+
+ int nameEnd = findClosingQuote(json, pos + 1);
+ if (nameEnd == -1) break;
+
+ String name = json.substring(pos + 1, nameEnd);
+ pos = nameEnd + 1; // Move past the closing quote
+
+ // Skip whitespace and colon
+ while (pos < json.length() && (Character.isWhitespace(json.charAt(pos)) || json.charAt(pos) == ':')) {
+ pos++;
+ }
+
+ // Expect opening brace for nested object
+ if (pos >= json.length() || json.charAt(pos) != '{') break;
+
+ int objEnd = findClosingBrace(json, pos + 1);
+ if (objEnd == -1) break;
+
+ String nestedContent = json.substring(pos + 1, objEnd);
+ pos = objEnd + 1;
+
+ String streamHandleRaw = extractStringValue(nestedContent, "streamHandle");
+ String contentRaw = extractStringValue(nestedContent, "content");
+
+ if (streamHandleRaw != null) {
+ // Streaming input: look up the InputStreamSession by handle
+ long streamHandle = Long.parseLong(streamHandleRaw);
+ InputStreamSession inputSession = InputStreamSession.get(streamHandle);
+ if (inputSession == null) {
+ throw new RuntimeException("Invalid streamHandle " + streamHandle + " for input '" + name + "'");
+ }
+ String mimeTypeRaw = extractStringValue(nestedContent, "mimeType");
+ if (mimeTypeRaw == null) {
+ mimeTypeRaw = inputSession.getMimeType();
+ }
+ String charsetRaw = extractStringValue(nestedContent, "charset");
+ Charset charset = Charset.forName(charsetRaw != null ? charsetRaw : inputSession.getCharset());
+ Option mimeType = Option.apply(mimeTypeRaw);
+
+ BindingValue bindingValue = new BindingValue(inputSession.getInputStream(), mimeType, Map$.MODULE$.empty(), charset);
+ bindings.addBinding(name, bindingValue);
+
+ } else if (contentRaw != null) {
+ String mimeTypeRaw = extractStringValue(nestedContent, "mimeType");
+ String propertiesRaw = null;
+ if (nestedContent.indexOf("\"properties\": {") != -1) {
+ propertiesRaw = nestedContent.substring(nestedContent.indexOf("\"properties\": {") + 14, nestedContent.lastIndexOf("}") + 1);
+ }
+ String charsetRaw = extractStringValue(nestedContent, "charset");
+
+ Map properties = Map$.MODULE$.empty();
+ if (propertiesRaw != null) {
+ properties = parseJsonProperties(propertiesRaw);
+ }
+ Charset charset = Charset.forName(charsetRaw != null ? charsetRaw : "UTF-8");
+ 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;
+ }
+
+ private Map parseJsonProperties(String jsonProperties) {
+ if (jsonProperties == null || jsonProperties.trim().isEmpty()) {
+ return Map$.MODULE$.empty();
+ }
+
+ String json = jsonProperties.trim();
+ if (json.charAt(0) != '{') {
+ throw new IllegalArgumentException("properties must be a JSON object (must start with '{'): " + jsonProperties);
+ }
+
+ int end = findClosingBrace(json, 1);
+ if (end == -1) {
+ throw new IllegalArgumentException("properties must be a valid JSON object (missing closing '}'): " + jsonProperties);
+ }
+
+ // Disallow trailing non-whitespace after the object
+ for (int i = end + 1; i < json.length(); i++) {
+ if (!Character.isWhitespace(json.charAt(i))) {
+ throw new IllegalArgumentException("properties must contain a single JSON object (unexpected trailing content): " + jsonProperties);
+ }
+ }
+
+ Map result = Map$.MODULE$.empty();
+ int pos = 1; // skip '{'
+
+ while (pos < end) {
+ while (pos < end && (Character.isWhitespace(json.charAt(pos)) || json.charAt(pos) == ',')) {
+ pos++;
+ }
+
+ if (pos >= end) {
+ break;
+ }
+
+ if (json.charAt(pos) != '"') {
+ throw new IllegalArgumentException("properties keys must be quoted strings at position " + pos + ": " + jsonProperties);
+ }
+
+ int keyEnd = findClosingQuote(json, pos + 1);
+ if (keyEnd == -1 || keyEnd > end) {
+ throw new IllegalArgumentException("properties has an unterminated key string: " + jsonProperties);
+ }
+
+ String key = unescapeJsonString(json.substring(pos + 1, keyEnd));
+ pos = keyEnd + 1;
+
+ while (pos < end && Character.isWhitespace(json.charAt(pos))) {
+ pos++;
+ }
+ if (pos >= end || json.charAt(pos) != ':') {
+ throw new IllegalArgumentException("properties expected ':' after key '" + key + "': " + jsonProperties);
+ }
+ pos++;
+
+ while (pos < end && Character.isWhitespace(json.charAt(pos))) {
+ pos++;
+ }
+ if (pos >= end) {
+ throw new IllegalArgumentException("properties missing value for key '" + key + "': " + jsonProperties);
+ }
+
+ Object value;
+ char c = json.charAt(pos);
+ if (c == '"') {
+ int valueEnd = findClosingQuote(json, pos + 1);
+ if (valueEnd == -1 || valueEnd > end) {
+ throw new IllegalArgumentException("properties has an unterminated string value for key '" + key + "': " + jsonProperties);
+ }
+ value = unescapeJsonString(json.substring(pos + 1, valueEnd));
+ pos = valueEnd + 1;
+ } else if (c == 't' || c == 'f') {
+ if (json.startsWith("true", pos)) {
+ value = Boolean.TRUE;
+ pos += 4;
+ } else if (json.startsWith("false", pos)) {
+ value = Boolean.FALSE;
+ pos += 5;
+ } else {
+ throw new IllegalArgumentException("properties invalid boolean value for key '" + key + "' at position " + pos + ": " + jsonProperties);
+ }
+ } else if (c == 'n') {
+ throw new IllegalArgumentException("properties values cannot be null (key '" + key + "'): " + jsonProperties);
+ } else if (c == '{' || c == '[') {
+ throw new IllegalArgumentException("properties values must be primitive (string/number/boolean) (key '" + key + "'): " + jsonProperties);
+ } else {
+ int numEnd = pos;
+ while (numEnd < end) {
+ char nc = json.charAt(numEnd);
+ if (nc == ',' || nc == '}' || Character.isWhitespace(nc)) {
+ break;
+ }
+ numEnd++;
+ }
+ String numStr = json.substring(pos, numEnd);
+ if (numStr.isEmpty()) {
+ throw new IllegalArgumentException("properties invalid number value for key '" + key + "' at position " + pos + ": " + jsonProperties);
+ }
+ try {
+ if (numStr.indexOf('.') >= 0 || numStr.indexOf('e') >= 0 || numStr.indexOf('E') >= 0) {
+ value = Double.parseDouble(numStr);
+ } else {
+ value = Long.parseLong(numStr);
+ }
+ } catch (NumberFormatException nfe) {
+ throw new IllegalArgumentException("properties invalid number value for key '" + key + "': " + numStr, nfe);
+ }
+ pos = numEnd;
+ }
+
+ result = (Map) result.$plus(new Tuple2<>(key, value));
+ }
+
+ return result;
+ }
+
+ /**
+ * Parse a JSON string value starting at position (which should be at the opening quote).
+ * Returns the unescaped string content (without quotes).
+ */
+ private String parseString(String json, int startPos) {
+ if (json.charAt(startPos) != '"') return null;
+
+ int endPos = findClosingQuote(json, startPos + 1);
+ if (endPos == -1) return null;
+
+ String escaped = json.substring(startPos + 1, endPos);
+ return unescapeJsonString(escaped);
+ }
+
+ /**
+ * Extract a string value by key from a JSON object content.
+ * Simplified version assuming all values are strings.
+ */
+ private String extractStringValue(String json, String key) {
+ String searchKey = "\"" + key + "\"";
+ int keyPos = json.indexOf(searchKey);
+ if (keyPos == -1) return null;
+
+ // Find the colon after the key
+ int colonPos = keyPos + searchKey.length();
+ while (colonPos < json.length() && json.charAt(colonPos) != ':') {
+ colonPos++;
+ }
+ if (colonPos >= json.length()) return null;
+
+ // Skip whitespace after colon
+ int valueStart = colonPos + 1;
+ while (valueStart < json.length() && Character.isWhitespace(json.charAt(valueStart))) {
+ valueStart++;
+ }
+
+ if (valueStart >= json.length() || json.charAt(valueStart) != '"') return null;
+
+ return parseString(json, valueStart);
+ }
+
+ /**
+ * Find the closing quote, skipping escaped quotes.
+ * Properly handles escaped backslashes.
+ */
+ private int findClosingQuote(String str, int startPos) {
+ for (int i = startPos; i < str.length(); i++) {
+ char c = str.charAt(i);
+
+ if (c == '\\' && i + 1 < str.length()) {
+ // Skip the escaped character
+ i++;
+ } else if (c == '"') {
+ // Found unescaped quote
+ return i;
+ }
+ }
+ return -1;
+ }
+
+ /**
+ * Find the closing brace, properly handling nested braces and strings.
+ */
+ private int findClosingBrace(String str, int startPos) {
+ int depth = 1;
+ boolean inString = false;
+
+ for (int i = startPos; i < str.length(); i++) {
+ char c = str.charAt(i);
+
+ if (inString) {
+ if (c == '\\' && i + 1 < str.length()) {
+ i++; // Skip escaped character
+ } else if (c == '"') {
+ inString = false;
+ }
+ } else {
+ if (c == '"') {
+ inString = true;
+ } else if (c == '{') {
+ depth++;
+ } else if (c == '}') {
+ depth--;
+ if (depth == 0) {
+ return i;
+ }
+ }
+ }
+ }
+ return -1;
+ }
+
+ /**
+ * Unescapes JSON string escape sequences.
+ * Order matters: handle \\\\ first to avoid conflicts with other escape sequences.
+ */
+ private String unescapeJsonString(String input) {
+ if (input == null) {
+ return null;
+ }
+ // Use a placeholder for escaped backslashes to avoid conflicts
+ String placeholder = "\u0000"; // null character as temporary placeholder
+ return input
+ .replace("\\\\", placeholder)
+ .replace("\\n", "\n")
+ .replace("\\r", "\r")
+ .replace("\\t", "\t")
+ .replace("\\\"", "\"")
+ .replace(placeholder, "\\");
+ }
+
+ 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..627f172
--- /dev/null
+++ b/native-lib/src/main/java/org/mule/weave/lib/StreamSession.java
@@ -0,0 +1,124 @@
+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;
+ }
+
+ /**
+ * 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..d6ab1d6
--- /dev/null
+++ b/native-lib/src/test/java/org/mule/weave/lib/ScriptRuntimeTest.java
@@ -0,0 +1,617 @@
+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();
+
+ String successString = json.substring(json.indexOf(":") + 1, json.indexOf(","));
+ result.success = Boolean.parseBoolean(successString);
+ if (result.success) {
+ String binaryString = json.substring(json.indexOf(",\"binary\":") + 10, json.indexOf("}"));
+ result.binary = Boolean.parseBoolean(binaryString);
+ String resultString = json.substring(json.indexOf(",\"result\":") + 11, json.indexOf(",\"mimeType\":")-1);
+ String mimeTypeString = json.substring(json.indexOf(",\"mimeType\":") + 13, json.indexOf(",\"charset\":")-1);
+ result.mimeType = mimeTypeString;
+ String charsetString = json.substring(json.indexOf(",\"charset\":") + 12, json.indexOf(",\"binary\":")-1);
+ result.charset = charsetString;
+ if (result.binary) {
+ result.result = resultString;
+ } else {
+ result.result = new String(Base64.getDecoder().decode(resultString), Charset.forName(result.charset));
+ }
+
+ } else {
+ result.error = json.substring(json.indexOf(",\"error\":") + 10, json.length()-2);
+ }
+ 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'