You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2026/10/12 19:53:39 UTC
(camel) branch main updated: CAMEL-25531: camel-wasm - Run WASI Preview 1 command modules (#27678)
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 2e7febe42485 CAMEL-25531: camel-wasm - Run WASI Preview 1 command modules (#27678)
2e7febe42485 is described below
commit 2e7febe4248570b189b06104d21d4412eaf301c3
Author: allthingssecurity <49...@users.noreply.github.com>
AuthorDate: Tue Oct 13 01:23:31 2026 +0530
CAMEL-25531: camel-wasm - Run WASI Preview 1 command modules (#27678)
Co-Authored-By: Claude Opus 5.5 <no...@anthropic.com>
---
.../org/apache/camel/catalog/components/wasm.json | 20 +-
.../apache/camel/catalog/docs/wasm-component.adoc | 315 ++++---
components/camel-wasm/pom.xml | 5 +
.../wasm/WasmConfigurationConfigurer.java | 44 +-
.../component/wasm/WasmEndpointConfigurer.java | 44 +-
.../component/wasm/WasmEndpointUriFactory.java | 18 +-
.../org/apache/camel/component/wasm/wasm.json | 20 +-
.../camel-wasm/src/main/docs/wasm-component.adoc | 315 ++++---
.../apache/camel/component/wasm/WasmComponent.java | 12 +-
.../camel/component/wasm/WasmConfiguration.java | 125 ++-
.../apache/camel/component/wasm/WasmEndpoint.java | 19 +-
.../component/wasm/WasmExitCodeException.java | 51 ++
.../apache/camel/component/wasm/WasmProducer.java | 425 +++++++++-
.../java/org/apache/camel/wasm/WasiCommand.java | 204 +++++
.../src/main/java/org/apache/camel/wasm/Wasm.java | 12 +
.../java/org/apache/camel/wasm/WasmSupport.java | 14 -
.../camel/component/wasm/WasmComponentTest.java | 927 ++++++++++++++++++++-
.../org/apache/camel/wasm/WasiCommandTest.java | 60 ++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 22 +
.../component/ComponentsBuilderFactory.java | 3 +-
.../component/dsl/WasmComponentBuilderFactory.java | 6 +-
.../builder/endpoint/EndpointHeaderBuilders.java | 14 +
.../builder/endpoint/StaticEndpointBuilders.java | 24 +-
.../endpoint/dsl/WasmEndpointBuilderFactory.java | 320 ++++++-
24 files changed, 2625 insertions(+), 394 deletions(-)
diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/wasm.json b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/wasm.json
index c931e1190ccf..076279f11820 100644
--- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/wasm.json
+++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/wasm.json
@@ -3,7 +3,7 @@
"kind": "component",
"name": "wasm",
"title": "Wasm",
- "description": "Invoke Wasm functions.",
+ "description": "Run Wasm programs (WASI Preview 1 command modules) with the message body as standard input.",
"deprecated": false,
"firstVersion": "4.4.0",
"label": "core,script",
@@ -14,7 +14,7 @@
"version": "4.23.0-SNAPSHOT",
"scheme": "wasm",
"extendsScheme": "",
- "syntax": "wasm:functionName",
+ "syntax": "wasm:module",
"async": false,
"api": false,
"consumerOnly": false,
@@ -27,9 +27,19 @@
"lazyStartProducer": { "index": 0, "kind": "property", "displayName": "Lazy Start Producer", "group": "producer", "label": "producer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether the producer should be started lazy (on the first message). By starting lazy you can use this to allow CamelContext and routes to startup in situations where a producer may otherwise fail [...]
"autowiredEnabled": { "index": 1, "kind": "property", "displayName": "Autowired Enabled", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "description": "Whether autowiring is enabled. This is used for automatic autowiring options (the option must be marked as autowired) by looking up in the registry to find if there is a single instance of matching t [...]
},
+ "headers": {
+ "CamelWasmExitCode": { "index": 0, "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "Integer", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The exit code of the module: 0 when it returned from its _start function, otherwise the value it passed to proc_exit. Not set when the module trapped, timed out or exceeded maxOutputSize.", "constantName": "org.apache.camel.wasm.Wasm$Headers#E [...]
+ "CamelWasmStderr": { "index": 1, "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "What the module wrote to its standard error, decoded as UTF-8. Not set when the module wrote nothing or timed out.", "constantName": "org.apache.camel.wasm.Wasm$Headers#STDERR" }
+ },
"properties": {
- "functionName": { "index": 0, "kind": "path", "displayName": "Function Name", "group": "producer", "label": "", "required": true, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The Function Name" },
- "module": { "index": 1, "kind": "parameter", "displayName": "Module", "group": "producer", "label": "", "required": true, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.wasm.WasmConfiguration", "configurationField": "configuration", "description": "Set the module (the distributable, loadable, and executable unit of code in WebAssembly) resource that pr [...]
- "lazyStartProducer": { "index": 2, "kind": "parameter", "displayName": "Lazy Start Producer", "group": "producer (advanced)", "label": "producer,advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether the producer should be started lazy (on the first message). By starting lazy you can use this to allow CamelContext and routes to startup in situations where a produc [...]
+ "module": { "index": 0, "kind": "path", "displayName": "Module", "group": "producer", "label": "", "required": true, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The WASI Preview 1 command module to run, as a resource: by default from the classpath, or with a prefix such as classpath:, file: or http:" },
+ "args": { "index": 1, "kind": "parameter", "displayName": "Args", "group": "producer", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.wasm.WasmConfiguration", "configurationField": "configuration", "description": "The command line arguments passed to the module, separated by whitespace. Use double quotes around an argument that contains whites [...]
+ "environment": { "index": 2, "kind": "parameter", "displayName": "Environment", "group": "producer", "label": "", "required": false, "type": "object", "javaType": "java.util.Map<java.lang.String, java.lang.Object>", "prefix": "environment.", "multiValue": true, "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.wasm.WasmConfiguration", "configurationField": "configuration", "description": "Environment variables for the module, [...]
+ "environmentHeaders": { "index": 3, "kind": "parameter", "displayName": "Environment Headers", "group": "producer", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.wasm.WasmConfiguration", "configurationField": "configuration", "description": "A comma separated allow-list of message headers whose values are passed to the module as environment v [...]
+ "failOnNonZeroExit": { "index": 4, "kind": "parameter", "displayName": "Fail On Non Zero Exit", "group": "producer", "label": "", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "configurationClass": "org.apache.camel.component.wasm.WasmConfiguration", "configurationField": "configuration", "description": "Whether a module that exits with a non-zero exit code fails the exchange with a WasmExi [...]
+ "maxMemoryPages": { "index": 5, "kind": "parameter", "displayName": "Max Memory Pages", "group": "producer", "label": "", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.wasm.WasmConfiguration", "configurationField": "configuration", "description": "The maximum size of the memory of the module, in WebAssembly pages of 64 KiB. Rust and Go programs declare no maximum by [...]
+ "maxOutputSize": { "index": 6, "kind": "parameter", "displayName": "Max Output Size", "group": "producer", "label": "", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.wasm.WasmConfiguration", "configurationField": "configuration", "description": "The maximum number of bytes the module may write to its standard output, and separately to its standard error, which are bu [...]
+ "timeout": { "index": 7, "kind": "parameter", "displayName": "Timeout", "group": "producer", "label": "", "required": false, "type": "duration", "javaType": "long", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.wasm.WasmConfiguration", "configurationField": "configuration", "description": "The maximum time the module may run for one exchange, in milliseconds. When it is exceeded the module is interrupted and the exchange f [...]
+ "lazyStartProducer": { "index": 8, "kind": "parameter", "displayName": "Lazy Start Producer", "group": "producer (advanced)", "label": "producer,advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether the producer should be started lazy (on the first message). By starting lazy you can use this to allow CamelContext and routes to startup in situations where a produc [...]
}
}
diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/wasm-component.adoc b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/wasm-component.adoc
index ad9aa3daf956..27b588693644 100644
--- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/wasm-component.adoc
+++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/wasm-component.adoc
@@ -2,7 +2,7 @@
:doctitle: Wasm
:shortname: wasm
:artifactid: camel-wasm
-:description: Invoke Wasm functions.
+:description: Run Wasm programs (WASI Preview 1 command modules) with the message body as standard input.
:since: 4.4
:supportlevel: Experimental
:tabs-sync-option:
@@ -13,9 +13,15 @@
*{component-header}*
-https://webassembly.org[WebAssembly (Wasm)] is a low-level bytecode format designed as a portable target for the compilation of high-level languages like C, C++, and Rust, enabling deployment on the web for client and server applications.
+https://webassembly.org[WebAssembly (Wasm)] is a portable bytecode format that languages such as Rust, Go, C and C++
+compile to. https://wasi.dev[WASI] is the system interface those programs use for their standard input, output,
+arguments and environment.
-This component provides support to leverage Wasm functions for message transformation.
+The Wasm component runs a WASI Preview 1 _command module_ (a program with a `_start` function) as a processor: the
+message body is the standard input of the program, and what the program writes to its standard output becomes the
+message body. Any program built for WASI Preview 1 can be used as it is, without a Camel specific API. The module runs
+inside the JVM on https://github.com/bytecodealliance/endive[Endive], a WebAssembly runtime written in Java, so no native
+code or external process is involved.
Maven users will need to add the following dependency to their `pom.xml`
for this component:
@@ -32,9 +38,18 @@ for this component:
== URI format
----
-wasm://functionName?[options]
+wasm:module[?options]
----
+The `module` is the location of the `.wasm` file, loaded as a resource: from the classpath by default, or with a
+prefix such as `classpath:`, `file:` or `http:`. For example `wasm:classpath:upper.wasm` or
+`wasm:file:/opt/transforms/upper.wasm`.
+
+WARNING: Use a static URI. With a dynamic one, such as `toD("wasm:${header.module}")`, data of the message chooses the
+program that runs, including a module downloaded from an `http:` location, and the module is loaded and parsed again
+for every new URI (0.2 to 0.4 seconds for a Go program). To choose between known modules, use a
+xref:eips:choice-eip.adoc[Choice] with one static `wasm` endpoint per module.
+
// component options: START
include::partial$component-configure-options.adoc[]
include::partial$component-endpoint-options.adoc[]
@@ -43,133 +58,168 @@ include::partial$component-endpoint-headers.adoc[]
== Usage
-=== Writing A Wasm processor
-
-In _Wasm_, sharing objects between the host, in this case the _JVM_, and the _Wasm_ module is deliberately restricted and as of today, it requires a number of steps:
-
-1. From the _host_, call a function inside the webassembly module that allocates a block of memory and returns its address, then save it
-2. From the _host_, write the data that should be exchanged with the _Wasm_ module to the saved address
-3. From the _host_, invoke the required function passing both the address where the data is written and its size
-4. From the _Wasm_ module, read the data and process it
-5. From the _host_, release the memory when done
+=== Writing a module
-==== Providing functions for memory management
-
-The module hosting the function *must* provide the functions to allocate/deallocate memory that *must* be named `alloc` and `dealloc` respectively.
-
-Here's an example of the mentioned functions implemented in https://www.rust-lang.org[Rust]:
+A module is an ordinary program that reads its standard input and writes its standard output. For example, in
+https://www.rust-lang.org[Rust]:
[source,rust]
----
-pub extern "C" fn alloc(size: u32) -> *mut u8 {
- let mut buf = Vec::with_capacity(size as usize);
- let ptr = buf.as_mut_ptr();
-
- // tell Rust not to clean this up
- mem::forget(buf);
+use std::io::{self, Read, Write};
- ptr
-}
-
-pub unsafe extern "C" fn dealloc(ptr: &mut u8, len: i32) {
- // Retakes the pointer which allows its memory to be freed.
- let _ = Vec::from_raw_parts(ptr, 0, len as usize);
+fn main() {
+ let mut input = String::new();
+ io::stdin().read_to_string(&mut input).unwrap();
+ io::stdout().write_all(input.to_uppercase().as_bytes()).unwrap();
}
----
-==== Data shapes
+Built with `rustup target add wasm32-wasip1` and `cargo build --release --target wasm32-wasip1`.
-It is not possible to share a Java object with the Wasm module directly, and as mentioned before,
-data exchange leverages Wasm's memory that can be accessed by both the host and the guest runtimes.
-At this stage, the data structure that the component exchange with the Wasm function is a subset of the Apache Camel Message,
-containing headers the body encoded as a base64 string:
+The same program in https://go.dev[Go]:
-._Java-only: data exchange wrapper class for Wasm module communication_
-[source,java]
+[source,go]
----
-public static class Wrapper {
- @JsonProperty
- public Map<String, String> headers = new HashMap<>();
-
- @JsonProperty
- public byte[] body;
-}
-----
-
-==== Data processing
-
-The component expects the processing function to have the following signature:
-
-[source,rust]
-----
-fn function(ptr: u32, len: u32) -> u64
-----
-
-* it accepts two 32bit unsigned integers arguments
-** a pointer to the memory location when the input data has been written (`ptr`)
-** the size of the input data (`len`)
-* it returns a 64bit unsigned integer where:
-** the first 32bit represents a pointer to the return data
-** the last 31bit represents the size of the return data, which must have the same data shape discussed above unless there is an error (see below)
-** the most significant bit of the returned data size is reserved to signal an error, so if it is set, then the return data could contain an error message/code/etc
-
-Here's an example of a complete function:
-
-[source,rust]
-----
-#[derive(Serialize, Deserialize)]
-struct Message {
- headers: HashMap<String, serde_json::Value>,
-
- #[serde(with = "Base64Standard")]
- body: Vec<u8>,
-}
-
-#[cfg_attr(all(target_arch = "wasm32"), export_name = "process")]
-#[no_mangle]
-pub extern fn process(ptr: u32, len: u32) -> u64 {
- let bytes = unsafe {
- slice::from_raw_parts_mut(
- ptr as *mut u8,
- len as usize)
- };
+package main
- let mut msg: Message = serde_json::from_slice(bytes).unwrap();
+import (
+ "bytes"
+ "io"
+ "os"
+)
-
- let out_vec = serde_json::to_vec(&msg).unwrap();
- let out_len = out_vec.len();
- let out_ptr = alloc(out_len as u32);
-
- unsafe {
- std::ptr::copy_nonoverlapping(
- out_vec.as_ptr(),
- out_ptr,
- out_len as usize)
- };
-
- return ((out_ptr as u64) << 32) | out_len as u64;
+func main() {
+ input, _ := io.ReadAll(os.Stdin)
+ os.Stdout.Write(bytes.ToUpper(input))
}
----
-==== Errors and module state
-
-The module is instantiated once per producer (each `to` of a `wasm` endpoint in a route has its own instance) and that
-instance handles every message, one at a time, so data the module keeps in its memory or globals carries over from one
-message to the next.
-
-* When the function returns with the error bit set, the error data is released with `dealloc` using its size without
-the error bit, the error data becomes the message of the exception, and the instance is kept.
-* When the call fails without returning, for example on a Rust `panic` or any other Wasm trap, the instance is
-discarded and a new one is created from the module for the next message. Whatever the failed call allocated can no
-longer be released, so keeping the instance would grow its memory with every failure. Any state the module kept in
-the discarded instance is lost.
-On a route where every call fails this way, the module is instantiated again for every message (its data segments
-are copied and any start function runs), which costs more than a call on a kept instance.
+Built with `GOOS=wasip1 GOARCH=wasm go build -o upper.wasm`.
+
+=== Running a module
+
+For every exchange the component:
+
+. creates a new instance of the module (the module is parsed once, when the producer starts),
+. runs its `_start` function with the message body as standard input, the arguments configured with `args` and
+the environment variables configured with `environment` and `environmentHeaders`,
+. sets the `CamelWasmExitCode` header to the exit code of the program: 0 when `_start` returns, otherwise the code
+the program passed to `proc_exit`,
+. sets the `CamelWasmStderr` header to what the program wrote to its standard error, if anything,
+. and replaces the message body with the standard output of the program, as a `byte[]`.
+
+When the program exits with a non-zero code, the exchange fails with a `WasmExitCodeException` (which holds the exit
+code and the standard error) and the standard output is not set as the body, unless `failOnNonZeroExit` is `false`:
+then the standard output replaces the body as for any other exit code. The other headers of the message are kept.
+See <<_the_message_body_after_a_failure>> for what the body holds when the exchange fails.
+
+Every exchange runs on its own instance, so exchanges never share the memory or the globals of the module and can
+run in parallel without a lock. A module cannot keep state from one message to the next.
+
+The `CamelWasmExitCode` and `CamelWasmStderr` headers always describe the current run: a value already on the
+message is removed before the module runs.
+
+=== What the module can access
+
+* *Standard input, output and error*: the message body, the new body and the `CamelWasmStderr` header. They are
+not terminals.
+* *Arguments*: the file name of the module as argument zero (`upper.wasm` for `wasm:file:/opt/modules/upper.wasm`;
+the directories, the host, user info and query of the location are not passed on), followed by the `args` option, for
+example `args=--level 2`. Use double quotes around an argument that contains spaces. Arguments are fixed by the route.
+* *Environment variables*: only the ones configured with `environment.NAME=value`, and the message headers listed in
+`environmentHeaders`, passed under the name they are listed with. Header names are matched the way Camel matches
+them, ignoring case: `environmentHeaders=TENANT` also passes a header named `tenant`, as the variable `TENANT`. A
+listed header that is not on the message is not set. Headers usually come from whoever sent the message, so no
+header reaches the module unless it is listed. The environment of the JVM is never passed on.
+* *Files and network*: none. The module gets no preopened directories, so it cannot open files, and no sockets.
+* *Clock*: the one of the host.
+* *Random numbers*: WASI `random_get` (used by Go `crypto/rand` and Rust `getrandom`) returns bytes from a
+`java.security.SecureRandom` of the host.
+
+=== Limits
+
+The following options bound what one run of a module can use. They are off by default.
+
+* `maxMemoryPages`: the maximum size of the memory of the module, in pages of 64 KiB. Rust and Go programs declare
+no maximum by default, so the memory of one run can grow to gigabytes of JVM heap. When the limit is reached, growing
+the memory fails inside the module, which then usually traps or exits with an error.
+* `timeout`: the maximum time one run may take. The module then runs on a thread of the producer while the route
+thread waits. When the time is up, the module is interrupted at its next function call or loop iteration and the
+exchange fails with an `ExchangeTimedOutException`. Without a `timeout`, a module that never ends blocks the route
+thread.
+* `maxOutputSize`: the maximum number of bytes the module may write to its standard output, and separately to its
+standard error. Both are buffered in memory. A module that writes more is stopped and the exchange fails.
+
+WARNING: Without `maxOutputSize` the standard output and the standard error are buffered without a limit: a module
+that keeps writing fills the heap of the JVM, which affects the whole application and not only one exchange. Set
+`maxOutputSize` unless the output size of the module is known and small, for example `maxOutputSize=16777216`
+(16 MiB).
+
+With a `timeout`, the message body is read into memory, on the route thread, before the module starts, and the module
+reads that copy. Without a `timeout` the body is streamed to the module. The copy is what makes the timeout safe: the
+thread of a module that timed out can run for a moment longer, and it must not read a stream that the route is using
+again. It also means that:
+
+* the whole body is on the heap for the time of the run, in addition to the buffered output,
+* the time to read the body is not part of the `timeout`: a body stream that delivers slowly, such as a slow network
+connection, holds the route thread as it does without a `timeout`,
+* the threads of the producer are never blocked on the body, so a module that timed out gives its thread back at its
+next function call or loop iteration. The pool of these threads has no maximum size and needs none: it holds one
+thread for each exchange that is in the endpoint at the same time.
+
+These limits are robustness controls, not a sandbox. A module is trusted code chosen by the route author, like a
+processor or a bean, as described in the xref:manual::security-model.adoc[security model]: the limits keep a
+misbehaving module from turning one message into an unbounded use of memory, time or threads, but they are not a
+defence against a module written to attack the host.
+
+=== Errors
+
+[width="100%",cols="2,4",options="header"]
+|===
+| What happens | Result
+| The program exits with code 0 | The body is the standard output. `CamelWasmExitCode` is 0.
+| The program exits with another code | A `WasmExitCodeException`, no new body, `CamelWasmExitCode` and
+`CamelWasmStderr` set. With `failOnNonZeroExit=false`: the body is the standard output.
+| The program traps, for example on a Rust `panic` | A `CamelExchangeException` caused by the `TrapException`
+of Endive, with the start of the standard error in its message. `CamelWasmStderr` is set, `CamelWasmExitCode` is not.
+| The program writes more than `maxOutputSize` | A `CamelExchangeException`. `CamelWasmStderr` is set when the program
+wrote to its standard error, `CamelWasmExitCode` is not.
+| The program runs longer than `timeout` | An `ExchangeTimedOutException`. Neither header is set.
+| The module is not a WASI Preview 1 command module, or the options are invalid | The route fails to start.
+|===
+
+In every case, the instance the program ran on is discarded.
+
+=== The message body after a failure
+
+When the exchange fails (a non-zero exit code, a trap, the output limit or the timeout), the component does not set
+a new body, but the program may already have read the old one as its standard input:
+
+[width="100%",cols="2,4",options="header"]
+|===
+| Body before the call | Body when the exchange has failed
+| A `String`, a `byte[]`, a file, or another value that is converted to a stream | Unchanged.
+| A cached stream (a `StreamCache`): what xref:manual::stream-caching.adoc[stream caching], which is on by default,
+makes of a stream body | The same stream cache. The program has read from it; Camel resets it for the error handler
+and for a redelivery, so it can be read again.
+| A stream that is not cached, and no `timeout` | The same stream, read as far as the program read it, usually to
+its end. Its content cannot be read again.
+| A stream that is not cached, with a `timeout` | A `byte[]` with the content of the stream: the body was read into
+memory before the run, and the component puts that copy back.
+|===
+
+So keep stream caching on when an `onException` or a dead letter channel needs the body of a failed exchange and the
+body is a stream.
+
+=== Performance
+
+Creating an instance per exchange is cheap for a small module, but the start-up code of the program itself runs
+for every message. Programs with a large runtime, such as standard Go programs, spend most of each run initializing
+that runtime; TinyGo, Rust and C programs start much faster. Modules are interpreted.
== Examples
-Supposing we have compiled a Wasm module containing the function above, then it can be called in a Camel Route by its name and module resource location:
+A route that upper-cases the message body with the module `upper.wasm` from the classpath:
[tabs]
====
@@ -178,7 +228,7 @@ Java::
[source,java]
----
from("direct:in")
- .to("wasm:process?module=classpath://functions.wasm");
+ .to("wasm:classpath:upper.wasm?timeout=5000&maxMemoryPages=256");
----
XML::
@@ -187,7 +237,7 @@ XML::
----
<route>
<from uri="direct:in"/>
- <to uri="wasm:process?module=classpath://functions.wasm"/>
+ <to uri="wasm:classpath:upper.wasm?timeout=5000&maxMemoryPages=256"/>
</route>
----
@@ -200,8 +250,41 @@ YAML::
uri: direct:in
steps:
- to:
- uri: wasm:process
+ uri: wasm:classpath:upper.wasm
parameters:
- module: classpath://functions.wasm
+ timeout: 5000
+ maxMemoryPages: 256
+----
+====
+
+A module with arguments and environment variables, which also receives the `TENANT` header of the message:
+
+[tabs]
+====
+Java::
++
+[source,java]
+----
+from("kafka:orders")
+ .to("wasm:file:/opt/transforms/normalize.wasm?args=--strict&environment.LOG_LEVEL=info&environmentHeaders=TENANT")
+ .to("kafka:normalized-orders");
+----
+
+YAML::
++
+[source,yaml]
+----
+- route:
+ from:
+ uri: kafka:orders
+ steps:
+ - to:
+ uri: wasm:file:/opt/transforms/normalize.wasm
+ parameters:
+ args: --strict
+ environment.LOG_LEVEL: info
+ environmentHeaders: TENANT
+ - to:
+ uri: kafka:normalized-orders
----
====
diff --git a/components/camel-wasm/pom.xml b/components/camel-wasm/pom.xml
index d340565ec437..dd3cea44b76f 100644
--- a/components/camel-wasm/pom.xml
+++ b/components/camel-wasm/pom.xml
@@ -49,6 +49,11 @@
<artifactId>runtime</artifactId>
<version>${endive-version}</version>
</dependency>
+ <dependency>
+ <groupId>run.endive</groupId>
+ <artifactId>wasi</artifactId>
+ <version>${endive-version}</version>
+ </dependency>
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
diff --git a/components/camel-wasm/src/generated/java/org/apache/camel/component/wasm/WasmConfigurationConfigurer.java b/components/camel-wasm/src/generated/java/org/apache/camel/component/wasm/WasmConfigurationConfigurer.java
index 298a5fa819d6..0666d1cf8d2f 100644
--- a/components/camel-wasm/src/generated/java/org/apache/camel/component/wasm/WasmConfigurationConfigurer.java
+++ b/components/camel-wasm/src/generated/java/org/apache/camel/component/wasm/WasmConfigurationConfigurer.java
@@ -23,7 +23,17 @@ public class WasmConfigurationConfigurer extends org.apache.camel.support.compon
public boolean configure(CamelContext camelContext, Object obj, String name, Object value, boolean ignoreCase) {
org.apache.camel.component.wasm.WasmConfiguration target = (org.apache.camel.component.wasm.WasmConfiguration) obj;
switch (ignoreCase ? name.toLowerCase() : name) {
- case "module": target.setModule(property(camelContext, java.lang.String.class, value)); return true;
+ case "args": target.setArgs(property(camelContext, java.lang.String.class, value)); return true;
+ case "environment": target.setEnvironment(property(camelContext, java.util.Map.class, value)); return true;
+ case "environmentheaders":
+ case "environmentHeaders": target.setEnvironmentHeaders(property(camelContext, java.lang.String.class, value)); return true;
+ case "failonnonzeroexit":
+ case "failOnNonZeroExit": target.setFailOnNonZeroExit(property(camelContext, boolean.class, value)); return true;
+ case "maxmemorypages":
+ case "maxMemoryPages": target.setMaxMemoryPages(property(camelContext, int.class, value)); return true;
+ case "maxoutputsize":
+ case "maxOutputSize": target.setMaxOutputSize(property(camelContext, int.class, value)); return true;
+ case "timeout": target.setTimeout(property(camelContext, long.class, value)); return true;
default: return false;
}
}
@@ -31,7 +41,17 @@ public class WasmConfigurationConfigurer extends org.apache.camel.support.compon
@Override
public Class<?> getOptionType(String name, boolean ignoreCase) {
switch (ignoreCase ? name.toLowerCase() : name) {
- case "module": return java.lang.String.class;
+ case "args": return java.lang.String.class;
+ case "environment": return java.util.Map.class;
+ case "environmentheaders":
+ case "environmentHeaders": return java.lang.String.class;
+ case "failonnonzeroexit":
+ case "failOnNonZeroExit": return boolean.class;
+ case "maxmemorypages":
+ case "maxMemoryPages": return int.class;
+ case "maxoutputsize":
+ case "maxOutputSize": return int.class;
+ case "timeout": return long.class;
default: return null;
}
}
@@ -40,7 +60,25 @@ public class WasmConfigurationConfigurer extends org.apache.camel.support.compon
public Object getOptionValue(Object obj, String name, boolean ignoreCase) {
org.apache.camel.component.wasm.WasmConfiguration target = (org.apache.camel.component.wasm.WasmConfiguration) obj;
switch (ignoreCase ? name.toLowerCase() : name) {
- case "module": return target.getModule();
+ case "args": return target.getArgs();
+ case "environment": return target.getEnvironment();
+ case "environmentheaders":
+ case "environmentHeaders": return target.getEnvironmentHeaders();
+ case "failonnonzeroexit":
+ case "failOnNonZeroExit": return target.isFailOnNonZeroExit();
+ case "maxmemorypages":
+ case "maxMemoryPages": return target.getMaxMemoryPages();
+ case "maxoutputsize":
+ case "maxOutputSize": return target.getMaxOutputSize();
+ case "timeout": return target.getTimeout();
+ default: return null;
+ }
+ }
+
+ @Override
+ public Object getCollectionValueType(Object target, String name, boolean ignoreCase) {
+ switch (ignoreCase ? name.toLowerCase() : name) {
+ case "environment": return java.lang.Object.class;
default: return null;
}
}
diff --git a/components/camel-wasm/src/generated/java/org/apache/camel/component/wasm/WasmEndpointConfigurer.java b/components/camel-wasm/src/generated/java/org/apache/camel/component/wasm/WasmEndpointConfigurer.java
index 75168668e2c1..66178fed5e04 100644
--- a/components/camel-wasm/src/generated/java/org/apache/camel/component/wasm/WasmEndpointConfigurer.java
+++ b/components/camel-wasm/src/generated/java/org/apache/camel/component/wasm/WasmEndpointConfigurer.java
@@ -23,9 +23,19 @@ public class WasmEndpointConfigurer extends PropertyConfigurerSupport implements
public boolean configure(CamelContext camelContext, Object obj, String name, Object value, boolean ignoreCase) {
WasmEndpoint target = (WasmEndpoint) obj;
switch (ignoreCase ? name.toLowerCase() : name) {
+ case "args": target.getConfiguration().setArgs(property(camelContext, java.lang.String.class, value)); return true;
+ case "environment": target.getConfiguration().setEnvironment(property(camelContext, java.util.Map.class, value)); return true;
+ case "environmentheaders":
+ case "environmentHeaders": target.getConfiguration().setEnvironmentHeaders(property(camelContext, java.lang.String.class, value)); return true;
+ case "failonnonzeroexit":
+ case "failOnNonZeroExit": target.getConfiguration().setFailOnNonZeroExit(property(camelContext, boolean.class, value)); return true;
case "lazystartproducer":
case "lazyStartProducer": target.setLazyStartProducer(property(camelContext, boolean.class, value)); return true;
- case "module": target.getConfiguration().setModule(property(camelContext, java.lang.String.class, value)); return true;
+ case "maxmemorypages":
+ case "maxMemoryPages": target.getConfiguration().setMaxMemoryPages(property(camelContext, int.class, value)); return true;
+ case "maxoutputsize":
+ case "maxOutputSize": target.getConfiguration().setMaxOutputSize(property(camelContext, int.class, value)); return true;
+ case "timeout": target.getConfiguration().setTimeout(property(camelContext, java.time.Duration.class, value).toMillis()); return true;
default: return false;
}
}
@@ -33,9 +43,19 @@ public class WasmEndpointConfigurer extends PropertyConfigurerSupport implements
@Override
public Class<?> getOptionType(String name, boolean ignoreCase) {
switch (ignoreCase ? name.toLowerCase() : name) {
+ case "args": return java.lang.String.class;
+ case "environment": return java.util.Map.class;
+ case "environmentheaders":
+ case "environmentHeaders": return java.lang.String.class;
+ case "failonnonzeroexit":
+ case "failOnNonZeroExit": return boolean.class;
case "lazystartproducer":
case "lazyStartProducer": return boolean.class;
- case "module": return java.lang.String.class;
+ case "maxmemorypages":
+ case "maxMemoryPages": return int.class;
+ case "maxoutputsize":
+ case "maxOutputSize": return int.class;
+ case "timeout": return long.class;
default: return null;
}
}
@@ -44,9 +64,27 @@ public class WasmEndpointConfigurer extends PropertyConfigurerSupport implements
public Object getOptionValue(Object obj, String name, boolean ignoreCase) {
WasmEndpoint target = (WasmEndpoint) obj;
switch (ignoreCase ? name.toLowerCase() : name) {
+ case "args": return target.getConfiguration().getArgs();
+ case "environment": return target.getConfiguration().getEnvironment();
+ case "environmentheaders":
+ case "environmentHeaders": return target.getConfiguration().getEnvironmentHeaders();
+ case "failonnonzeroexit":
+ case "failOnNonZeroExit": return target.getConfiguration().isFailOnNonZeroExit();
case "lazystartproducer":
case "lazyStartProducer": return target.isLazyStartProducer();
- case "module": return target.getConfiguration().getModule();
+ case "maxmemorypages":
+ case "maxMemoryPages": return target.getConfiguration().getMaxMemoryPages();
+ case "maxoutputsize":
+ case "maxOutputSize": return target.getConfiguration().getMaxOutputSize();
+ case "timeout": return target.getConfiguration().getTimeout();
+ default: return null;
+ }
+ }
+
+ @Override
+ public Object getCollectionValueType(Object target, String name, boolean ignoreCase) {
+ switch (ignoreCase ? name.toLowerCase() : name) {
+ case "environment": return java.lang.Object.class;
default: return null;
}
}
diff --git a/components/camel-wasm/src/generated/java/org/apache/camel/component/wasm/WasmEndpointUriFactory.java b/components/camel-wasm/src/generated/java/org/apache/camel/component/wasm/WasmEndpointUriFactory.java
index 4f9dbe13a95e..e1f028bd1f60 100644
--- a/components/camel-wasm/src/generated/java/org/apache/camel/component/wasm/WasmEndpointUriFactory.java
+++ b/components/camel-wasm/src/generated/java/org/apache/camel/component/wasm/WasmEndpointUriFactory.java
@@ -17,21 +17,29 @@ import org.apache.camel.spi.EndpointUriFactory;
@Generated("org.apache.camel.maven.packaging.GenerateEndpointUriFactoryMojo")
public class WasmEndpointUriFactory extends org.apache.camel.support.component.EndpointUriFactorySupport implements EndpointUriFactory {
- private static final String BASE = ":functionName";
+ private static final String BASE = ":module";
private static final Set<String> PROPERTY_NAMES;
private static final Set<String> SECRET_PROPERTY_NAMES;
private static final Set<String> ENDPOINT_IDENTITY_PROPERTY_NAMES;
private static final Map<String, String> MULTI_VALUE_PREFIXES;
static {
- Set<String> props = new HashSet<>(3);
- props.add("functionName");
+ Set<String> props = new HashSet<>(9);
+ props.add("args");
+ props.add("environment");
+ props.add("environmentHeaders");
+ props.add("failOnNonZeroExit");
props.add("lazyStartProducer");
+ props.add("maxMemoryPages");
+ props.add("maxOutputSize");
props.add("module");
+ props.add("timeout");
PROPERTY_NAMES = Collections.unmodifiableSet(props);
SECRET_PROPERTY_NAMES = Collections.emptySet();
ENDPOINT_IDENTITY_PROPERTY_NAMES = Collections.emptySet();
- MULTI_VALUE_PREFIXES = Collections.emptyMap();
+ Map<String, String> prefixes = new HashMap<>(1);
+ prefixes.put("environment", "environment.");
+ MULTI_VALUE_PREFIXES = Collections.unmodifiableMap(prefixes);
}
@Override
@@ -46,7 +54,7 @@ public class WasmEndpointUriFactory extends org.apache.camel.support.component.E
Map<String, Object> copy = copyParameters(properties);
- uri = buildPathParameter(syntax, uri, "functionName", null, true, copy);
+ uri = buildPathParameter(syntax, uri, "module", null, true, copy);
uri = buildQueryParameters(uri, copy, encode);
return uri;
}
diff --git a/components/camel-wasm/src/generated/resources/META-INF/org/apache/camel/component/wasm/wasm.json b/components/camel-wasm/src/generated/resources/META-INF/org/apache/camel/component/wasm/wasm.json
index c931e1190ccf..076279f11820 100644
--- a/components/camel-wasm/src/generated/resources/META-INF/org/apache/camel/component/wasm/wasm.json
+++ b/components/camel-wasm/src/generated/resources/META-INF/org/apache/camel/component/wasm/wasm.json
@@ -3,7 +3,7 @@
"kind": "component",
"name": "wasm",
"title": "Wasm",
- "description": "Invoke Wasm functions.",
+ "description": "Run Wasm programs (WASI Preview 1 command modules) with the message body as standard input.",
"deprecated": false,
"firstVersion": "4.4.0",
"label": "core,script",
@@ -14,7 +14,7 @@
"version": "4.23.0-SNAPSHOT",
"scheme": "wasm",
"extendsScheme": "",
- "syntax": "wasm:functionName",
+ "syntax": "wasm:module",
"async": false,
"api": false,
"consumerOnly": false,
@@ -27,9 +27,19 @@
"lazyStartProducer": { "index": 0, "kind": "property", "displayName": "Lazy Start Producer", "group": "producer", "label": "producer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether the producer should be started lazy (on the first message). By starting lazy you can use this to allow CamelContext and routes to startup in situations where a producer may otherwise fail [...]
"autowiredEnabled": { "index": 1, "kind": "property", "displayName": "Autowired Enabled", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "description": "Whether autowiring is enabled. This is used for automatic autowiring options (the option must be marked as autowired) by looking up in the registry to find if there is a single instance of matching t [...]
},
+ "headers": {
+ "CamelWasmExitCode": { "index": 0, "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "Integer", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The exit code of the module: 0 when it returned from its _start function, otherwise the value it passed to proc_exit. Not set when the module trapped, timed out or exceeded maxOutputSize.", "constantName": "org.apache.camel.wasm.Wasm$Headers#E [...]
+ "CamelWasmStderr": { "index": 1, "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "What the module wrote to its standard error, decoded as UTF-8. Not set when the module wrote nothing or timed out.", "constantName": "org.apache.camel.wasm.Wasm$Headers#STDERR" }
+ },
"properties": {
- "functionName": { "index": 0, "kind": "path", "displayName": "Function Name", "group": "producer", "label": "", "required": true, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The Function Name" },
- "module": { "index": 1, "kind": "parameter", "displayName": "Module", "group": "producer", "label": "", "required": true, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.wasm.WasmConfiguration", "configurationField": "configuration", "description": "Set the module (the distributable, loadable, and executable unit of code in WebAssembly) resource that pr [...]
- "lazyStartProducer": { "index": 2, "kind": "parameter", "displayName": "Lazy Start Producer", "group": "producer (advanced)", "label": "producer,advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether the producer should be started lazy (on the first message). By starting lazy you can use this to allow CamelContext and routes to startup in situations where a produc [...]
+ "module": { "index": 0, "kind": "path", "displayName": "Module", "group": "producer", "label": "", "required": true, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The WASI Preview 1 command module to run, as a resource: by default from the classpath, or with a prefix such as classpath:, file: or http:" },
+ "args": { "index": 1, "kind": "parameter", "displayName": "Args", "group": "producer", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.wasm.WasmConfiguration", "configurationField": "configuration", "description": "The command line arguments passed to the module, separated by whitespace. Use double quotes around an argument that contains whites [...]
+ "environment": { "index": 2, "kind": "parameter", "displayName": "Environment", "group": "producer", "label": "", "required": false, "type": "object", "javaType": "java.util.Map<java.lang.String, java.lang.Object>", "prefix": "environment.", "multiValue": true, "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.wasm.WasmConfiguration", "configurationField": "configuration", "description": "Environment variables for the module, [...]
+ "environmentHeaders": { "index": 3, "kind": "parameter", "displayName": "Environment Headers", "group": "producer", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.wasm.WasmConfiguration", "configurationField": "configuration", "description": "A comma separated allow-list of message headers whose values are passed to the module as environment v [...]
+ "failOnNonZeroExit": { "index": 4, "kind": "parameter", "displayName": "Fail On Non Zero Exit", "group": "producer", "label": "", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "configurationClass": "org.apache.camel.component.wasm.WasmConfiguration", "configurationField": "configuration", "description": "Whether a module that exits with a non-zero exit code fails the exchange with a WasmExi [...]
+ "maxMemoryPages": { "index": 5, "kind": "parameter", "displayName": "Max Memory Pages", "group": "producer", "label": "", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.wasm.WasmConfiguration", "configurationField": "configuration", "description": "The maximum size of the memory of the module, in WebAssembly pages of 64 KiB. Rust and Go programs declare no maximum by [...]
+ "maxOutputSize": { "index": 6, "kind": "parameter", "displayName": "Max Output Size", "group": "producer", "label": "", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.wasm.WasmConfiguration", "configurationField": "configuration", "description": "The maximum number of bytes the module may write to its standard output, and separately to its standard error, which are bu [...]
+ "timeout": { "index": 7, "kind": "parameter", "displayName": "Timeout", "group": "producer", "label": "", "required": false, "type": "duration", "javaType": "long", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.wasm.WasmConfiguration", "configurationField": "configuration", "description": "The maximum time the module may run for one exchange, in milliseconds. When it is exceeded the module is interrupted and the exchange f [...]
+ "lazyStartProducer": { "index": 8, "kind": "parameter", "displayName": "Lazy Start Producer", "group": "producer (advanced)", "label": "producer,advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether the producer should be started lazy (on the first message). By starting lazy you can use this to allow CamelContext and routes to startup in situations where a produc [...]
}
}
diff --git a/components/camel-wasm/src/main/docs/wasm-component.adoc b/components/camel-wasm/src/main/docs/wasm-component.adoc
index ad9aa3daf956..27b588693644 100644
--- a/components/camel-wasm/src/main/docs/wasm-component.adoc
+++ b/components/camel-wasm/src/main/docs/wasm-component.adoc
@@ -2,7 +2,7 @@
:doctitle: Wasm
:shortname: wasm
:artifactid: camel-wasm
-:description: Invoke Wasm functions.
+:description: Run Wasm programs (WASI Preview 1 command modules) with the message body as standard input.
:since: 4.4
:supportlevel: Experimental
:tabs-sync-option:
@@ -13,9 +13,15 @@
*{component-header}*
-https://webassembly.org[WebAssembly (Wasm)] is a low-level bytecode format designed as a portable target for the compilation of high-level languages like C, C++, and Rust, enabling deployment on the web for client and server applications.
+https://webassembly.org[WebAssembly (Wasm)] is a portable bytecode format that languages such as Rust, Go, C and C++
+compile to. https://wasi.dev[WASI] is the system interface those programs use for their standard input, output,
+arguments and environment.
-This component provides support to leverage Wasm functions for message transformation.
+The Wasm component runs a WASI Preview 1 _command module_ (a program with a `_start` function) as a processor: the
+message body is the standard input of the program, and what the program writes to its standard output becomes the
+message body. Any program built for WASI Preview 1 can be used as it is, without a Camel specific API. The module runs
+inside the JVM on https://github.com/bytecodealliance/endive[Endive], a WebAssembly runtime written in Java, so no native
+code or external process is involved.
Maven users will need to add the following dependency to their `pom.xml`
for this component:
@@ -32,9 +38,18 @@ for this component:
== URI format
----
-wasm://functionName?[options]
+wasm:module[?options]
----
+The `module` is the location of the `.wasm` file, loaded as a resource: from the classpath by default, or with a
+prefix such as `classpath:`, `file:` or `http:`. For example `wasm:classpath:upper.wasm` or
+`wasm:file:/opt/transforms/upper.wasm`.
+
+WARNING: Use a static URI. With a dynamic one, such as `toD("wasm:${header.module}")`, data of the message chooses the
+program that runs, including a module downloaded from an `http:` location, and the module is loaded and parsed again
+for every new URI (0.2 to 0.4 seconds for a Go program). To choose between known modules, use a
+xref:eips:choice-eip.adoc[Choice] with one static `wasm` endpoint per module.
+
// component options: START
include::partial$component-configure-options.adoc[]
include::partial$component-endpoint-options.adoc[]
@@ -43,133 +58,168 @@ include::partial$component-endpoint-headers.adoc[]
== Usage
-=== Writing A Wasm processor
-
-In _Wasm_, sharing objects between the host, in this case the _JVM_, and the _Wasm_ module is deliberately restricted and as of today, it requires a number of steps:
-
-1. From the _host_, call a function inside the webassembly module that allocates a block of memory and returns its address, then save it
-2. From the _host_, write the data that should be exchanged with the _Wasm_ module to the saved address
-3. From the _host_, invoke the required function passing both the address where the data is written and its size
-4. From the _Wasm_ module, read the data and process it
-5. From the _host_, release the memory when done
+=== Writing a module
-==== Providing functions for memory management
-
-The module hosting the function *must* provide the functions to allocate/deallocate memory that *must* be named `alloc` and `dealloc` respectively.
-
-Here's an example of the mentioned functions implemented in https://www.rust-lang.org[Rust]:
+A module is an ordinary program that reads its standard input and writes its standard output. For example, in
+https://www.rust-lang.org[Rust]:
[source,rust]
----
-pub extern "C" fn alloc(size: u32) -> *mut u8 {
- let mut buf = Vec::with_capacity(size as usize);
- let ptr = buf.as_mut_ptr();
-
- // tell Rust not to clean this up
- mem::forget(buf);
+use std::io::{self, Read, Write};
- ptr
-}
-
-pub unsafe extern "C" fn dealloc(ptr: &mut u8, len: i32) {
- // Retakes the pointer which allows its memory to be freed.
- let _ = Vec::from_raw_parts(ptr, 0, len as usize);
+fn main() {
+ let mut input = String::new();
+ io::stdin().read_to_string(&mut input).unwrap();
+ io::stdout().write_all(input.to_uppercase().as_bytes()).unwrap();
}
----
-==== Data shapes
+Built with `rustup target add wasm32-wasip1` and `cargo build --release --target wasm32-wasip1`.
-It is not possible to share a Java object with the Wasm module directly, and as mentioned before,
-data exchange leverages Wasm's memory that can be accessed by both the host and the guest runtimes.
-At this stage, the data structure that the component exchange with the Wasm function is a subset of the Apache Camel Message,
-containing headers the body encoded as a base64 string:
+The same program in https://go.dev[Go]:
-._Java-only: data exchange wrapper class for Wasm module communication_
-[source,java]
+[source,go]
----
-public static class Wrapper {
- @JsonProperty
- public Map<String, String> headers = new HashMap<>();
-
- @JsonProperty
- public byte[] body;
-}
-----
-
-==== Data processing
-
-The component expects the processing function to have the following signature:
-
-[source,rust]
-----
-fn function(ptr: u32, len: u32) -> u64
-----
-
-* it accepts two 32bit unsigned integers arguments
-** a pointer to the memory location when the input data has been written (`ptr`)
-** the size of the input data (`len`)
-* it returns a 64bit unsigned integer where:
-** the first 32bit represents a pointer to the return data
-** the last 31bit represents the size of the return data, which must have the same data shape discussed above unless there is an error (see below)
-** the most significant bit of the returned data size is reserved to signal an error, so if it is set, then the return data could contain an error message/code/etc
-
-Here's an example of a complete function:
-
-[source,rust]
-----
-#[derive(Serialize, Deserialize)]
-struct Message {
- headers: HashMap<String, serde_json::Value>,
-
- #[serde(with = "Base64Standard")]
- body: Vec<u8>,
-}
-
-#[cfg_attr(all(target_arch = "wasm32"), export_name = "process")]
-#[no_mangle]
-pub extern fn process(ptr: u32, len: u32) -> u64 {
- let bytes = unsafe {
- slice::from_raw_parts_mut(
- ptr as *mut u8,
- len as usize)
- };
+package main
- let mut msg: Message = serde_json::from_slice(bytes).unwrap();
+import (
+ "bytes"
+ "io"
+ "os"
+)
-
- let out_vec = serde_json::to_vec(&msg).unwrap();
- let out_len = out_vec.len();
- let out_ptr = alloc(out_len as u32);
-
- unsafe {
- std::ptr::copy_nonoverlapping(
- out_vec.as_ptr(),
- out_ptr,
- out_len as usize)
- };
-
- return ((out_ptr as u64) << 32) | out_len as u64;
+func main() {
+ input, _ := io.ReadAll(os.Stdin)
+ os.Stdout.Write(bytes.ToUpper(input))
}
----
-==== Errors and module state
-
-The module is instantiated once per producer (each `to` of a `wasm` endpoint in a route has its own instance) and that
-instance handles every message, one at a time, so data the module keeps in its memory or globals carries over from one
-message to the next.
-
-* When the function returns with the error bit set, the error data is released with `dealloc` using its size without
-the error bit, the error data becomes the message of the exception, and the instance is kept.
-* When the call fails without returning, for example on a Rust `panic` or any other Wasm trap, the instance is
-discarded and a new one is created from the module for the next message. Whatever the failed call allocated can no
-longer be released, so keeping the instance would grow its memory with every failure. Any state the module kept in
-the discarded instance is lost.
-On a route where every call fails this way, the module is instantiated again for every message (its data segments
-are copied and any start function runs), which costs more than a call on a kept instance.
+Built with `GOOS=wasip1 GOARCH=wasm go build -o upper.wasm`.
+
+=== Running a module
+
+For every exchange the component:
+
+. creates a new instance of the module (the module is parsed once, when the producer starts),
+. runs its `_start` function with the message body as standard input, the arguments configured with `args` and
+the environment variables configured with `environment` and `environmentHeaders`,
+. sets the `CamelWasmExitCode` header to the exit code of the program: 0 when `_start` returns, otherwise the code
+the program passed to `proc_exit`,
+. sets the `CamelWasmStderr` header to what the program wrote to its standard error, if anything,
+. and replaces the message body with the standard output of the program, as a `byte[]`.
+
+When the program exits with a non-zero code, the exchange fails with a `WasmExitCodeException` (which holds the exit
+code and the standard error) and the standard output is not set as the body, unless `failOnNonZeroExit` is `false`:
+then the standard output replaces the body as for any other exit code. The other headers of the message are kept.
+See <<_the_message_body_after_a_failure>> for what the body holds when the exchange fails.
+
+Every exchange runs on its own instance, so exchanges never share the memory or the globals of the module and can
+run in parallel without a lock. A module cannot keep state from one message to the next.
+
+The `CamelWasmExitCode` and `CamelWasmStderr` headers always describe the current run: a value already on the
+message is removed before the module runs.
+
+=== What the module can access
+
+* *Standard input, output and error*: the message body, the new body and the `CamelWasmStderr` header. They are
+not terminals.
+* *Arguments*: the file name of the module as argument zero (`upper.wasm` for `wasm:file:/opt/modules/upper.wasm`;
+the directories, the host, user info and query of the location are not passed on), followed by the `args` option, for
+example `args=--level 2`. Use double quotes around an argument that contains spaces. Arguments are fixed by the route.
+* *Environment variables*: only the ones configured with `environment.NAME=value`, and the message headers listed in
+`environmentHeaders`, passed under the name they are listed with. Header names are matched the way Camel matches
+them, ignoring case: `environmentHeaders=TENANT` also passes a header named `tenant`, as the variable `TENANT`. A
+listed header that is not on the message is not set. Headers usually come from whoever sent the message, so no
+header reaches the module unless it is listed. The environment of the JVM is never passed on.
+* *Files and network*: none. The module gets no preopened directories, so it cannot open files, and no sockets.
+* *Clock*: the one of the host.
+* *Random numbers*: WASI `random_get` (used by Go `crypto/rand` and Rust `getrandom`) returns bytes from a
+`java.security.SecureRandom` of the host.
+
+=== Limits
+
+The following options bound what one run of a module can use. They are off by default.
+
+* `maxMemoryPages`: the maximum size of the memory of the module, in pages of 64 KiB. Rust and Go programs declare
+no maximum by default, so the memory of one run can grow to gigabytes of JVM heap. When the limit is reached, growing
+the memory fails inside the module, which then usually traps or exits with an error.
+* `timeout`: the maximum time one run may take. The module then runs on a thread of the producer while the route
+thread waits. When the time is up, the module is interrupted at its next function call or loop iteration and the
+exchange fails with an `ExchangeTimedOutException`. Without a `timeout`, a module that never ends blocks the route
+thread.
+* `maxOutputSize`: the maximum number of bytes the module may write to its standard output, and separately to its
+standard error. Both are buffered in memory. A module that writes more is stopped and the exchange fails.
+
+WARNING: Without `maxOutputSize` the standard output and the standard error are buffered without a limit: a module
+that keeps writing fills the heap of the JVM, which affects the whole application and not only one exchange. Set
+`maxOutputSize` unless the output size of the module is known and small, for example `maxOutputSize=16777216`
+(16 MiB).
+
+With a `timeout`, the message body is read into memory, on the route thread, before the module starts, and the module
+reads that copy. Without a `timeout` the body is streamed to the module. The copy is what makes the timeout safe: the
+thread of a module that timed out can run for a moment longer, and it must not read a stream that the route is using
+again. It also means that:
+
+* the whole body is on the heap for the time of the run, in addition to the buffered output,
+* the time to read the body is not part of the `timeout`: a body stream that delivers slowly, such as a slow network
+connection, holds the route thread as it does without a `timeout`,
+* the threads of the producer are never blocked on the body, so a module that timed out gives its thread back at its
+next function call or loop iteration. The pool of these threads has no maximum size and needs none: it holds one
+thread for each exchange that is in the endpoint at the same time.
+
+These limits are robustness controls, not a sandbox. A module is trusted code chosen by the route author, like a
+processor or a bean, as described in the xref:manual::security-model.adoc[security model]: the limits keep a
+misbehaving module from turning one message into an unbounded use of memory, time or threads, but they are not a
+defence against a module written to attack the host.
+
+=== Errors
+
+[width="100%",cols="2,4",options="header"]
+|===
+| What happens | Result
+| The program exits with code 0 | The body is the standard output. `CamelWasmExitCode` is 0.
+| The program exits with another code | A `WasmExitCodeException`, no new body, `CamelWasmExitCode` and
+`CamelWasmStderr` set. With `failOnNonZeroExit=false`: the body is the standard output.
+| The program traps, for example on a Rust `panic` | A `CamelExchangeException` caused by the `TrapException`
+of Endive, with the start of the standard error in its message. `CamelWasmStderr` is set, `CamelWasmExitCode` is not.
+| The program writes more than `maxOutputSize` | A `CamelExchangeException`. `CamelWasmStderr` is set when the program
+wrote to its standard error, `CamelWasmExitCode` is not.
+| The program runs longer than `timeout` | An `ExchangeTimedOutException`. Neither header is set.
+| The module is not a WASI Preview 1 command module, or the options are invalid | The route fails to start.
+|===
+
+In every case, the instance the program ran on is discarded.
+
+=== The message body after a failure
+
+When the exchange fails (a non-zero exit code, a trap, the output limit or the timeout), the component does not set
+a new body, but the program may already have read the old one as its standard input:
+
+[width="100%",cols="2,4",options="header"]
+|===
+| Body before the call | Body when the exchange has failed
+| A `String`, a `byte[]`, a file, or another value that is converted to a stream | Unchanged.
+| A cached stream (a `StreamCache`): what xref:manual::stream-caching.adoc[stream caching], which is on by default,
+makes of a stream body | The same stream cache. The program has read from it; Camel resets it for the error handler
+and for a redelivery, so it can be read again.
+| A stream that is not cached, and no `timeout` | The same stream, read as far as the program read it, usually to
+its end. Its content cannot be read again.
+| A stream that is not cached, with a `timeout` | A `byte[]` with the content of the stream: the body was read into
+memory before the run, and the component puts that copy back.
+|===
+
+So keep stream caching on when an `onException` or a dead letter channel needs the body of a failed exchange and the
+body is a stream.
+
+=== Performance
+
+Creating an instance per exchange is cheap for a small module, but the start-up code of the program itself runs
+for every message. Programs with a large runtime, such as standard Go programs, spend most of each run initializing
+that runtime; TinyGo, Rust and C programs start much faster. Modules are interpreted.
== Examples
-Supposing we have compiled a Wasm module containing the function above, then it can be called in a Camel Route by its name and module resource location:
+A route that upper-cases the message body with the module `upper.wasm` from the classpath:
[tabs]
====
@@ -178,7 +228,7 @@ Java::
[source,java]
----
from("direct:in")
- .to("wasm:process?module=classpath://functions.wasm");
+ .to("wasm:classpath:upper.wasm?timeout=5000&maxMemoryPages=256");
----
XML::
@@ -187,7 +237,7 @@ XML::
----
<route>
<from uri="direct:in"/>
- <to uri="wasm:process?module=classpath://functions.wasm"/>
+ <to uri="wasm:classpath:upper.wasm?timeout=5000&maxMemoryPages=256"/>
</route>
----
@@ -200,8 +250,41 @@ YAML::
uri: direct:in
steps:
- to:
- uri: wasm:process
+ uri: wasm:classpath:upper.wasm
parameters:
- module: classpath://functions.wasm
+ timeout: 5000
+ maxMemoryPages: 256
+----
+====
+
+A module with arguments and environment variables, which also receives the `TENANT` header of the message:
+
+[tabs]
+====
+Java::
++
+[source,java]
+----
+from("kafka:orders")
+ .to("wasm:file:/opt/transforms/normalize.wasm?args=--strict&environment.LOG_LEVEL=info&environmentHeaders=TENANT")
+ .to("kafka:normalized-orders");
+----
+
+YAML::
++
+[source,yaml]
+----
+- route:
+ from:
+ uri: kafka:orders
+ steps:
+ - to:
+ uri: wasm:file:/opt/transforms/normalize.wasm
+ parameters:
+ args: --strict
+ environment.LOG_LEVEL: info
+ environmentHeaders: TENANT
+ - to:
+ uri: kafka:normalized-orders
----
====
diff --git a/components/camel-wasm/src/main/java/org/apache/camel/component/wasm/WasmComponent.java b/components/camel-wasm/src/main/java/org/apache/camel/component/wasm/WasmComponent.java
index ba969314c6fa..b42fa42a8299 100644
--- a/components/camel-wasm/src/main/java/org/apache/camel/component/wasm/WasmComponent.java
+++ b/components/camel-wasm/src/main/java/org/apache/camel/component/wasm/WasmComponent.java
@@ -41,14 +41,6 @@ public class WasmComponent extends DefaultComponent {
this.configuration = new WasmConfiguration();
}
- public String getModule() {
- return configuration.getModule();
- }
-
- public void setModule(String resource) {
- configuration.setModule(resource);
- }
-
@Override
protected Endpoint createEndpoint(
String uri,
@@ -57,7 +49,9 @@ public class WasmComponent extends DefaultComponent {
throws Exception {
if (ObjectHelper.isEmpty(remaining)) {
- throw new IllegalArgumentException("Expecting URI in the form of: 'wasm:functionName', got '" + uri + "'");
+ throw new IllegalArgumentException(
+ "Expecting URI in the form of: 'wasm:module', for example wasm:classpath:upper.wasm, got '" + uri
+ + "'");
}
WasmConfiguration configuration = this.configuration.copy();
diff --git a/components/camel-wasm/src/main/java/org/apache/camel/component/wasm/WasmConfiguration.java b/components/camel-wasm/src/main/java/org/apache/camel/component/wasm/WasmConfiguration.java
index 44895088670f..fc98087ed06f 100644
--- a/components/camel-wasm/src/main/java/org/apache/camel/component/wasm/WasmConfiguration.java
+++ b/components/camel-wasm/src/main/java/org/apache/camel/component/wasm/WasmConfiguration.java
@@ -16,9 +16,11 @@
*/
package org.apache.camel.component.wasm;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
import org.apache.camel.RuntimeCamelException;
import org.apache.camel.spi.Configurer;
-import org.apache.camel.spi.Metadata;
import org.apache.camel.spi.UriParam;
import org.apache.camel.spi.UriParams;
@@ -26,20 +28,118 @@ import org.apache.camel.spi.UriParams;
@UriParams
public class WasmConfiguration implements Cloneable {
- @Metadata(required = true)
@UriParam
- private String module;
+ private String args;
+ @UriParam(prefix = "environment.", multiValue = true)
+ private Map<String, Object> environment;
+ @UriParam
+ private String environmentHeaders;
+ @UriParam(defaultValue = "true")
+ private boolean failOnNonZeroExit = true;
+ @UriParam
+ private int maxMemoryPages;
+ @UriParam
+ private int maxOutputSize;
+ @UriParam(javaType = "java.time.Duration")
+ private long timeout;
+
+ public String getArgs() {
+ return args;
+ }
+
+ /**
+ * The command line arguments passed to the module, separated by whitespace. Use double quotes around an argument
+ * that contains whitespace. The module receives the file name of the module (for example upper.wasm, without
+ * scheme, directories or query) as argument zero, followed by these arguments. The arguments are fixed by the
+ * route: they cannot be set from message headers.
+ */
+ public void setArgs(String args) {
+ this.args = args;
+ }
+
+ public Map<String, Object> getEnvironment() {
+ return environment;
+ }
+
+ /**
+ * Environment variables for the module, for example environment.LOG_LEVEL=debug. The module sees no environment
+ * variables other than these and the headers listed in environmentHeaders; the environment of the JVM is never
+ * passed on.
+ */
+ public void setEnvironment(Map<String, Object> environment) {
+ this.environment = environment;
+ }
+
+ public String getEnvironmentHeaders() {
+ return environmentHeaders;
+ }
+
+ /**
+ * A comma separated allow-list of message headers whose values are passed to the module as environment variables of
+ * the same name. Headers usually come from whoever sent the message, so no header reaches the module unless it is
+ * listed here. A listed header that is not on the message is left unset. Camel matches header names ignoring case,
+ * so listing TENANT also passes a header named tenant, as the variable TENANT. A name may not be listed here and
+ * also be configured with environment.
+ */
+ public void setEnvironmentHeaders(String environmentHeaders) {
+ this.environmentHeaders = environmentHeaders;
+ }
+
+ public boolean isFailOnNonZeroExit() {
+ return failOnNonZeroExit;
+ }
+
+ /**
+ * Whether a module that exits with a non-zero exit code fails the exchange with a WasmExitCodeException. No new
+ * body is set then, but a body that is a stream has been read by the module: it can be read again when handling the
+ * failure if stream caching is on (the default). When false, the standard output of the module replaces the message
+ * body whatever the exit code, which is available in the CamelWasmExitCode header.
+ */
+ public void setFailOnNonZeroExit(boolean failOnNonZeroExit) {
+ this.failOnNonZeroExit = failOnNonZeroExit;
+ }
+
+ public int getMaxMemoryPages() {
+ return maxMemoryPages;
+ }
+
+ /**
+ * The maximum size of the memory of the module, in WebAssembly pages of 64 KiB. Rust and Go programs declare no
+ * maximum by default. When the limit is reached, growing the memory fails inside the module, which then usually
+ * traps or exits with an error. 0 (the default) keeps the maximum declared by the module. The value must not be
+ * lower than the initial size the module declares.
+ */
+ public void setMaxMemoryPages(int maxMemoryPages) {
+ this.maxMemoryPages = maxMemoryPages;
+ }
+
+ public int getMaxOutputSize() {
+ return maxOutputSize;
+ }
+
+ /**
+ * The maximum number of bytes the module may write to its standard output, and separately to its standard error,
+ * which are buffered in memory. A module that writes more is stopped and the exchange fails. 0 (the default) means
+ * no limit: a module that keeps writing then fills the heap of the JVM, so set a limit for modules whose output
+ * size is not known.
+ */
+ public void setMaxOutputSize(int maxOutputSize) {
+ this.maxOutputSize = maxOutputSize;
+ }
- public String getModule() {
- return module;
+ public long getTimeout() {
+ return timeout;
}
/**
- * Set the module (the distributable, loadable, and executable unit of code in WebAssembly) resource that provides
- * the producer function.
+ * The maximum time the module may run for one exchange, in milliseconds. When it is exceeded the module is
+ * interrupted and the exchange fails with an ExchangeTimedOutException. When a timeout is set, the module runs on a
+ * separate thread while the calling thread waits, and the message body is read into memory before the module
+ * starts, instead of being streamed to it. The time to read the body does not count. 0 (the default) means no
+ * timeout.
*/
- public void setModule(String module) {
- this.module = module;
+ public void setTimeout(long timeout) {
+ this.timeout = timeout;
}
// ************************
@@ -50,7 +150,12 @@ public class WasmConfiguration implements Cloneable {
public WasmConfiguration copy() {
try {
- return (WasmConfiguration) super.clone();
+ WasmConfiguration answer = (WasmConfiguration) super.clone();
+ if (environment != null) {
+ // each endpoint gets its own map
+ answer.environment = new LinkedHashMap<>(environment);
+ }
+ return answer;
} catch (CloneNotSupportedException e) {
throw new RuntimeCamelException(e);
}
diff --git a/components/camel-wasm/src/main/java/org/apache/camel/component/wasm/WasmEndpoint.java b/components/camel-wasm/src/main/java/org/apache/camel/component/wasm/WasmEndpoint.java
index 79512b58311a..96393a04c4ce 100644
--- a/components/camel-wasm/src/main/java/org/apache/camel/component/wasm/WasmEndpoint.java
+++ b/components/camel-wasm/src/main/java/org/apache/camel/component/wasm/WasmEndpoint.java
@@ -29,23 +29,24 @@ import org.apache.camel.support.DefaultEndpoint;
import org.apache.camel.wasm.Wasm;
/**
- * Invoke Wasm functions.
+ * Run Wasm programs (WASI Preview 1 command modules) with the message body as standard input.
*/
-@UriEndpoint(firstVersion = "4.4.0", scheme = Wasm.SCHEME, title = "Wasm", syntax = "wasm:functionName",
+@UriEndpoint(firstVersion = "4.4.0", scheme = Wasm.SCHEME, title = "Wasm", syntax = "wasm:module",
producerOnly = true, remote = false, category = { Category.CORE, Category.SCRIPT },
headersClass = Wasm.Headers.class)
public class WasmEndpoint extends DefaultEndpoint {
@Metadata(required = true)
- @UriPath(description = "The Function Name")
- private final String functionName;
+ @UriPath(description = "The WASI Preview 1 command module to run, as a resource: by default from the classpath, or"
+ + " with a prefix such as classpath:, file: or http:")
+ private final String module;
@UriParam
private WasmConfiguration configuration;
- public WasmEndpoint(String endpointUri, Component component, String functionName,
+ public WasmEndpoint(String endpointUri, Component component, String module,
WasmConfiguration configuration) {
super(endpointUri, component);
- this.functionName = functionName;
+ this.module = module;
this.configuration = configuration;
}
@@ -54,13 +55,17 @@ public class WasmEndpoint extends DefaultEndpoint {
return false;
}
+ public String getModule() {
+ return module;
+ }
+
public WasmConfiguration getConfiguration() {
return configuration;
}
@Override
public Producer createProducer() throws Exception {
- return new WasmProducer(this, configuration.getModule(), functionName);
+ return new WasmProducer(this);
}
@Override
diff --git a/components/camel-wasm/src/main/java/org/apache/camel/component/wasm/WasmExitCodeException.java b/components/camel-wasm/src/main/java/org/apache/camel/component/wasm/WasmExitCodeException.java
new file mode 100644
index 000000000000..bda38c1ea8bc
--- /dev/null
+++ b/components/camel-wasm/src/main/java/org/apache/camel/component/wasm/WasmExitCodeException.java
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.wasm;
+
+import org.apache.camel.CamelExchangeException;
+import org.apache.camel.Exchange;
+
+/**
+ * Thrown when a module exits with a non-zero exit code and the endpoint fails on that (failOnNonZeroExit, the default).
+ * <p/>
+ * The standard output of the module is not set as the body. The body is the object it was before the call, but the
+ * module may have read it: a body that is a stream has been read, and can only be read again when it is a
+ * {@link org.apache.camel.StreamCache} (stream caching, the default), which Camel resets. When the endpoint has a
+ * timeout, a body that was a stream but not a StreamCache is replaced by the bytes that were read from it.
+ */
+public class WasmExitCodeException extends CamelExchangeException {
+
+ private final int exitCode;
+ private final String stderr;
+
+ public WasmExitCodeException(String message, int exitCode, String stderr, Exchange exchange) {
+ super(message, exchange);
+ this.exitCode = exitCode;
+ this.stderr = stderr;
+ }
+
+ public int getExitCode() {
+ return exitCode;
+ }
+
+ /**
+ * What the module wrote to its standard error, decoded as UTF-8, or an empty string.
+ */
+ public String getStderr() {
+ return stderr;
+ }
+}
diff --git a/components/camel-wasm/src/main/java/org/apache/camel/component/wasm/WasmProducer.java b/components/camel-wasm/src/main/java/org/apache/camel/component/wasm/WasmProducer.java
index 4f37664da67b..acda97e2730c 100644
--- a/components/camel-wasm/src/main/java/org/apache/camel/component/wasm/WasmProducer.java
+++ b/components/camel-wasm/src/main/java/org/apache/camel/component/wasm/WasmProducer.java
@@ -16,76 +16,431 @@
*/
package org.apache.camel.component.wasm;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
-import org.apache.camel.Endpoint;
+import javax.xml.transform.Source;
+
+import org.apache.camel.CamelExchangeException;
import org.apache.camel.Exchange;
-import org.apache.camel.spi.Resource;
-import org.apache.camel.spi.ResourceLoader;
+import org.apache.camel.ExchangeTimedOutException;
+import org.apache.camel.Message;
+import org.apache.camel.StreamCache;
import org.apache.camel.support.DefaultProducer;
-import org.apache.camel.support.PluginHelper;
-import org.apache.camel.wasm.WasmFunction;
-import org.apache.camel.wasm.WasmSupport;
+import org.apache.camel.support.ResourceHelper;
+import org.apache.camel.util.IOHelper;
+import org.apache.camel.util.ObjectHelper;
+import org.apache.camel.util.URISupport;
+import org.apache.camel.wasm.WasiCommand;
+import org.apache.camel.wasm.Wasm;
+import run.endive.runtime.TrapException;
import run.endive.wasm.Parser;
import run.endive.wasm.WasmModule;
+/**
+ * Runs a WASI Preview 1 command module once per exchange: the message body is the standard input of the module and its
+ * standard output becomes the message body.
+ * <p/>
+ * The module is parsed once, when the producer is initialized, and every exchange runs on a new instance of it, so
+ * exchanges never share guest memory or globals and need no lock. The instance is dropped after its run, whatever the
+ * outcome.
+ * <p/>
+ * Without a timeout the module runs on the calling thread and reads the body as a stream. With a timeout the module
+ * runs on a thread of the producer, and the body is first read into memory on the calling thread: the module then reads
+ * a private copy, so a module that is abandoned after a timeout never reads a stream that the route is still using.
+ * <p/>
+ * A body that is not a stream is converted to one. A stream that the type converter opens for that (a file stream for a
+ * File, Path or GenericFile body) is closed by the producer when the module has ended, or when the body has been read
+ * into memory. A stream that is the body is never closed by the producer.
+ * <p/>
+ * When the exchange fails, the producer does not set a new body, but the module may have read the old one: a body that
+ * is a stream has been read, partly or to its end. With stream caching (the default) the body can be read again: Camel
+ * resets a {@link StreamCache}. When a timeout is set, a body that was a stream but not a StreamCache is replaced by
+ * the bytes that were read from it, so that error handling still sees the content.
+ */
public class WasmProducer extends DefaultProducer {
- private final String functionModule;
- private final String functionName;
+ private static final int MAX_STDERR_IN_MESSAGE = 256;
+ private static final String DEFAULT_PROGRAM_NAME = "module.wasm";
- private WasmModule module;
- private WasmFunction function;
+ private WasiCommand command;
+ private Map<String, String> environment = Collections.emptyMap();
+ private List<String> environmentHeaders = Collections.emptyList();
+ private volatile ExecutorService executor;
- public WasmProducer(Endpoint endpoint, String functionModule, String functionName) throws Exception {
+ public WasmProducer(WasmEndpoint endpoint) {
super(endpoint);
+ }
- this.functionModule = functionModule;
- this.functionName = functionName;
+ @Override
+ public WasmEndpoint getEndpoint() {
+ return (WasmEndpoint) super.getEndpoint();
}
@Override
- public void doInit() throws Exception {
- final ResourceLoader rl = PluginHelper.getResourceLoader(getEndpoint().getCamelContext());
- final Resource res = rl.resolveResource(this.functionModule);
+ protected void doInit() throws Exception {
+ super.doInit();
+
+ WasmEndpoint endpoint = getEndpoint();
+ WasmConfiguration conf = endpoint.getConfiguration();
+
+ Map<String, String> env = new LinkedHashMap<>();
+ if (conf.getEnvironment() != null) {
+ for (Map.Entry<String, Object> e : conf.getEnvironment().entrySet()) {
+ String value = e.getValue() != null ? e.getValue().toString() : "";
+ env.put(checkName(e.getKey()), checkNoNul("environment variable " + e.getKey(), value));
+ }
+ }
+ List<String> headers = new ArrayList<>();
+ if (ObjectHelper.isNotEmpty(conf.getEnvironmentHeaders())) {
+ for (String header : conf.getEnvironmentHeaders().split(",")) {
+ String name = checkName(header.trim());
+ if (env.containsKey(name)) {
+ throw new IllegalArgumentException(
+ "Environment variable " + name + " is configured both in environment and environmentHeaders");
+ }
+ headers.add(name);
+ }
+ }
+ List<String> arguments = arguments(programName(endpoint.getModule()), conf.getArgs());
+ for (String argument : arguments) {
+ checkNoNul("argument " + argument, argument);
+ }
- try (InputStream is = res.getInputStream()) {
- this.module = Parser.parse(is);
+ WasmModule module;
+ try (InputStream is = ResourceHelper.resolveMandatoryResourceAsInputStream(
+ endpoint.getCamelContext(), endpoint.getModule())) {
+ module = Parser.parse(is);
}
+
+ this.environment = Collections.unmodifiableMap(env);
+ this.environmentHeaders = List.copyOf(headers);
+ this.command = new WasiCommand(module, arguments, conf.getMaxMemoryPages());
}
@Override
- public void doStart() throws Exception {
+ protected void doStart() throws Exception {
super.doStart();
- if (this.module != null && this.function == null) {
- this.function = new WasmFunction(this.module, this.functionName);
+ if (getEndpoint().getConfiguration().getTimeout() > 0 && executor == null) {
+ executor = getEndpoint().getCamelContext().getExecutorServiceManager()
+ .newCachedThreadPool(this, "Wasm");
}
}
@Override
- public void doStop() throws Exception {
+ protected void doStop() throws Exception {
+ ExecutorService pool = executor;
+ executor = null;
+ if (pool != null) {
+ // interrupts the modules that are still running
+ getEndpoint().getCamelContext().getExecutorServiceManager().shutdownNow(pool);
+ }
+
super.doStop();
+ }
+
+ @Override
+ public void process(Exchange exchange) throws Exception {
+ WasmConfiguration conf = getEndpoint().getConfiguration();
+ Message message = exchange.getMessage();
+
+ // the result headers only ever describe this run: never a previous one, nor a value sent with the message
+ message.removeHeader(Wasm.Headers.EXIT_CODE);
+ message.removeHeader(Wasm.Headers.STDERR);
- if (this.function != null) {
- this.function.close();
+ Map<String, String> env = environment(message);
+ // no body is empty standard input; a body that cannot be read as a stream fails the exchange instead of
+ // silently running the module with empty input
+ Object body = message.getBody();
+ InputStream in = body != null ? message.getMandatoryBody(InputStream.class) : InputStream.nullInputStream();
+ // a stream that the type converter opened for this run (for a File, Path or GenericFile body it is a new file
+ // stream) is closed here: nobody else has a reference to it
+ InputStream opened = isOpenedByConverter(body, in) ? in : null;
+ byte[] restore = null;
+ if (conf.getTimeout() > 0 && body != null) {
+ // with a timeout the module runs on another thread, which goes on for a moment after the timeout and is
+ // not stopped by an interrupt while it is blocked in a read of the body. So the body is read here, on the
+ // calling thread, and the module gets a copy that no other thread uses and that never blocks.
+ byte[] input;
+ try {
+ input = in.readAllBytes();
+ } finally {
+ IOHelper.close(opened);
+ opened = null;
+ }
+ in = new ByteArrayInputStream(input);
+ if (body instanceof InputStream && !(body instanceof StreamCache)) {
+ // the stream of the body is used up: if the exchange fails, the body becomes what was read from it
+ restore = input;
+ }
+ }
+ InputStream stdin = in;
+ LimitedOutputStream stdout = new LimitedOutputStream("standard output", conf.getMaxOutputSize());
+ LimitedOutputStream stderr = new LimitedOutputStream("standard error", conf.getMaxOutputSize());
+
+ boolean done = false;
+ try {
+ int exitCode;
+ try {
+ exitCode = run(exchange, conf.getTimeout(), () -> command.run(stdin, stdout, stderr, env));
+ } catch (TrapException e) {
+ String err = setStderr(message, stderr);
+ throw new CamelExchangeException(
+ "The Wasm module " + name() + " trapped: " + e.getMessage() + excerpt(err),
+ exchange, e);
+ } catch (OutputLimitExceededException e) {
+ setStderr(message, stderr);
+ throw new CamelExchangeException(
+ "The Wasm module " + name() + " was stopped: " + e.getMessage(), exchange);
+ }
+
+ message.setHeader(Wasm.Headers.EXIT_CODE, exitCode);
+ String err = setStderr(message, stderr);
+ if (exitCode != 0 && conf.isFailOnNonZeroExit()) {
+ throw new WasmExitCodeException(
+ "The Wasm module " + name() + " exited with code " + exitCode + excerpt(err),
+ exitCode, err, exchange);
+ }
+ message.setBody(stdout.toByteArray());
+ done = true;
+ } finally {
+ // without a timeout the module ran on this thread and has ended, so no other thread reads the stream
+ IOHelper.close(opened);
+ if (!done && restore != null) {
+ message.setBody(restore);
+ }
}
- this.function = null;
}
- @Override
- public void doShutdown() throws Exception {
- super.doShutdown();
+ /**
+ * Whether the type converter created the stream for this run, so that the producer has to close it. A stream that
+ * is the body is left to its owner, and so is the stream of a {@link StreamCache} or of an XML {@link Source}: for
+ * these the converter may return the stream that the body holds.
+ */
+ private static boolean isOpenedByConverter(Object body, InputStream in) {
+ return body != null && in != body && !(body instanceof StreamCache) && !(body instanceof Source);
+ }
- this.function = null;
- this.module = null;
+ /**
+ * Runs the module on the calling thread, or on a thread of the producer when a timeout is set. On timeout that
+ * thread is interrupted: Endive checks the interrupt flag on every call and every branch, and its WASI sleep
+ * (poll_oneoff) ends on an interrupt, so a module stops at its next call or loop iteration. Interrupting a thread
+ * of the producer, not the calling thread, keeps the interrupt flag away from the threads of the route.
+ * <p/>
+ * The threads come from a cached pool without a maximum size. It needs none: a thread is only taken while an
+ * exchange waits for it, so there are as many threads as exchanges in this producer at the same time, plus the
+ * threads of timed out runs for the moment they need to reach their next call or branch. The standard streams of a
+ * run with a timeout are all in memory, so no call into the host can keep a thread after its exchange is gone.
+ */
+ private int run(Exchange exchange, long timeout, Callable<Integer> task) throws Exception {
+ if (timeout <= 0) {
+ return task.call();
+ }
+ ExecutorService pool = executor;
+ if (pool == null) {
+ throw new RejectedExecutionException("The wasm producer is not started");
+ }
+ Future<Integer> future = pool.submit(task);
+ try {
+ return future.get(timeout, TimeUnit.MILLISECONDS);
+ } catch (TimeoutException e) {
+ future.cancel(true);
+ throw new ExchangeTimedOutException(
+ exchange, timeout, "the Wasm module " + name() + " did not finish in time and was interrupted");
+ } catch (InterruptedException e) {
+ future.cancel(true);
+ Thread.currentThread().interrupt();
+ throw e;
+ } catch (ExecutionException e) {
+ Throwable cause = e.getCause();
+ if (cause instanceof Exception ex) {
+ throw ex;
+ }
+ if (cause instanceof Error err) {
+ throw err;
+ }
+ throw e;
+ }
}
- @Override
- public void process(Exchange exchange) throws Exception {
- byte[] in = WasmSupport.serialize(exchange);
- byte[] result = function.run(in);
+ private String name() {
+ return URISupport.sanitizeUri(getEndpoint().getModule());
+ }
+
+ private Map<String, String> environment(Message message) {
+ if (environmentHeaders.isEmpty()) {
+ return environment;
+ }
+ Map<String, String> answer = new LinkedHashMap<>(environment);
+ for (String name : environmentHeaders) {
+ String value = message.getHeader(name, String.class);
+ if (value != null) {
+ answer.put(name, checkNoNul("header " + name, value));
+ }
+ }
+ return answer;
+ }
+
+ private static String setStderr(Message message, LimitedOutputStream stderr) {
+ String answer = stderr.toString();
+ if (!answer.isEmpty()) {
+ message.setHeader(Wasm.Headers.STDERR, answer);
+ }
+ return answer;
+ }
+
+ private static String excerpt(String stderr) {
+ String s = stderr.strip();
+ if (s.isEmpty()) {
+ return "";
+ }
+ return ": " + (s.length() > MAX_STDERR_IN_MESSAGE ? s.substring(0, MAX_STDERR_IN_MESSAGE) + "..." : s);
+ }
+
+ private static String checkName(String name) {
+ if (ObjectHelper.isEmpty(name) || name.indexOf('=') >= 0 || name.indexOf('\0') >= 0) {
+ throw new IllegalArgumentException("Invalid environment variable name: '" + name + "'");
+ }
+ return name;
+ }
+
+ private static String checkNoNul(String what, String value) {
+ // WASI passes arguments and environment variables as NUL terminated strings
+ if (value.indexOf('\0') >= 0) {
+ throw new IllegalArgumentException("The " + what + " contains a NUL character, which a module cannot receive");
+ }
+ return value;
+ }
+
+ /**
+ * The program name the module receives as argument zero: the file name of the module location, without scheme,
+ * directories, user info or query. The location itself is not passed on: it can carry credentials (an http location
+ * with user info or a token in the query) and shows host paths to a module that has no file access.
+ */
+ static String programName(String location) {
+ String name = location;
+ int query = name.indexOf('?');
+ if (query >= 0) {
+ name = name.substring(0, query);
+ }
+ int scheme = name.indexOf(':');
+ if (scheme > 0) {
+ String prefix = name.substring(0, scheme);
+ if ("base64".equals(prefix) || "gzip".equals(prefix)) {
+ // the rest is the module itself, not a name
+ return DEFAULT_PROGRAM_NAME;
+ }
+ name = name.substring(scheme + 1);
+ }
+ if (name.startsWith("//")) {
+ // host, port and user info are not part of the name
+ int path = name.indexOf('/', 2);
+ name = path < 0 ? "" : name.substring(path);
+ }
+ name = name.substring(Math.max(name.lastIndexOf('/'), name.lastIndexOf('\\')) + 1);
+ return name.isEmpty() ? DEFAULT_PROGRAM_NAME : name;
+ }
+
+ /**
+ * Splits the args option into the argument list of the module, with the program name first. Arguments are separated
+ * by whitespace; double quotes group an argument that contains whitespace.
+ */
+ static List<String> arguments(String program, String args) {
+ List<String> answer = new ArrayList<>();
+ answer.add(program);
+ if (args == null) {
+ return answer;
+ }
+ StringBuilder current = null;
+ boolean quoted = false;
+ for (int i = 0; i < args.length(); i++) {
+ char c = args.charAt(i);
+ if (c == '"') {
+ quoted = !quoted;
+ if (current == null) {
+ current = new StringBuilder();
+ }
+ } else if (Character.isWhitespace(c) && !quoted) {
+ if (current != null) {
+ answer.add(current.toString());
+ current = null;
+ }
+ } else {
+ if (current == null) {
+ current = new StringBuilder();
+ }
+ current.append(c);
+ }
+ }
+ if (quoted) {
+ throw new IllegalArgumentException("Unbalanced double quote in args: " + args);
+ }
+ if (current != null) {
+ answer.add(current.toString());
+ }
+ return answer;
+ }
- WasmSupport.deserialize(result, exchange);
+ /**
+ * Thrown by a {@link LimitedOutputStream} that is full. It is not an IOException, so that it stops the module
+ * instead of being returned to the module as a failed write.
+ */
+ private static final class OutputLimitExceededException extends RuntimeException {
+ OutputLimitExceededException(String message) {
+ super(message, null, false, false);
+ }
+ }
+
+ /**
+ * Buffers what the module writes to one of its output streams, up to maxOutputSize bytes (0 means no limit).
+ */
+ private static final class LimitedOutputStream extends OutputStream {
+ private final ByteArrayOutputStream buffer = new ByteArrayOutputStream();
+ private final String name;
+ private final int max;
+
+ LimitedOutputStream(String name, int max) {
+ this.name = name;
+ this.max = max;
+ }
+
+ @Override
+ public void write(int b) {
+ write(new byte[] { (byte) b }, 0, 1);
+ }
+
+ @Override
+ public void write(byte[] b, int off, int len) {
+ if (max > 0 && len > max - buffer.size()) {
+ throw new OutputLimitExceededException(
+ "it wrote more than " + max + " bytes to its " + name + " (maxOutputSize)");
+ }
+ buffer.write(b, off, len);
+ }
+
+ byte[] toByteArray() {
+ return buffer.toByteArray();
+ }
+
+ @Override
+ public String toString() {
+ return buffer.toString(StandardCharsets.UTF_8);
+ }
}
}
diff --git a/components/camel-wasm/src/main/java/org/apache/camel/wasm/WasiCommand.java b/components/camel-wasm/src/main/java/org/apache/camel/wasm/WasiCommand.java
new file mode 100644
index 000000000000..6702d5247c81
--- /dev/null
+++ b/components/camel-wasm/src/main/java/org/apache/camel/wasm/WasiCommand.java
@@ -0,0 +1,204 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.wasm;
+
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.security.SecureRandom;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+import run.endive.runtime.ImportValues;
+import run.endive.runtime.Instance;
+import run.endive.runtime.Memory;
+import run.endive.runtime.TrapException;
+import run.endive.wasi.WasiExitException;
+import run.endive.wasi.WasiOptions;
+import run.endive.wasi.WasiPreview1;
+import run.endive.wasm.WasmModule;
+import run.endive.wasm.types.Export;
+import run.endive.wasm.types.ExportSection;
+import run.endive.wasm.types.ExternalType;
+import run.endive.wasm.types.MemoryLimits;
+import run.endive.wasm.types.MemorySection;
+
+/**
+ * Runs a WASI Preview 1 command module (a program with a {@code _start} function, such as a Rust program built for
+ * {@code wasm32-wasip1} or a Go program built with {@code GOOS=wasip1 GOARCH=wasm}) to completion.
+ * <p/>
+ * The module is parsed once by the caller and every {@link #run} creates a new instance of it, so runs never share
+ * guest memory or globals and need no lock. A run gets exactly the standard streams, arguments and environment
+ * variables it is given: no preopened directories (so no files) and no sockets. The clock is the host's, and
+ * {@code random_get} is served by a {@link SecureRandom} of the host.
+ */
+public final class WasiCommand {
+
+ private static final String START = "_start";
+ private static final String WASI_MODULE = "wasi_snapshot_preview1";
+
+ /**
+ * The source of WASI {@code random_get}, which guests use for keys, nonces, UUIDs and hash seeds (Go
+ * {@code crypto/rand}, Rust {@code getrandom}). Endive's default is {@code ThreadLocalRandom}, which is
+ * predictable. A SecureRandom is thread-safe, so one is shared by all runs.
+ */
+ private static final SecureRandom RANDOM = new SecureRandom();
+
+ private final WasmModule module;
+ private final List<String> arguments;
+ private final MemoryLimits memoryLimits;
+
+ /**
+ * @param module the parsed module
+ * @param arguments the program's arguments, the program name first
+ * @param maxMemoryPages the maximum number of 64 KiB pages the module's memory may grow to, or 0 to keep
+ * the maximum the module declares
+ * @throws IllegalArgumentException if the module is not a WASI Preview 1 command module, or maxMemoryPages is lower
+ * than the initial memory the module declares
+ */
+ public WasiCommand(WasmModule module, List<String> arguments, int maxMemoryPages) {
+ this.module = Objects.requireNonNull(module);
+ this.arguments = List.copyOf(arguments);
+ validate(module);
+ this.memoryLimits = memoryLimits(module, maxMemoryPages);
+ }
+
+ /**
+ * Runs the program once, on a new instance, on the calling thread.
+ * <p/>
+ * A {@link RuntimeException} thrown by one of the given streams stops the program and is rethrown.
+ *
+ * @param stdin the program's standard input
+ * @param stdout receives the program's standard output
+ * @param stderr receives the program's standard error
+ * @param environment the program's complete environment
+ * @return the exit code: 0 when {@code _start} returned, otherwise the
+ * value passed to {@code proc_exit}
+ * @throws TrapException if the program trapped (for example a Rust panic)
+ * @throws run.endive.runtime.WasmInterruptedException if the calling thread was interrupted
+ */
+ public int run(InputStream stdin, OutputStream stdout, OutputStream stderr, Map<String, String> environment) {
+ WasiOptions options = options(stdin, stdout, stderr, environment);
+
+ Instance instance = null;
+ try (WasiPreview1 wasi = WasiPreview1.builder().withOptions(options).build()) {
+ instance = newInstance(ImportValues.builder().addFunction(wasi.toHostFunctions()).build());
+ instance.export(START).apply();
+ return 0;
+ } catch (WasiExitException e) {
+ return e.exitCode();
+ } catch (TrapException e) {
+ String hint = memoryHint(instance);
+ if (hint == null) {
+ throw e;
+ }
+ TrapException trap = new TrapException(e.getMessage() + hint);
+ trap.initCause(e);
+ throw trap;
+ }
+ }
+
+ /**
+ * The WASI context of one run. Package-private for tests only.
+ */
+ WasiOptions options(InputStream stdin, OutputStream stdout, OutputStream stderr, Map<String, String> environment) {
+ // Endive reports the standard streams as terminals unless told otherwise; they are not, and a program that
+ // sees a terminal may add terminal output, such as colours, to the message body
+ WasiOptions.Builder options = WasiOptions.builder()
+ .withRandom(RANDOM)
+ .withStdin(stdin, false)
+ .withStdout(stdout, false)
+ .withStderr(stderr, false)
+ .withArguments(arguments)
+ .withThrowOnExit0(true);
+ environment.forEach(options::withEnvironment);
+ return options.build();
+ }
+
+ private Instance newInstance(ImportValues imports) {
+ Instance.Builder builder = Instance.builder(module)
+ .withImportValues(imports)
+ // _start is called explicitly, so that its exit code and traps are handled in one place
+ .withStart(false);
+ if (memoryLimits != null) {
+ builder.withMemoryLimits(memoryLimits);
+ }
+ return builder.build();
+ }
+
+ private static void validate(WasmModule module) {
+ ExportSection exports = module.exportSection();
+ boolean hasStart = false;
+ for (int i = 0; i < exports.exportCount(); i++) {
+ Export export = exports.getExport(i);
+ if (START.equals(export.name()) && export.exportType() == ExternalType.FUNCTION) {
+ hasStart = true;
+ break;
+ }
+ }
+ if (!hasStart) {
+ throw new IllegalArgumentException(
+ "The module has no " + START + " function, so it is not a WASI Preview 1 command module."
+ + " Build it as a program, for example with cargo build --target"
+ + " wasm32-wasip1 or GOOS=wasip1 GOARCH=wasm go build");
+ }
+ String unsupported = module.importSection().stream()
+ .filter(i -> !WASI_MODULE.equals(i.module()))
+ .map(i -> i.module() + "." + i.name())
+ .distinct()
+ .collect(Collectors.joining(", "));
+ if (!unsupported.isEmpty()) {
+ throw new IllegalArgumentException(
+ "The module imports from modules other than " + WASI_MODULE + ", which cannot be provided: "
+ + unsupported);
+ }
+ }
+
+ private static MemoryLimits memoryLimits(WasmModule module, int maxMemoryPages) {
+ if (maxMemoryPages <= 0) {
+ return null;
+ }
+ Optional<MemorySection> section = module.memorySection();
+ if (section.isEmpty() || section.get().memoryCount() == 0) {
+ // the module defines no memory of its own, so there is nothing to limit
+ return null;
+ }
+ MemoryLimits declared = section.get().getMemory(0).limits();
+ if (maxMemoryPages < declared.initialPages()) {
+ throw new IllegalArgumentException(
+ "maxMemoryPages (" + maxMemoryPages + ") is lower than the " + declared.initialPages()
+ + " pages of memory the module declares as its initial size");
+ }
+ return new MemoryLimits(declared.initialPages(), Math.min(maxMemoryPages, declared.maximumPages()));
+ }
+
+ /**
+ * A hint for the message of a trap when the memory is at its maximum, the usual reason a module traps after
+ * maxMemoryPages was lowered: its allocator could not grow the memory and aborted. Null when there is no hint.
+ */
+ private static String memoryHint(Instance instance) {
+ // the memory is null when the module has none
+ Memory memory = instance != null ? instance.memory() : null;
+ if (memory != null && memory.pages() >= memory.maximumPages()) {
+ return " (the module's memory is at its maximum of " + memory.maximumPages()
+ + " pages, see the maxMemoryPages option)";
+ }
+ return null;
+ }
+}
diff --git a/components/camel-wasm/src/main/java/org/apache/camel/wasm/Wasm.java b/components/camel-wasm/src/main/java/org/apache/camel/wasm/Wasm.java
index 82232a02179f..c93336f4889b 100644
--- a/components/camel-wasm/src/main/java/org/apache/camel/wasm/Wasm.java
+++ b/components/camel-wasm/src/main/java/org/apache/camel/wasm/Wasm.java
@@ -16,6 +16,8 @@
*/
package org.apache.camel.wasm;
+import org.apache.camel.spi.Metadata;
+
public final class Wasm {
public static final String SCHEME = "wasm";
public static final String FN_ALLOC = "alloc";
@@ -26,6 +28,16 @@ public final class Wasm {
public static class Headers {
+ @Metadata(description = "The exit code of the module: 0 when it returned from its `_start` function, otherwise the"
+ + " value it passed to `proc_exit`. Not set when the module trapped, timed out or exceeded"
+ + " maxOutputSize.",
+ javaType = "Integer")
+ public static final String EXIT_CODE = "CamelWasmExitCode";
+
+ @Metadata(description = "What the module wrote to its standard error, decoded as UTF-8. Not set when the module"
+ + " wrote nothing or timed out.",
+ javaType = "String")
+ public static final String STDERR = "CamelWasmStderr";
}
}
diff --git a/components/camel-wasm/src/main/java/org/apache/camel/wasm/WasmSupport.java b/components/camel-wasm/src/main/java/org/apache/camel/wasm/WasmSupport.java
index 343f237793de..0fe5bddcb0f9 100644
--- a/components/camel-wasm/src/main/java/org/apache/camel/wasm/WasmSupport.java
+++ b/components/camel-wasm/src/main/java/org/apache/camel/wasm/WasmSupport.java
@@ -41,20 +41,6 @@ public final class WasmSupport {
return MAPPER.writeValueAsBytes(env);
}
- @SuppressWarnings({ "unchecked", "rawtypes" })
- public static void deserialize(byte[] in, Exchange out) throws Exception {
- // cleanup
- out.getMessage().getHeaders().clear();
- out.getMessage().setBody(null);
-
- Wrapper w = MAPPER.readValue(in, Wrapper.class);
- out.getMessage().setBody(w.body);
-
- if (w.headers != null) {
- out.getMessage().setHeaders((Map) w.headers);
- }
- }
-
public static class Wrapper {
@JsonProperty
public Map<String, String> headers = new HashMap<>();
diff --git a/components/camel-wasm/src/test/java/org/apache/camel/component/wasm/WasmComponentTest.java b/components/camel-wasm/src/test/java/org/apache/camel/component/wasm/WasmComponentTest.java
index 3faf9eea62c8..3474981562cd 100644
--- a/components/camel-wasm/src/test/java/org/apache/camel/component/wasm/WasmComponentTest.java
+++ b/components/camel-wasm/src/test/java/org/apache/camel/component/wasm/WasmComponentTest.java
@@ -16,72 +16,927 @@
*/
package org.apache.camel.component.wasm;
+import java.io.ByteArrayInputStream;
+import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Locale;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+import javax.xml.transform.stream.StreamSource;
+
import org.apache.camel.CamelContext;
+import org.apache.camel.CamelExchangeException;
import org.apache.camel.Exchange;
+import org.apache.camel.ExchangeTimedOutException;
import org.apache.camel.FluentProducerTemplate;
+import org.apache.camel.InvalidPayloadException;
+import org.apache.camel.ProducerTemplate;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.support.TypeConverterSupport;
+import org.apache.camel.wasm.Wasm;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
import org.junit.jupiter.params.provider.ValueSource;
+import run.endive.runtime.TrapException;
+import run.endive.wabt.Wat2Wasm;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.awaitility.Awaitility.await;
-public class WasmComponentTest {
+/**
+ * Runs small WASI Preview 1 command modules, written in WAT in this class and compiled when the tests run.
+ */
+class WasmComponentTest {
- @ParameterizedTest
- @ValueSource(strings = {
- "functions.wasm",
- "file:{{wasm.resources.path}}/functions.wasm"
- })
- public void testComponent(String module) throws Exception {
- try (CamelContext cc = new DefaultCamelContext()) {
- FluentProducerTemplate pt = cc.createFluentProducerTemplate();
+ /**
+ * The start of every test module: the WASI functions the modules use, one page of memory, and two helpers. The
+ * helpers keep their I/O vector at address 0 and the byte count at address 8.
+ */
+ private static final String PRELUDE = """
+ (module
+ (import "wasi_snapshot_preview1" "fd_read" (func $fd_read (param i32 i32 i32 i32) (result i32)))
+ (import "wasi_snapshot_preview1" "fd_write" (func $fd_write (param i32 i32 i32 i32) (result i32)))
+ (import "wasi_snapshot_preview1" "proc_exit" (func $proc_exit (param i32)))
+ (import "wasi_snapshot_preview1" "environ_sizes_get" (func $environ_sizes_get (param i32 i32) (result i32)))
+ (import "wasi_snapshot_preview1" "environ_get" (func $environ_get (param i32 i32) (result i32)))
+ (import "wasi_snapshot_preview1" "args_sizes_get" (func $args_sizes_get (param i32 i32) (result i32)))
+ (import "wasi_snapshot_preview1" "args_get" (func $args_get (param i32 i32) (result i32)))
+ (import "wasi_snapshot_preview1" "fd_fdstat_get" (func $fd_fdstat_get (param i32 i32) (result i32)))
+ (memory (export "memory") 1)
- cc.addRoutes(new RouteBuilder() {
- @Override
- public void configure() throws Exception {
- from("direct:in")
- .toF("wasm:process?module=%s", module);
- }
- });
- cc.start();
+ (func $write (param $fd i32) (param $ptr i32) (param $len i32)
+ (i32.store (i32.const 0) (local.get $ptr))
+ (i32.store (i32.const 4) (local.get $len))
+ (drop (call $fd_write (local.get $fd) (i32.const 0) (i32.const 1) (i32.const 8))))
+
+ (func $read (param $ptr i32) (param $len i32) (result i32)
+ (i32.store (i32.const 0) (local.get $ptr))
+ (i32.store (i32.const 4) (local.get $len))
+ (drop (call $fd_read (i32.const 0) (i32.const 0) (i32.const 1) (i32.const 8)))
+ (i32.load (i32.const 8)))
+ """;
+
+ /** Copies standard input to standard output, with ASCII letters in upper case. */
+ private static final String UPPER = """
+ (func (export "_start")
+ (local $n i32) (local $i i32) (local $c i32)
+ (block $done
+ (loop $more
+ (local.set $n (call $read (i32.const 1024) (i32.const 4096)))
+ (br_if $done (i32.eqz (local.get $n)))
+ (local.set $i (i32.const 0))
+ (block $end
+ (loop $each
+ (br_if $end (i32.ge_u (local.get $i) (local.get $n)))
+ (local.set $c (i32.load8_u (i32.add (i32.const 1024) (local.get $i))))
+ (if (i32.and (i32.ge_u (local.get $c) (i32.const 97)) (i32.le_u (local.get $c) (i32.const 122)))
+ (then (i32.store8 (i32.add (i32.const 1024) (local.get $i))
+ (i32.sub (local.get $c) (i32.const 32)))))
+ (local.set $i (i32.add (local.get $i) (i32.const 1)))
+ (br $each)))
+ (call $write (i32.const 1) (i32.const 1024) (local.get $n))
+ (br $more))))
+ """;
+
+ /** Writes "out" to standard output and "boom" to standard error, then exits with the given code. */
+ private static final String EXIT = """
+ (data (i32.const 16) "boom")
+ (data (i32.const 32) "out")
+ (func (export "_start")
+ (call $write (i32.const 1) (i32.const 32) (i32.const 3))
+ (call $write (i32.const 2) (i32.const 16) (i32.const 4))
+ (call $proc_exit (i32.const %d)))
+ """;
+
+ /** Writes its environment to standard output: NAME=value entries, each followed by a NUL. */
+ private static final String ENV = """
+ (func (export "_start")
+ (drop (call $environ_sizes_get (i32.const 100) (i32.const 104)))
+ (drop (call $environ_get (i32.const 200) (i32.const 1024)))
+ (call $write (i32.const 1) (i32.const 1024) (i32.load (i32.const 104))))
+ """;
+
+ /** Writes its arguments to standard output, each followed by a NUL. */
+ private static final String ARGS = """
+ (func (export "_start")
+ (drop (call $args_sizes_get (i32.const 100) (i32.const 104)))
+ (drop (call $args_get (i32.const 200) (i32.const 1024)))
+ (call $write (i32.const 1) (i32.const 1024) (i32.load (i32.const 104))))
+ """;
+
+ /** Writes the WASI file type of its standard input, output and error as three digits. */
+ private static final String FILE_TYPES = """
+ (func $type (param $fd i32) (param $at i32)
+ (drop (call $fd_fdstat_get (local.get $fd) (i32.const 100)))
+ (i32.store8 (local.get $at) (i32.add (i32.const 48) (i32.load8_u (i32.const 100)))))
+ (func (export "_start")
+ (call $type (i32.const 0) (i32.const 1024))
+ (call $type (i32.const 1) (i32.const 1025))
+ (call $type (i32.const 2) (i32.const 1026))
+ (call $write (i32.const 1) (i32.const 1024) (i32.const 3)))
+ """;
+
+ /** Never returns. */
+ private static final String SPIN = """
+ (func (export "_start")
+ (loop $forever (br $forever)))
+ """;
+
+ /** Reads its standard input to the end, then never returns. */
+ private static final String READ_SPIN = """
+ (func (export "_start")
+ (block $done
+ (loop $more
+ (br_if $done (i32.eqz (call $read (i32.const 1024) (i32.const 4096))))
+ (br $more)))
+ (loop $forever (br $forever)))
+ """;
+
+ /** Reads its standard input to the end, then exits with code 3. */
+ private static final String READ_EXIT = """
+ (func (export "_start")
+ (block $done
+ (loop $more
+ (br_if $done (i32.eqz (call $read (i32.const 1024) (i32.const 4096))))
+ (br $more)))
+ (call $proc_exit (i32.const 3)))
+ """;
+
+ /**
+ * Sleeps for an hour in the WASI poll_oneoff call: one subscription at address 0, to the monotonic clock (type 0 at
+ * offset 8, clock 1 at offset 16) with a relative timeout in nanoseconds at offset 24.
+ */
+ private static final String SLEEP = """
+ (module
+ (import "wasi_snapshot_preview1" "poll_oneoff" (func $poll_oneoff (param i32 i32 i32 i32) (result i32)))
+ (memory (export "memory") 1)
+ (func (export "_start")
+ (i32.store8 (i32.const 8) (i32.const 0))
+ (i32.store (i32.const 16) (i32.const 1))
+ (i64.store (i32.const 24) (i64.const 3600000000000))
+ (drop (call $poll_oneoff (i32.const 0) (i32.const 64) (i32.const 1) (i32.const 128)))))
+ """;
+
+ /** Writes "panicked" to standard error, then traps, as a Rust program does on a panic. */
+ private static final String TRAP = """
+ (data (i32.const 16) "panicked")
+ (func (export "_start")
+ (call $write (i32.const 2) (i32.const 16) (i32.const 8))
+ unreachable)
+ """;
+
+ /** Grows its memory one page at a time, at most 16 times, and exits with the number of pages it ended up with. */
+ private static final String GROW = """
+ (func (export "_start")
+ (local $i i32)
+ (block $stop
+ (loop $grow
+ (br_if $stop (i32.ge_u (local.get $i) (i32.const 16)))
+ (br_if $stop (i32.eq (memory.grow (i32.const 1)) (i32.const -1)))
+ (local.set $i (i32.add (local.get $i) (i32.const 1)))
+ (br $grow)))
+ (call $proc_exit (memory.size)))
+ """;
+
+ /** Writes 10 blocks of 100 bytes to the given file descriptor. */
+ private static final String FLOOD = """
+ (func (export "_start")
+ (local $i i32)
+ (block $done
+ (loop $block
+ (br_if $done (i32.ge_u (local.get $i) (i32.const 10)))
+ (call $write (i32.const %d) (i32.const 1024) (i32.const 100))
+ (local.set $i (i32.add (local.get $i) (i32.const 1)))
+ (br $block))))
+ """;
+
+ /** Counts its runs in a global and writes the count as a digit. */
+ private static final String COUNTER = """
+ (global $count (mut i32) (i32.const 0))
+ (func (export "_start")
+ (global.set $count (i32.add (global.get $count) (i32.const 1)))
+ (i32.store8 (i32.const 1024) (i32.add (i32.const 48) (global.get $count)))
+ (call $write (i32.const 1) (i32.const 1024) (i32.const 1)))
+ """;
+
+ /** Writes 32 bytes of WASI random_get to standard output. */
+ private static final String RANDOM = """
+ (module
+ (import "wasi_snapshot_preview1" "random_get" (func $random_get (param i32 i32) (result i32)))
+ (import "wasi_snapshot_preview1" "fd_write" (func $fd_write (param i32 i32 i32 i32) (result i32)))
+ (memory (export "memory") 1)
+ (func (export "_start")
+ (drop (call $random_get (i32.const 1024) (i32.const 32)))
+ (i32.store (i32.const 0) (i32.const 1024))
+ (i32.store (i32.const 4) (i32.const 32))
+ (drop (call $fd_write (i32.const 1) (i32.const 0) (i32.const 1) (i32.const 8)))))
+ """;
+
+ @TempDir
+ static Path dir;
+
+ /**
+ * Compiles the prelude plus the given functions to a module file and returns its location for the endpoint URI.
+ */
+ private static String module(String name, String functions) throws Exception {
+ return wat(name, PRELUDE + functions + ")");
+ }
+
+ private static String wat(String name, String wat) throws Exception {
+ Path file = dir.resolve(name + ".wasm");
+ Files.write(file, Wat2Wasm.parse(wat));
+ return "file:" + file.toAbsolutePath();
+ }
+
+ private static CamelContext context(String uri) throws Exception {
+ CamelContext cc = new DefaultCamelContext();
+ cc.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:in").routeId("wasm").to(uri);
+ }
+ });
+ return cc;
+ }
- Exchange out = pt.to("direct:in")
- .withHeader("foo", "bar")
- .withBody("hello")
- .request(Exchange.class);
+ private static CamelContext start(String uri) throws Exception {
+ CamelContext cc = context(uri);
+ cc.start();
+ return cc;
+ }
+
+ /**
+ * Starts a context that hands a stream body to the endpoint as it is, instead of as a StreamCache that was already
+ * read into memory when the exchange entered the route.
+ */
+ private static CamelContext startWithoutStreamCaching(String uri) throws Exception {
+ CamelContext cc = context(uri);
+ cc.setStreamCaching(false);
+ cc.start();
+ return cc;
+ }
+
+ private static void assertStartFails(String uri, String message) throws Exception {
+ try (CamelContext cc = context(uri)) {
+ assertThatThrownBy(cc::start)
+ .rootCause()
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining(message);
+ }
+ }
+
+ private static Exchange send(CamelContext cc, Object body) {
+ return send(cc.createFluentProducerTemplate().withBody(body));
+ }
+
+ private static Exchange send(FluentProducerTemplate pt) {
+ return pt.to("direct:in").request(Exchange.class);
+ }
+
+ private static String text(Exchange exchange) {
+ return new String(exchange.getMessage().getBody(byte[].class), StandardCharsets.UTF_8);
+ }
+
+ /**
+ * Whether a thread of a wasm producer (named "... - Wasm" by Camel) is running guest code.
+ */
+ private static boolean moduleRunning() {
+ return Thread.getAllStackTraces().entrySet().stream()
+ .filter(e -> e.getKey().getName().endsWith(" - Wasm"))
+ .anyMatch(e -> Arrays.stream(e.getValue()).anyMatch(f -> f.getClassName().startsWith("run.endive.runtime")));
+ }
+
+ @Test
+ void bodyIsStandardInputAndStandardOutputIsBody() throws Exception {
+ try (CamelContext cc = start("wasm:" + module("upper", UPPER))) {
+ Exchange out = send(cc.createFluentProducerTemplate()
+ .withBody("hello wasi".getBytes(StandardCharsets.UTF_8))
+ .withHeader("foo", "bar"));
+ assertThat(out.getException()).isNull();
+ assertThat(out.getMessage().getBody()).isInstanceOf(byte[].class);
+ assertThat(text(out)).isEqualTo("HELLO WASI");
assertThat(out.getMessage().getHeaders())
- .containsEntry("foo", "bar");
- assertThat(out.getMessage().getBody(String.class))
- .isEqualTo("HELLO");
+ .containsEntry("foo", "bar")
+ .containsEntry(Wasm.Headers.EXIT_CODE, 0)
+ .doesNotContainKey(Wasm.Headers.STDERR);
+ }
+ }
+
+ @Test
+ void largeAndEmptyBodies() throws Exception {
+ try (CamelContext cc = start("wasm:" + module("upper", UPPER))) {
+ byte[] large = new byte[1024 * 1024];
+ Arrays.fill(large, (byte) 'a');
+ byte[] expected = new byte[large.length];
+ Arrays.fill(expected, (byte) 'A');
+ assertThat(send(cc, large).getMessage().getBody(byte[].class)).isEqualTo(expected);
+ assertThat(send(cc, null).getMessage().getBody(byte[].class)).isEmpty();
}
}
@Test
- public void testComponentFailure() throws Exception {
- try (CamelContext cc = new DefaultCamelContext()) {
- FluentProducerTemplate pt = cc.createFluentProducerTemplate();
+ void nonZeroExitFailsTheExchange() throws Exception {
+ try (CamelContext cc = start("wasm:" + module("exit3", EXIT.formatted(3)))) {
+ Exchange out = send(cc, "in");
+
+ assertThat(out.getException())
+ .isInstanceOfSatisfying(WasmExitCodeException.class, e -> {
+ assertThat(e.getExitCode()).isEqualTo(3);
+ assertThat(e.getStderr()).isEqualTo("boom");
+ })
+ .hasMessageContaining("exited with code 3: boom");
+ assertThat(out.getMessage().getBody(String.class)).isEqualTo("in");
+ assertThat(out.getMessage().getHeaders())
+ .containsEntry(Wasm.Headers.EXIT_CODE, 3)
+ .containsEntry(Wasm.Headers.STDERR, "boom");
+ }
+ }
+
+ @Test
+ void nonZeroExitWithFailOnNonZeroExitFalse() throws Exception {
+ try (CamelContext cc = start("wasm:" + module("exit3", EXIT.formatted(3)) + "?failOnNonZeroExit=false")) {
+ Exchange out = send(cc, "in");
+
+ assertThat(out.getException()).isNull();
+ assertThat(text(out)).isEqualTo("out");
+ assertThat(out.getMessage().getHeaders())
+ .containsEntry(Wasm.Headers.EXIT_CODE, 3)
+ .containsEntry(Wasm.Headers.STDERR, "boom");
+ }
+ }
+
+ @Test
+ void exitZeroWithStandardError() throws Exception {
+ try (CamelContext cc = start("wasm:" + module("exit0", EXIT.formatted(0)))) {
+ Exchange out = send(cc, "in");
+
+ assertThat(out.getException()).isNull();
+ assertThat(text(out)).isEqualTo("out");
+ assertThat(out.getMessage().getHeaders())
+ .containsEntry(Wasm.Headers.EXIT_CODE, 0)
+ .containsEntry(Wasm.Headers.STDERR, "boom");
+ }
+ }
+
+ @Test
+ void resultHeadersSentWithTheMessageAreReplaced() throws Exception {
+ try (CamelContext cc = start("wasm:" + module("upper", UPPER))) {
+ Exchange out = send(cc.createFluentProducerTemplate()
+ .withBody("a")
+ .withHeader(Wasm.Headers.EXIT_CODE, 99)
+ .withHeader(Wasm.Headers.STDERR, "forged"));
+
+ assertThat(out.getMessage().getHeaders())
+ .containsEntry(Wasm.Headers.EXIT_CODE, 0)
+ .doesNotContainKey(Wasm.Headers.STDERR);
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = { "", "?timeout=30000" })
+ void trapFailsTheExchangeWithStandardError(String options) throws Exception {
+ try (CamelContext cc = start("wasm:" + module("trap", TRAP) + options)) {
+ Exchange out = send(cc.createFluentProducerTemplate()
+ .withBody("in")
+ .withHeader(Wasm.Headers.EXIT_CODE, 99)
+ .withHeader(Wasm.Headers.STDERR, "forged"));
+
+ assertThat(out.getException())
+ .isInstanceOf(CamelExchangeException.class)
+ .hasMessageContaining("trapped")
+ .hasMessageContaining(": panicked")
+ .hasCauseInstanceOf(TrapException.class);
+ assertThat(out.getMessage().getBody(String.class)).isEqualTo("in");
+ assertThat(out.getMessage().getHeaders())
+ .doesNotContainKey(Wasm.Headers.EXIT_CODE)
+ .containsEntry(Wasm.Headers.STDERR, "panicked");
+ }
+ }
+
+ @Test
+ void environmentOnlyFromTheRouteAndTheAllowListedHeaders() throws Exception {
+ String uri = "wasm:" + module("env", ENV)
+ + "?environment.LEVEL=debug&environmentHeaders=TENANT,REGION";
+ try (CamelContext cc = start(uri)) {
+ Exchange out = send(cc.createFluentProducerTemplate()
+ .withBody("")
+ .withHeader("TENANT", "acme")
+ .withHeader("SECRET", "s3cr3t")
+ .withHeader("CamelHttpUri", "http://example"));
+
+ // REGION is listed but not on the message, so it is not set; nothing from the JVM environment
+ assertThat(text(out)).isEqualTo("LEVEL=debug\0TENANT=acme\0");
+ }
+ }
+
+ @Test
+ void headerWithNulCannotBePassed() throws Exception {
+ try (CamelContext cc = start("wasm:" + module("env", ENV) + "?environmentHeaders=TENANT")) {
+ Exchange out = send(cc.createFluentProducerTemplate().withBody("").withHeader("TENANT", "a\0b"));
+
+ assertThat(out.getException())
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("header TENANT contains a NUL");
+ }
+ }
+
+ @Test
+ void environmentNameConfiguredTwiceIsRejected() throws Exception {
+ String uri = "wasm:" + module("env", ENV) + "?environment.TENANT=x&environmentHeaders=TENANT";
+
+ assertStartFails(uri, "TENANT is configured both in environment and environmentHeaders");
+ }
+
+ @Test
+ void argumentsFromTheRoute() throws Exception {
+ String module = module("args", ARGS);
+ try (CamelContext cc = start("wasm:" + module + "?args=--level 2 \"two words\"")) {
+ Exchange out = send(cc, "");
+
+ // argument zero is the file name of the module, not its location
+ assertThat(module).startsWith("file:").endsWith("/args.wasm");
+ assertThat(text(out)).isEqualTo("args.wasm\0--level\0" + "2\0two words\0");
+ }
+ }
+
+ @ParameterizedTest
+ @CsvSource({
+ "file:/opt/modules/upper.wasm, upper.wasm",
+ "file:C:\\modules\\upper.wasm, upper.wasm",
+ "classpath:upper.wasm, upper.wasm",
+ "classpath:org/acme/upper.wasm, upper.wasm",
+ "upper.wasm, upper.wasm",
+ "ref:upperModule, upperModule",
+ "https://user:secret@example.com/mods/upper.wasm?token=abc123, upper.wasm",
+ "http://user:secret@example.com, module.wasm",
+ "http://example.com:8080/, module.wasm",
+ "base64:AGFzbQEAAAA=, module.wasm",
+ "gzip:H4sIAAAA/abc, module.wasm" })
+ void argumentZeroIsTheFileNameOfTheModule(String location, String expected) {
+ assertThat(WasmProducer.programName(location)).isEqualTo(expected);
+ assertThat(WasmProducer.arguments(WasmProducer.programName(location), "-v")).containsExactly(expected, "-v");
+ }
+
+ @Test
+ void bodyThatIsNotAStreamFailsTheExchange() throws Exception {
+ try (CamelContext cc = start("wasm:" + module("upper", UPPER))) {
+ // a body without a converter to InputStream must not become empty standard input
+ Exchange out = send(cc, new Object());
+
+ assertThat(out.getException()).isInstanceOf(InvalidPayloadException.class);
+ assertThat(out.getMessage().getHeaders()).doesNotContainKey(Wasm.Headers.EXIT_CODE);
+ }
+ }
+
+ @Test
+ void randomGetReturnsBytesOfTheHost() throws Exception {
+ try (CamelContext cc = start("wasm:" + wat("random", RANDOM))) {
+ byte[] first = send(cc, "").getMessage().getBody(byte[].class);
+ byte[] second = send(cc, "").getMessage().getBody(byte[].class);
+
+ assertThat(first).hasSize(32).isNotEqualTo(new byte[32]);
+ assertThat(second).hasSize(32).isNotEqualTo(first);
+ }
+ }
+
+ @Test
+ void unbalancedQuoteInArgumentsIsRejected() throws Exception {
+ String uri = "wasm:" + module("args", ARGS) + "?args=\"open";
+
+ assertStartFails(uri, "Unbalanced double quote");
+ }
+
+ @Test
+ void standardStreamsAreNotTerminals() throws Exception {
+ try (CamelContext cc = start("wasm:" + module("filetypes", FILE_TYPES))) {
+ // 0 is the WASI file type "unknown", which Endive reports for a stream that is not a terminal; a terminal
+ // is 2 (character device)
+ assertThat(text(send(cc, ""))).isEqualTo("000");
+ }
+ }
+
+ @Test
+ void timeoutInterruptsASpinningModule() throws Exception {
+ try (CamelContext cc = start("wasm:" + module("spin", SPIN) + "?timeout=500")) {
+ for (int i = 0; i < 2; i++) {
+ Exchange out = send(cc, "in");
+
+ assertThat(out.getException())
+ .isInstanceOf(ExchangeTimedOutException.class)
+ .hasMessageContaining("was interrupted");
+ assertThat(out.getMessage().getHeaders())
+ .doesNotContainKey(Wasm.Headers.EXIT_CODE)
+ .doesNotContainKey(Wasm.Headers.STDERR);
+ }
+
+ // the interrupted modules stopped: no thread is left running guest code
+ await().atMost(10, TimeUnit.SECONDS).until(() -> !moduleRunning());
+ }
+ }
+
+ @Test
+ void timeoutInterruptsAModuleSleepingInAHostCall() throws Exception {
+ try (CamelContext cc = start("wasm:" + wat("sleep", SLEEP) + "?timeout=500")) {
+ Exchange out = send(cc, "in");
+
+ // a module whose sleep ended early would have exited with code 0
+ assertThat(out.getException()).isInstanceOf(ExchangeTimedOutException.class);
+
+ // the sleep ended on the interrupt: the thread of the producer is free again
+ await().atMost(10, TimeUnit.SECONDS).until(() -> !moduleRunning());
+ }
+ }
+
+ @Test
+ void timeoutReadsTheBodyOnTheCallingThread() throws Exception {
+ try (CamelContext cc = startWithoutStreamCaching("wasm:" + module("upper", UPPER) + "?timeout=30000")) {
+ ObservedStream body = new ObservedStream("hello wasi", false);
+
+ Exchange out = send(cc, body);
+
+ assertThat(out.getException()).isNull();
+ assertThat(text(out)).isEqualTo("HELLO WASI");
+ // the module got a copy: the stream of the route was never read by a thread of the producer
+ assertThat(body.readers).containsExactly(Thread.currentThread());
+ }
+ }
+
+ @Test
+ void timeoutWithABodyThatBlocksKeepsTheProducerThreadsOffTheStream() throws Exception {
+ ObservedStream body = new ObservedStream("hello wasi", true);
+ ExecutorService caller = Executors.newSingleThreadExecutor(r -> new Thread(r, "caller"));
+ try (CamelContext cc = startWithoutStreamCaching("wasm:" + module("upper", UPPER) + "?timeout=30000")) {
+ try {
+ Future<Exchange> result = caller.submit(() -> send(cc, body));
+
+ // a thread is blocked in a read of the body, which an interrupt does not end: it is the caller, and
+ // not a thread of the producer that a timeout would leave behind on a stream the route uses again
+ assertThat(body.reading.await(10, TimeUnit.SECONDS)).isTrue();
+ assertThat(body.readers).extracting(Thread::getName).containsExactly("caller");
+
+ body.release.countDown();
+ Exchange out = result.get(30, TimeUnit.SECONDS);
+
+ assertThat(out.getException()).isNull();
+ assertThat(text(out)).isEqualTo("HELLO WASI");
+ assertThat(body.readers).extracting(Thread::getName).containsExactly("caller");
+ } finally {
+ // never leave a reader blocked, and release it before the context stops
+ body.release.countDown();
+ caller.shutdownNow();
+ }
+ }
+ }
+
+ @Test
+ void timedOutModuleLeavesNoThreadReadingTheBody() throws Exception {
+ try (CamelContext cc = startWithoutStreamCaching("wasm:" + module("readspin", READ_SPIN) + "?timeout=500")) {
+ ObservedStream body = new ObservedStream("hello wasi", false);
+
+ Exchange out = send(cc, body);
+
+ assertThat(out.getException()).isInstanceOf(ExchangeTimedOutException.class);
+ int reads = body.reads.get();
+ assertThat(body.readers).containsExactly(Thread.currentThread());
+ assertThat(body.inRead).hasValue(0);
+ // the body was a stream that is used up: it is replaced by what was read from it
+ assertThat(out.getMessage().getBody()).isEqualTo("hello wasi".getBytes(StandardCharsets.UTF_8));
+
+ // and the module that was left behind ends without touching the stream
+ await().atMost(10, TimeUnit.SECONDS).until(() -> !moduleRunning());
+ assertThat(body.reads).hasValue(reads);
+ }
+ }
+
+ @Test
+ void streamBodyAfterAFailedRunWithoutStreamCaching() throws Exception {
+ String module = module("readexit", READ_EXIT);
+ try (CamelContext cc = startWithoutStreamCaching("wasm:" + module)) {
+ InputStream body = new ByteArrayInputStream("in".getBytes(StandardCharsets.UTF_8));
+
+ Exchange out = send(cc, body);
+
+ // without a timeout the module read the stream itself: the body is the same stream, at its end
+ assertThat(out.getException()).isInstanceOf(WasmExitCodeException.class);
+ assertThat(out.getMessage().getBody()).isSameAs(body);
+ assertThat(body.read()).isEqualTo(-1);
+ }
+ try (CamelContext cc = startWithoutStreamCaching("wasm:" + module + "?timeout=30000")) {
+ Exchange out = send(cc, new ByteArrayInputStream("in".getBytes(StandardCharsets.UTF_8)));
+
+ // with a timeout the body was read into memory first, and that copy is the body after the failure
+ assertThat(out.getException()).isInstanceOf(WasmExitCodeException.class);
+ assertThat(out.getMessage().getBody()).isEqualTo("in".getBytes(StandardCharsets.UTF_8));
+ }
+ }
+ @ParameterizedTest
+ @ValueSource(strings = { "", "?timeout=30000" })
+ void streamBodyAfterAFailedRunWithStreamCaching(String options) throws Exception {
+ String uri = "wasm:" + module("readexit", READ_EXIT) + options;
+ AtomicReference<String> seen = new AtomicReference<>();
+ try (CamelContext cc = new DefaultCamelContext()) {
cc.addRoutes(new RouteBuilder() {
@Override
- public void configure() throws Exception {
- from("direct:in")
- .to("wasm:process_err?module=functions.wasm");
+ public void configure() {
+ onException(WasmExitCodeException.class).handled(true)
+ .process(e -> seen.set(e.getMessage().getBody(String.class)));
+ from("direct:in").to(uri);
}
});
cc.start();
- Exchange out = pt.to("direct:in")
- .withHeader("foo", "bar")
- .withBody("hello")
- .request(Exchange.class);
+ send(cc, new ByteArrayInputStream("in".getBytes(StandardCharsets.UTF_8)));
+
+ // stream caching is on by default: the module read the cached stream, and Camel reset it for the handler
+ assertThat(seen).hasValue("in");
+ }
+ }
+
+ @ParameterizedTest
+ @CsvSource({ "upper,''", "upper,?timeout=30000", "readexit,''", "readexit,?timeout=30000" })
+ void streamOpenedByTheTypeConverterIsClosed(String name, String options) throws Exception {
+ boolean fails = "readexit".equals(name);
+ try (CamelContext cc = context("wasm:" + module(name, fails ? READ_EXIT : UPPER) + options)) {
+ cc.getTypeConverterRegistry().addTypeConverter(InputStream.class, Content.class, new ContentConverter());
+ cc.start();
+ Content body = new Content("hello wasi");
+
+ Exchange out = send(cc, body);
+
+ if (fails) {
+ assertThat(out.getException()).isInstanceOf(WasmExitCodeException.class);
+ assertThat(out.getMessage().getBody()).isSameAs(body);
+ } else {
+ assertThat(out.getException()).isNull();
+ assertThat(text(out)).isEqualTo("HELLO WASI");
+ }
+ // like the converters for File, Path and GenericFile, the converter opened a stream for this run: nobody but
+ // the producer can close it
+ assertThat(body.opened).singleElement().satisfies(stream -> assertThat(stream.closed).isTrue());
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = { "", "?timeout=30000" })
+ void fileBodyIsStandardInputAndCanBeDeletedAfterwards(String options) throws Exception {
+ try (CamelContext cc = start("wasm:" + module("upper", UPPER) + options)) {
+ Path file = Files.writeString(dir.resolve("input.txt"), "hello wasi");
+
+ Exchange out = send(cc, file.toFile());
+
+ assertThat(out.getException()).isNull();
+ assertThat(text(out)).isEqualTo("HELLO WASI");
+ // an open handle would keep the file in place on Windows
+ Files.delete(file);
+ assertThat(file).doesNotExist();
+ }
+ }
+
+ @ParameterizedTest
+ @CsvSource({ "upper,''", "upper,?timeout=30000", "readexit,''", "readexit,?timeout=30000" })
+ void streamOfTheCallerIsNotClosed(String name, String options) throws Exception {
+ boolean fails = "readexit".equals(name);
+ String uri = "wasm:" + module(name, fails ? READ_EXIT : UPPER) + options;
+ try (CamelContext cc = startWithoutStreamCaching(uri)) {
+ // the body is the stream
+ TrackedStream body = new TrackedStream("hello wasi");
+
+ Exchange out = send(cc, body);
+
+ assertThat(out.getException() != null).isEqualTo(fails);
+ assertThat(body.available()).isZero();
+ assertThat(body.closed).isFalse();
+ }
+ try (CamelContext cc = startWithoutStreamCaching(uri)) {
+ // the body holds the stream, and the type converter returns that stream
+ TrackedStream stream = new TrackedStream("hello wasi");
+
+ Exchange out = send(cc, new StreamSource(stream));
+
+ assertThat(out.getException() != null).isEqualTo(fails);
+ assertThat(stream.available()).isZero();
+ assertThat(stream.closed).isFalse();
+ }
+ }
+
+ @Test
+ void timeoutAcrossARouteRestart() throws Exception {
+ try (CamelContext cc = start("wasm:" + module("upper", UPPER) + "?timeout=30000")) {
+ assertThat(text(send(cc, "before"))).isEqualTo("BEFORE");
+
+ cc.getRouteController().stopRoute("wasm");
+ cc.getRouteController().startRoute("wasm");
+
+ assertThat(text(send(cc, "after"))).isEqualTo("AFTER");
+ }
+ }
+
+ @Test
+ void maxMemoryPagesLimitsMemoryGrowth() throws Exception {
+ String module = module("grow", GROW);
+ try (CamelContext cc = start("wasm:" + module + "?failOnNonZeroExit=false&maxMemoryPages=4")) {
+ assertThat(send(cc, "").getMessage().getHeader(Wasm.Headers.EXIT_CODE)).isEqualTo(4);
+ }
+ try (CamelContext cc = start("wasm:" + module + "?failOnNonZeroExit=false")) {
+ assertThat(send(cc, "").getMessage().getHeader(Wasm.Headers.EXIT_CODE)).isEqualTo(17);
+ }
+ }
+
+ @Test
+ void maxMemoryPagesBelowTheInitialMemoryIsRejected() throws Exception {
+ String uri = "wasm:" + wat("big", "(module (memory 2) (func (export \"_start\")))") + "?maxMemoryPages=1";
+
+ assertStartFails(uri, "maxMemoryPages (1) is lower than the 2 pages");
+ }
+
+ @Test
+ void maxOutputSizeLimitsStandardOutput() throws Exception {
+ String module = module("flood1", FLOOD.formatted(1));
+ try (CamelContext cc = start("wasm:" + module + "?maxOutputSize=1000")) {
+ assertThat(send(cc, "").getMessage().getBody(byte[].class)).hasSize(1000);
+ }
+ try (CamelContext cc = start("wasm:" + module + "?maxOutputSize=999")) {
+ Exchange out = send(cc, "in");
+
+ assertThat(out.getException())
+ .isInstanceOf(CamelExchangeException.class)
+ .hasMessageContaining("wrote more than 999 bytes to its standard output (maxOutputSize)");
+ assertThat(out.getMessage().getBody(String.class)).isEqualTo("in");
+ assertThat(out.getMessage().getHeaders()).doesNotContainKey(Wasm.Headers.EXIT_CODE);
+ }
+ }
+
+ @Test
+ void maxOutputSizeLimitsStandardError() throws Exception {
+ try (CamelContext cc = start("wasm:" + module("flood2", FLOOD.formatted(2)) + "?maxOutputSize=999")) {
+ Exchange out = send(cc, "in");
assertThat(out.getException())
- .isNotNull()
- .hasMessage("this is an error");
+ .isInstanceOf(CamelExchangeException.class)
+ .hasMessageContaining("wrote more than 999 bytes to its standard error (maxOutputSize)");
+ assertThat(out.getMessage().getHeader(Wasm.Headers.STDERR, String.class)).hasSize(900);
+ }
+ }
+
+ @Test
+ void everyExchangeRunsOnANewInstance() throws Exception {
+ try (CamelContext cc = start("wasm:" + module("counter", COUNTER))) {
+ for (int i = 0; i < 3; i++) {
+ assertThat(text(send(cc, ""))).isEqualTo("1");
+ }
+ }
+ }
+
+ @Test
+ void parallelExchanges() throws Exception {
+ try (CamelContext cc = start("wasm:" + module("upper", UPPER))) {
+ ProducerTemplate pt = cc.createProducerTemplate();
+ ExecutorService pool = Executors.newFixedThreadPool(8);
+ try {
+ List<Future<Integer>> results = new ArrayList<>();
+ for (int t = 0; t < 8; t++) {
+ final int thread = t;
+ results.add(pool.submit(() -> {
+ int ok = 0;
+ for (int i = 0; i < 50; i++) {
+ String in = ("thread " + thread + " message " + i + " ").repeat(1 + (i % 7));
+ byte[] out = pt.requestBody("direct:in", in, byte[].class);
+ if (new String(out, StandardCharsets.UTF_8).equals(in.toUpperCase(Locale.ROOT))) {
+ ok++;
+ }
+ }
+ return ok;
+ }));
+ }
+ for (Future<Integer> result : results) {
+ assertThat(result.get(60, TimeUnit.SECONDS)).isEqualTo(50);
+ }
+ } finally {
+ pool.shutdownNow();
+ }
+ }
+ }
+
+ @Test
+ void moduleWithoutStartIsRejected() throws Exception {
+ // the test module of the wasm language exports functions, not a program
+ assertStartFails("wasm:classpath:functions.wasm", "has no _start function");
+ }
+
+ @Test
+ void moduleWithOtherImportsIsRejected() throws Exception {
+ String uri = "wasm:" + wat("other", "(module (import \"env\" \"f\" (func)) (func (export \"_start\")))");
+
+ assertStartFails(uri, "cannot be provided: env.f");
+ }
+
+ /** A body that is not a stream. Every conversion to a stream opens a new one, as for a file. */
+ private static final class Content {
+ final List<TrackedStream> opened = new CopyOnWriteArrayList<>();
+ private final String content;
+
+ Content(String content) {
+ this.content = content;
+ }
+ }
+
+ private static final class ContentConverter extends TypeConverterSupport {
+ @Override
+ public <T> T convertTo(Class<T> type, Exchange exchange, Object value) {
+ Content content = (Content) value;
+ TrackedStream answer = new TrackedStream(content.content);
+ content.opened.add(answer);
+ return type.cast(answer);
+ }
+ }
+
+ /** A stream that records that it was closed. */
+ private static final class TrackedStream extends ByteArrayInputStream {
+ volatile boolean closed;
+
+ TrackedStream(String content) {
+ super(content.getBytes(StandardCharsets.UTF_8));
+ }
+
+ @Override
+ public void close() {
+ closed = true;
+ }
+ }
+
+ /**
+ * A body stream that records the threads that read it. When blocking, a read waits until the stream is released
+ * (for 30 seconds at most) and, like a read from a socket, is not ended by an interrupt.
+ */
+ private static final class ObservedStream extends InputStream {
+ final Set<Thread> readers = ConcurrentHashMap.newKeySet();
+ final AtomicInteger reads = new AtomicInteger();
+ final AtomicInteger inRead = new AtomicInteger();
+ final CountDownLatch reading = new CountDownLatch(1);
+ final CountDownLatch release;
+ private final ByteArrayInputStream data;
+
+ ObservedStream(String content, boolean blocking) {
+ this.data = new ByteArrayInputStream(content.getBytes(StandardCharsets.UTF_8));
+ this.release = new CountDownLatch(blocking ? 1 : 0);
+ }
+
+ @Override
+ public int read() {
+ byte[] one = new byte[1];
+ return read(one, 0, 1) < 0 ? -1 : one[0] & 0xFF;
+ }
+
+ @Override
+ public int read(byte[] b, int off, int len) {
+ readers.add(Thread.currentThread());
+ reads.incrementAndGet();
+ inRead.incrementAndGet();
+ try {
+ reading.countDown();
+ awaitRelease();
+ return data.read(b, off, len);
+ } finally {
+ inRead.decrementAndGet();
+ }
+ }
+
+ private void awaitRelease() {
+ long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(30);
+ boolean interrupted = false;
+ while (true) {
+ try {
+ // returns at once when the stream is released or the deadline has passed
+ release.await(deadline - System.nanoTime(), TimeUnit.NANOSECONDS);
+ break;
+ } catch (InterruptedException e) {
+ interrupted = true;
+ }
+ }
+ if (interrupted) {
+ Thread.currentThread().interrupt();
+ }
}
}
}
diff --git a/components/camel-wasm/src/test/java/org/apache/camel/wasm/WasiCommandTest.java b/components/camel-wasm/src/test/java/org/apache/camel/wasm/WasiCommandTest.java
new file mode 100644
index 000000000000..178610edaa39
--- /dev/null
+++ b/components/camel-wasm/src/test/java/org/apache/camel/wasm/WasiCommandTest.java
@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.wasm;
+
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.security.SecureRandom;
+import java.util.List;
+import java.util.Map;
+
+import org.junit.jupiter.api.Test;
+import run.endive.runtime.TrapException;
+import run.endive.wabt.Wat2Wasm;
+import run.endive.wasi.WasiOptions;
+import run.endive.wasm.Parser;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+class WasiCommandTest {
+
+ @Test
+ void randomGetIsServedByASecureRandom() {
+ WasiCommand command = new WasiCommand(
+ Parser.parse(Wat2Wasm.parse("(module (func (export \"_start\")))")), List.of("noop.wasm"), 0);
+
+ WasiOptions options = command.options(
+ InputStream.nullInputStream(), OutputStream.nullOutputStream(), OutputStream.nullOutputStream(), Map.of());
+
+ // Endive's default is ThreadLocalRandom, which is predictable; guests use random_get for keys and nonces
+ assertThat(options.random()).isInstanceOf(SecureRandom.class);
+ assertThat(options.arguments()).containsExactly("noop.wasm");
+ }
+
+ @Test
+ void trapOfAModuleWithoutMemory() {
+ WasiCommand command = new WasiCommand(
+ Parser.parse(Wat2Wasm.parse("(module (func (export \"_start\") unreachable))")), List.of("trap.wasm"), 0);
+
+ // the hint about a memory at its maximum does not apply, and looking for it must not fail
+ assertThatThrownBy(() -> command.run(
+ InputStream.nullInputStream(), OutputStream.nullOutputStream(), OutputStream.nullOutputStream(), Map.of()))
+ .isInstanceOf(TrapException.class)
+ .hasMessageNotContaining("maxMemoryPages");
+ }
+}
diff --git a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index 60ebcf049b21..75b4abd55eff 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -74,3 +74,25 @@ The `route-parser` plugin (`camel route-parser`) and its `camel-jbang-plugin-rou
`camel cmd route-structure MyRoute.java` instead: it prints the structure of the routes of a Java source, read
without compiling it, and of YAML and XML sources too. If you installed the plugin, remove it with
`camel plugin delete route-parser`.
+
+=== camel-wasm - the endpoint runs WASI command modules (Breaking change)
+
+The `wasm` endpoint now runs WASI Preview 1 command modules (the body is standard input, standard output becomes the
+body, the exit code and standard error are in `CamelWasmExitCode` and `CamelWasmStderr`) and no longer supports the
+custom `alloc`/`dealloc` and JSON message ABI. The URI path is now the module (`wasm:classpath:upper.wasm`); the
+function name and the `module` option are removed, and a header reaches the module only when listed in
+`environmentHeaders`. To migrate, turn the function into a program that reads standard input and writes standard
+output, built for `wasm32-wasip1` (Rust) or with `GOOS=wasip1 GOARCH=wasm` (Go). The `wasm` language is unchanged.
+
+The endpoint runs every exchange on a new instance of the module, so a module can no longer keep state from one
+message to the next. The 4.23 note that a failed call discards the Wasm instance therefore now only applies to the
+`wasm` expression.
+
+The following API of `camel-wasm` is removed or changed with the old ABI:
+
+* `WasmComponent.getModule()` and `setModule(String)`, and `WasmConfiguration.getModule()` and `setModule(String)`:
+the module is the path of the endpoint URI, available as `WasmEndpoint.getModule()`.
+* The `functionName` path parameter of the endpoint: the third argument of the `WasmEndpoint` constructor is now the
+module.
+* `WasmSupport.deserialize(byte[], Exchange)`.
+* The constructor `WasmProducer(Endpoint, String, String)` is replaced by `WasmProducer(WasmEndpoint)`.
diff --git a/dsl/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/ComponentsBuilderFactory.java b/dsl/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/ComponentsBuilderFactory.java
index c6cc2d7c4830..d938f3faa0b4 100644
--- a/dsl/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/ComponentsBuilderFactory.java
+++ b/dsl/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/ComponentsBuilderFactory.java
@@ -5275,7 +5275,8 @@ public interface ComponentsBuilderFactory {
}
/**
* Wasm (camel-wasm)
- * Invoke Wasm functions.
+ * Run Wasm programs (WASI Preview 1 command modules) with the message body
+ * as standard input.
*
* Category: core,script
* Since: 4.4
diff --git a/dsl/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/WasmComponentBuilderFactory.java b/dsl/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/WasmComponentBuilderFactory.java
index 08c0d0057e0b..1281597f692a 100644
--- a/dsl/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/WasmComponentBuilderFactory.java
+++ b/dsl/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/WasmComponentBuilderFactory.java
@@ -24,7 +24,8 @@ import org.apache.camel.builder.component.ComponentBuilder;
import org.apache.camel.component.wasm.WasmComponent;
/**
- * Invoke Wasm functions.
+ * Run Wasm programs (WASI Preview 1 command modules) with the message body as
+ * standard input.
*
* Generated by camel build tools - do NOT edit this file!
*/
@@ -33,7 +34,8 @@ public interface WasmComponentBuilderFactory {
/**
* Wasm (camel-wasm)
- * Invoke Wasm functions.
+ * Run Wasm programs (WASI Preview 1 command modules) with the message body
+ * as standard input.
*
* Category: core,script
* Since: 4.4
diff --git a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/EndpointHeaderBuilders.java b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/EndpointHeaderBuilders.java
index 7e39461b9f4f..1829e977cd1a 100644
--- a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/EndpointHeaderBuilders.java
+++ b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/EndpointHeaderBuilders.java
@@ -4056,6 +4056,20 @@ public class EndpointHeaderBuilders {
public static VertxWebsocketEndpointBuilderFactory.VertxWebsocketHeaderNameBuilder vertxWebsocket() {
return VertxWebsocketEndpointBuilderFactory.VertxWebsocketHeaderNameBuilder.INSTANCE;
}
+ /**
+ * Wasm (camel-wasm)
+ * Run Wasm programs (WASI Preview 1 command modules) with the message body
+ * as standard input.
+ *
+ * Category: core,script
+ * Since: 4.4
+ * Maven coordinates: org.apache.camel:camel-wasm
+ *
+ * @return the dsl builder for the headers' name.
+ */
+ public static WasmEndpointBuilderFactory.WasmHeaderNameBuilder wasm() {
+ return WasmEndpointBuilderFactory.WasmHeaderNameBuilder.INSTANCE;
+ }
/**
* Weather (camel-weather)
* Poll the weather information from Open Weather Map.
diff --git a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/StaticEndpointBuilders.java b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/StaticEndpointBuilders.java
index d1221528d7e9..a39ba3501a4d 100644
--- a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/StaticEndpointBuilders.java
+++ b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/StaticEndpointBuilders.java
@@ -17695,18 +17695,20 @@ public class StaticEndpointBuilders {
}
/**
* Wasm (camel-wasm)
- * Invoke Wasm functions.
+ * Run Wasm programs (WASI Preview 1 command modules) with the message body
+ * as standard input.
*
* Category: core,script
* Since: 4.4
* Maven coordinates: org.apache.camel:camel-wasm
*
- * Syntax: <code>wasm:functionName</code>
+ * Syntax: <code>wasm:module</code>
*
- * Path parameter: functionName (required)
- * The Function Name
+ * Path parameter: module (required)
+ * The WASI Preview 1 command module to run, as a resource: by default from
+ * the classpath, or with a prefix such as classpath:, file: or http:
*
- * @param path functionName
+ * @param path module
* @return the dsl builder
*/
public static WasmEndpointBuilderFactory.WasmEndpointBuilder wasm(String path) {
@@ -17714,20 +17716,22 @@ public class StaticEndpointBuilders {
}
/**
* Wasm (camel-wasm)
- * Invoke Wasm functions.
+ * Run Wasm programs (WASI Preview 1 command modules) with the message body
+ * as standard input.
*
* Category: core,script
* Since: 4.4
* Maven coordinates: org.apache.camel:camel-wasm
*
- * Syntax: <code>wasm:functionName</code>
+ * Syntax: <code>wasm:module</code>
*
- * Path parameter: functionName (required)
- * The Function Name
+ * Path parameter: module (required)
+ * The WASI Preview 1 command module to run, as a resource: by default from
+ * the classpath, or with a prefix such as classpath:, file: or http:
*
* @param componentName to use a custom component name for the endpoint
* instead of the default name
- * @param path functionName
+ * @param path module
* @return the dsl builder
*/
public static WasmEndpointBuilderFactory.WasmEndpointBuilder wasm(String componentName, String path) {
diff --git a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/WasmEndpointBuilderFactory.java b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/WasmEndpointBuilderFactory.java
index 14de613a2fd1..3ead57c9daca 100644
--- a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/WasmEndpointBuilderFactory.java
+++ b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/WasmEndpointBuilderFactory.java
@@ -27,7 +27,8 @@ import org.apache.camel.builder.EndpointProducerBuilder;
import org.apache.camel.builder.endpoint.AbstractEndpointBuilder;
/**
- * Invoke Wasm functions.
+ * Run Wasm programs (WASI Preview 1 command modules) with the message body as
+ * standard input.
*
* Generated by camel build tools - do NOT edit this file!
*/
@@ -45,19 +46,246 @@ public interface WasmEndpointBuilderFactory {
}
/**
- * Set the module (the distributable, loadable, and executable unit of
- * code in WebAssembly) resource that provides the producer function.
+ * The command line arguments passed to the module, separated by
+ * whitespace. Use double quotes around an argument that contains
+ * whitespace. The module receives the file name of the module (for
+ * example upper.wasm, without scheme, directories or query) as argument
+ * zero, followed by these arguments. The arguments are fixed by the
+ * route: they cannot be set from message headers.
*
* The option is a: <code>java.lang.String</code> type.
*
- * Required: true
* Group: producer
*
- * @param module the value to set
+ * @param args the value to set
* @return the dsl builder
*/
- default WasmEndpointBuilder module(String module) {
- doSetProperty("module", module);
+ default WasmEndpointBuilder args(String args) {
+ doSetProperty("args", args);
+ return this;
+ }
+ /**
+ * Environment variables for the module, for example
+ * environment.LOG_LEVEL=debug. The module sees no environment variables
+ * other than these and the headers listed in environmentHeaders; the
+ * environment of the JVM is never passed on. This is a multi-value
+ * option with prefix: environment.
+ *
+ * The option is a: <code>java.util.Map<java.lang.String,
+ * java.lang.Object></code> type.
+ * The option is multivalued, and you can use the environment(String,
+ * Object) method to add a value (call the method multiple times to set
+ * more values).
+ *
+ * Group: producer
+ *
+ * @param key the option key
+ * @param value the option value
+ * @return the dsl builder
+ */
+ default WasmEndpointBuilder environment(String key, Object value) {
+ doSetMultiValueProperty("environment", "environment." + key, value);
+ return this;
+ }
+ /**
+ * Environment variables for the module, for example
+ * environment.LOG_LEVEL=debug. The module sees no environment variables
+ * other than these and the headers listed in environmentHeaders; the
+ * environment of the JVM is never passed on. This is a multi-value
+ * option with prefix: environment.
+ *
+ * The option is a: <code>java.util.Map<java.lang.String,
+ * java.lang.Object></code> type.
+ * The option is multivalued, and you can use the environment(String,
+ * Object) method to add a value (call the method multiple times to set
+ * more values).
+ *
+ * Group: producer
+ *
+ * @param values the values
+ * @return the dsl builder
+ */
+ default WasmEndpointBuilder environment(Map values) {
+ doSetMultiValueProperties("environment", "environment.", values);
+ return this;
+ }
+ /**
+ * A comma separated allow-list of message headers whose values are
+ * passed to the module as environment variables of the same name.
+ * Headers usually come from whoever sent the message, so no header
+ * reaches the module unless it is listed here. A listed header that is
+ * not on the message is left unset. Camel matches header names ignoring
+ * case, so listing TENANT also passes a header named tenant, as the
+ * variable TENANT. A name may not be listed here and also be configured
+ * with environment.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Group: producer
+ *
+ * @param environmentHeaders the value to set
+ * @return the dsl builder
+ */
+ default WasmEndpointBuilder environmentHeaders(String environmentHeaders) {
+ doSetProperty("environmentHeaders", environmentHeaders);
+ return this;
+ }
+ /**
+ * Whether a module that exits with a non-zero exit code fails the
+ * exchange with a WasmExitCodeException. No new body is set then, but a
+ * body that is a stream has been read by the module: it can be read
+ * again when handling the failure if stream caching is on (the
+ * default). When false, the standard output of the module replaces the
+ * message body whatever the exit code, which is available in the
+ * CamelWasmExitCode header.
+ *
+ * The option is a: <code>boolean</code> type.
+ *
+ * Default: true
+ * Group: producer
+ *
+ * @param failOnNonZeroExit the value to set
+ * @return the dsl builder
+ */
+ default WasmEndpointBuilder failOnNonZeroExit(boolean failOnNonZeroExit) {
+ doSetProperty("failOnNonZeroExit", failOnNonZeroExit);
+ return this;
+ }
+ /**
+ * Whether a module that exits with a non-zero exit code fails the
+ * exchange with a WasmExitCodeException. No new body is set then, but a
+ * body that is a stream has been read by the module: it can be read
+ * again when handling the failure if stream caching is on (the
+ * default). When false, the standard output of the module replaces the
+ * message body whatever the exit code, which is available in the
+ * CamelWasmExitCode header.
+ *
+ * The option will be converted to a <code>boolean</code> type.
+ *
+ * Default: true
+ * Group: producer
+ *
+ * @param failOnNonZeroExit the value to set
+ * @return the dsl builder
+ */
+ default WasmEndpointBuilder failOnNonZeroExit(String failOnNonZeroExit) {
+ doSetProperty("failOnNonZeroExit", failOnNonZeroExit);
+ return this;
+ }
+ /**
+ * The maximum size of the memory of the module, in WebAssembly pages of
+ * 64 KiB. Rust and Go programs declare no maximum by default. When the
+ * limit is reached, growing the memory fails inside the module, which
+ * then usually traps or exits with an error. 0 (the default) keeps the
+ * maximum declared by the module. The value must not be lower than the
+ * initial size the module declares.
+ *
+ * The option is a: <code>int</code> type.
+ *
+ * Group: producer
+ *
+ * @param maxMemoryPages the value to set
+ * @return the dsl builder
+ */
+ default WasmEndpointBuilder maxMemoryPages(int maxMemoryPages) {
+ doSetProperty("maxMemoryPages", maxMemoryPages);
+ return this;
+ }
+ /**
+ * The maximum size of the memory of the module, in WebAssembly pages of
+ * 64 KiB. Rust and Go programs declare no maximum by default. When the
+ * limit is reached, growing the memory fails inside the module, which
+ * then usually traps or exits with an error. 0 (the default) keeps the
+ * maximum declared by the module. The value must not be lower than the
+ * initial size the module declares.
+ *
+ * The option will be converted to a <code>int</code> type.
+ *
+ * Group: producer
+ *
+ * @param maxMemoryPages the value to set
+ * @return the dsl builder
+ */
+ default WasmEndpointBuilder maxMemoryPages(String maxMemoryPages) {
+ doSetProperty("maxMemoryPages", maxMemoryPages);
+ return this;
+ }
+ /**
+ * The maximum number of bytes the module may write to its standard
+ * output, and separately to its standard error, which are buffered in
+ * memory. A module that writes more is stopped and the exchange fails.
+ * 0 (the default) means no limit: a module that keeps writing then
+ * fills the heap of the JVM, so set a limit for modules whose output
+ * size is not known.
+ *
+ * The option is a: <code>int</code> type.
+ *
+ * Group: producer
+ *
+ * @param maxOutputSize the value to set
+ * @return the dsl builder
+ */
+ default WasmEndpointBuilder maxOutputSize(int maxOutputSize) {
+ doSetProperty("maxOutputSize", maxOutputSize);
+ return this;
+ }
+ /**
+ * The maximum number of bytes the module may write to its standard
+ * output, and separately to its standard error, which are buffered in
+ * memory. A module that writes more is stopped and the exchange fails.
+ * 0 (the default) means no limit: a module that keeps writing then
+ * fills the heap of the JVM, so set a limit for modules whose output
+ * size is not known.
+ *
+ * The option will be converted to a <code>int</code> type.
+ *
+ * Group: producer
+ *
+ * @param maxOutputSize the value to set
+ * @return the dsl builder
+ */
+ default WasmEndpointBuilder maxOutputSize(String maxOutputSize) {
+ doSetProperty("maxOutputSize", maxOutputSize);
+ return this;
+ }
+ /**
+ * The maximum time the module may run for one exchange, in
+ * milliseconds. When it is exceeded the module is interrupted and the
+ * exchange fails with an ExchangeTimedOutException. When a timeout is
+ * set, the module runs on a separate thread while the calling thread
+ * waits, and the message body is read into memory before the module
+ * starts, instead of being streamed to it. The time to read the body
+ * does not count. 0 (the default) means no timeout.
+ *
+ * The option is a: <code>long</code> type.
+ *
+ * Group: producer
+ *
+ * @param timeout the value to set
+ * @return the dsl builder
+ */
+ default WasmEndpointBuilder timeout(long timeout) {
+ doSetProperty("timeout", timeout);
+ return this;
+ }
+ /**
+ * The maximum time the module may run for one exchange, in
+ * milliseconds. When it is exceeded the module is interrupted and the
+ * exchange fails with an ExchangeTimedOutException. When a timeout is
+ * set, the module runs on a separate thread while the calling thread
+ * waits, and the message body is read into memory before the module
+ * starts, instead of being streamed to it. The time to read the body
+ * does not count. 0 (the default) means no timeout.
+ *
+ * The option will be converted to a <code>long</code> type.
+ *
+ * Group: producer
+ *
+ * @param timeout the value to set
+ * @return the dsl builder
+ */
+ default WasmEndpointBuilder timeout(String timeout) {
+ doSetProperty("timeout", timeout);
return this;
}
}
@@ -123,18 +351,35 @@ public interface WasmEndpointBuilderFactory {
public interface WasmBuilders {
/**
* Wasm (camel-wasm)
- * Invoke Wasm functions.
+ * Run Wasm programs (WASI Preview 1 command modules) with the message
+ * body as standard input.
+ *
+ * Category: core,script
+ * Since: 4.4
+ * Maven coordinates: org.apache.camel:camel-wasm
+ *
+ * @return the dsl builder for the headers' name.
+ */
+ default WasmHeaderNameBuilder wasm() {
+ return WasmHeaderNameBuilder.INSTANCE;
+ }
+ /**
+ * Wasm (camel-wasm)
+ * Run Wasm programs (WASI Preview 1 command modules) with the message
+ * body as standard input.
*
* Category: core,script
* Since: 4.4
* Maven coordinates: org.apache.camel:camel-wasm
*
- * Syntax: <code>wasm:functionName</code>
+ * Syntax: <code>wasm:module</code>
*
- * Path parameter: functionName (required)
- * The Function Name
+ * Path parameter: module (required)
+ * The WASI Preview 1 command module to run, as a resource: by default
+ * from the classpath, or with a prefix such as classpath:, file: or
+ * http:
*
- * @param path functionName
+ * @param path module
* @return the dsl builder
*/
default WasmEndpointBuilder wasm(String path) {
@@ -142,20 +387,23 @@ public interface WasmEndpointBuilderFactory {
}
/**
* Wasm (camel-wasm)
- * Invoke Wasm functions.
+ * Run Wasm programs (WASI Preview 1 command modules) with the message
+ * body as standard input.
*
* Category: core,script
* Since: 4.4
* Maven coordinates: org.apache.camel:camel-wasm
*
- * Syntax: <code>wasm:functionName</code>
+ * Syntax: <code>wasm:module</code>
*
- * Path parameter: functionName (required)
- * The Function Name
+ * Path parameter: module (required)
+ * The WASI Preview 1 command module to run, as a resource: by default
+ * from the classpath, or with a prefix such as classpath:, file: or
+ * http:
*
* @param componentName to use a custom component name for the endpoint
* instead of the default name
- * @param path functionName
+ * @param path module
* @return the dsl builder
*/
default WasmEndpointBuilder wasm(String componentName, String path) {
@@ -163,6 +411,44 @@ public interface WasmEndpointBuilderFactory {
}
}
+ /**
+ * The builder of headers' name for the Wasm component.
+ */
+ public static class WasmHeaderNameBuilder {
+ /**
+ * The internal instance of the builder used to access to all the
+ * methods representing the name of headers.
+ */
+ public static final WasmHeaderNameBuilder INSTANCE = new WasmHeaderNameBuilder();
+
+ /**
+ * The exit code of the module: 0 when it returned from its _start
+ * function, otherwise the value it passed to proc_exit. Not set when
+ * the module trapped, timed out or exceeded maxOutputSize.
+ *
+ * The option is a: {@code Integer} type.
+ *
+ * Group: producer
+ *
+ * @return the name of the header {@code WasmExitCode}.
+ */
+ public String wasmExitCode() {
+ return "CamelWasmExitCode";
+ }
+ /**
+ * What the module wrote to its standard error, decoded as UTF-8. Not
+ * set when the module wrote nothing or timed out.
+ *
+ * The option is a: {@code String} type.
+ *
+ * Group: producer
+ *
+ * @return the name of the header {@code WasmStderr}.
+ */
+ public String wasmStderr() {
+ return "CamelWasmStderr";
+ }
+ }
static WasmEndpointBuilder endpointBuilder(String componentName, String path) {
class WasmEndpointBuilderImpl extends AbstractEndpointBuilder implements WasmEndpointBuilder, AdvancedWasmEndpointBuilder {
public WasmEndpointBuilderImpl(String path) {