Rust developer guide: embed Cameleer, write processors and components

Use Cameleer as a Rust library: build a CamelContext, add routes with the Rust DSL or load YAML route files, write processors, aggregation strategies, data formats and your own components, bind them so YAML routes can use them by name, and send messages with a ProducerTemplate. Everything runs on standard threads with no async runtime and no external crates.

Applies to: the Rust crates (cameleer-core and the component crates, cameleer-main) · Apache Camel: Writing components, Processor, Registry

Camel's Java extension points have Rust counterparts with the same names:

Which crates

cameleer-main has a cargo feature per component family (full by default), so a program can leave out what it does not use: see Build features. The API reference for every crate is on docs.rs (cameleer-core, cameleer-main).

Routes in the Rust DSL

use cameleer_core::prelude::*;

fn main() -> Result<()> {
    let ctx = CamelContext::new();
    ctx.add_routes(|b| {
        b.from("timer:orders?period=1000")
            .route_id("orders")
            .set_body(constant(r#"{"id": 1, "priority": "high"}"#))
            .choice()
            .when(simple("${body} contains 'high'"))
            .to("log:express")
            .otherwise()
            .to("log:standard")
            .end();
    })?;
    ctx.start()?;
    std::thread::sleep(std::time::Duration::from_secs(5));
    ctx.stop()
}

The same route in YAML runs with the CLI or the container image, with no Rust toolchain:

- route:
    id: orders
    from:
      uri: timer:orders?period=1000
      steps:
        - setBody:
            constant: '{"id": 1, "priority": "high"}'
        - choice:
            when:
              - simple: "${body} contains 'high'"
                steps:
                  - to: log:express
            otherwise:
              steps:
                - to: log:standard

cameleer run orders.camel.yaml
docker run --rm -v "$PWD:/routes:ro" pmsudhi/cameleer run /routes/orders.camel.yaml

Every EIP has a Rust method with Camel's name in snake case (split, aggregate, recipient_list, circuit_breaker, on_exception, ...); the EIP pages show each one in YAML and in Rust, and Error handling shows error_handler and on_exception. Expressions come from the prelude: simple, constant, header, body, exchange_property, jsonpath, xpath, jq, tokenize, method, and language(name, text) for any other language.

A worked example: a component, a processor and a strategy used from YAML

This program adds a greet: component, binds a processor named upper and an aggregation strategy named joinLines, adds two routes in the Rust DSL, and loads YAML route files that use the processor by name:

use std::any::Any;
use std::path::Path;
use std::sync::Arc;

use cameleer_core::prelude::*;

/// `greet:NAME?punctuation=!` replaces the body with "Hello, BODY!".
struct GreetComponent;

struct GreetEndpoint {
    uri: String,
    punctuation: String,
}

struct GreetProducer {
    punctuation: String,
}

impl Component for GreetComponent {
    fn create_endpoint(&self, _ctx: &CamelContext, uri: &str, _remaining: &str, params: &mut Parameters) -> Result<Arc<dyn Endpoint>> {
        // take the options you know; an option left over fails, as in Camel
        let punctuation = params.take_string("punctuation", ".");
        Ok(Arc::new(GreetEndpoint { uri: uri.to_string(), punctuation }))
    }
    fn as_any(&self) -> &dyn Any {
        self
    }
}

impl Endpoint for GreetEndpoint {
    fn uri(&self) -> &str {
        &self.uri
    }
    fn create_producer(&self) -> Result<Arc<dyn Producer>> {
        Ok(Arc::new(GreetProducer { punctuation: self.punctuation.clone() }))
    }
    fn as_any(&self) -> &dyn Any {
        self
    }
}

impl Service for GreetProducer {}

impl Processor for GreetProducer {
    fn process(&self, exchange: &mut Exchange) -> Result<()> {
        let greeting = format!("Hello, {}{}", exchange.body(), self.punctuation);
        exchange.set_body(greeting);
        Ok(())
    }
}

/// Every built-in component, plus `greet:`, the `upper` processor that the YAML routes in
/// `routes` refer to by name, and the `joinLines` aggregation strategy.
fn start(routes: &Path) -> Result<cameleer_main::Main> {
    let mut main = cameleer_main::new_main();
    let ctx = main.context().clone();
    ctx.add_component("greet", GreetComponent);
    ctx.registry().bind_processor("upper", |ex: &mut Exchange| -> Result<()> {
        let upper = ex.body().to_string().to_uppercase();
        ex.set_body(upper);
        Ok(())
    });
    ctx.registry().bind_aggregation_strategy("joinLines", |old: Option<Exchange>, mut new: Exchange| -> Result<Exchange> {
        if let Some(old) = old {
            let joined = format!("{}\n{}", old.body(), new.body());
            new.set_body(joined);
        }
        Ok(new)
    });
    // Rust DSL routes next to the YAML ones: split a list, send each item to the YAML route
    // `greet-one`, and join the replies with `joinLines`
    ctx.add_routes(|b| {
        b.from("direct:greetings")
            .split(simple("${body}"))
            .aggregation_strategy_ref("joinLines")
            .to("direct:greet-one")
            .end();
        b.from("direct:hello").to("greet:hello?punctuation=!");
    })?;
    main.add_route_file(routes);
    main.start()?;
    Ok(main)
}

The Rust route direct:greetings splits a list and sends each item to the YAML route, which runs upper on it and sends it through the second Rust route to greet:; joinLines joins the replies:

- route:
    id: greet-one
    from:
      uri: direct:greet-one
      steps:
        - process:
            ref: upper
        - to: direct:hello

A YAML route can name a Rust aggregation strategy too (aggregationStrategy: joinLines on a split), but cameleer validate reports a strategy that is not bound as an error, and it cannot see one bound in code; a processor reference (process: {ref: ...}) passes.

This example needs a Rust program: the cameleer binary and the container image cannot load a component, processor or strategy written in Rust. To ship routes that need them as one binary or image, cameleer export writes a Rust project around your route files (cameleer export), and you add the code there.

start(Path::new("greetings.camel.yaml")), then request_body("direct:greetings", vec!["ann", "bob"]) on a producer template returns Hello, ANN! and Hello, BOB! on two lines. greet:x?colour=red fails to resolve, because the component did not take colour. To block until Ctrl+C instead of returning, call main.run() in place of main.start(). The example runs in the test suite as written (rust_guide_worked_example in crates/cameleer-main/tests/rust_guide/mod.rs, which also checks that this page shows the same code).

Processors

A processor is anything that implements Processor; a closure |ex: &mut Exchange| -> Result<()> already does. It reads and changes the exchange in place: ex.body(), ex.set_body(...), ex.header("name"), ex.message.headers, ex.property(...) and ex.set_property(...). Returning Err fails the exchange, and the route's error handler and onException clauses take over, as when a Java processor throws:

use cameleer_core::prelude::*;

fn check_amount(ex: &mut Exchange) -> Result<()> {
    let amount: i64 = ex.body().to_string().trim().parse()
        .map_err(|e| CamelError::new(kinds::NUMBER_FORMAT, format!("not an amount: {e}")))?;
    if amount <= 0 {
        return Err(CamelError::illegal_argument("the amount must be positive"));
    }
    ex.set_header("amount", amount);
    Ok(())
}

Use it in the Rust DSL with .process(check_amount), or bind it with ctx.registry().bind_processor("checkAmount", check_amount) and refer to it from YAML with process: {ref: checkAmount}. Errors carry a kind named like Camel's Java exception classes, so an onException for NumberFormatException catches the first error and one for IllegalArgumentException the second; the kinds and their hierarchy are in Troubleshooting. A custom kind gets a parent with cameleer_core::error::register_kind_parent("OrderRejected", "IllegalArgumentException").

Beans: what YAML routes can refer to by name

YAML routes cannot create Rust values from a class name, so a program binds them in the context's registry first, and the routes refer to them by name (a leading # or #bean: is accepted):

The built-in aggregation strategies (useLatest, useOriginal, groupedBody, ...) work by name with nothing bound. A bean factory receives the context and the bean definition (its name, type_name and properties) and binds whatever it builds, which lets one Rust type be configured several ways from YAML or from application.properties. Without a factory for its type, a beans: entry fails with No bean factory registered for bean 'name'.

Components

A component is a factory of endpoints for a URI scheme. The worked example above shows the whole shape:

• Component::create_endpoint gets the full URI, the part after scheme: and the query options as Parameters. Take each option you support (take_string, take_bool, take_u64, take_duration, ...). Whatever is left fails the endpoint with Camel's "Unknown parameters" error, unless the component returns true from lenient_properties.
• Endpoint::create_producer returns a Producer, which is a Processor plus a Service (start and stop, both optional).
• Endpoint::create_consumer(processor) returns a Consumer for from:. It is a Service: start opens the connection or thread, and each received message becomes an Exchange passed to processor.process(...). Endpoint::create_polling_consumer serves pollEnrich and ConsumerTemplate.
• ctx.add_component("scheme", component) registers it; add_component_factory creates it on first use. camel.component.<scheme>.<option> properties become defaults for the options.

The spi module docs describe every method, and every component in the repository is an example: crates/cameleer-core/src/component/ has the simplest ones (direct, seda, timer, mock).

Data formats

A data format implements DataFormat (name, marshal, unmarshal, each changing the body of the exchange). bind_data_format registers one instance under a name. To take options from YAML (marshal: {avro: {schema: ...}}), register a factory instead: cameleer_core::dataformat::register_data_format_factory(ctx.registry(), "avro", |options| ...), which receives the options as a map and returns the format; a factory takes precedence over a built-in format of the same name.

TLS from Rust

Most TLS needs no code: camel.ssl.* properties and each component's SSL options are described in TLS and security. From Rust, build an SslContextParameters and either make it the global one (used by endpoints with useGlobalSslContextParameters=true) or bind it under a name for sslContextParameters=#name:

use cameleer_tls::{PemSource, SslContextParameters};

let partner = SslContextParameters {
    trust_store: vec![PemSource::File("certs/partner-ca.pem".into())],
    cert_chain: Some(PemSource::File("certs/client.pem".into())),
    private_key: Some(PemSource::File("certs/client.key".into())),
    ..Default::default()
};
ctx.registry().bind("partnerTls", partner.clone());
cameleer_tls::camel_support::set_global_ssl_context_parameters(&ctx, partner);

Configuration and Main

cameleer_main::new_main() returns a Main with every component of the build registered and the YAML and XML route loaders installed. Before start() or run():

Properties are the same keys as for the binary: see Configuration. With only cameleer-core, ctx.properties().set("key", "value") sets a placeholder value directly.

Threads, not async

Cameleer uses standard library threads and blocking I/O; it has no async runtime and depends on no external crates. Consumers run on their own threads, seda and threads use thread pools, and a ProducerTemplate call blocks until the route returns. From async code (tokio, async-std), call templates inside spawn_blocking or from a dedicated thread, and keep the CamelContext in an Arc or a static: it is Send + Sync and cheap to clone.

Faster small allocations

Routing allocates many small, short-lived values for each message. The cameleer binary installs Cameleer's thread-caching allocator, which is opt-in for a program of your own:

#[global_allocator]
static GLOBAL: cameleer_core::alloc::CamelAllocator = cameleer_core::alloc::CamelAllocator;

It keeps freed small blocks for reuse instead of returning them to the operating system; the measurements are in the benchmarks.

Testing

ProducerTemplate and mock: endpoints are how Rust tests drive and check routes: see Testing routes.

Common questions

Can I use Cameleer from an async (tokio) application?

Yes. Template calls block, so run them with tokio::task::spawn_blocking or on a thread of your own. Cameleer starts its own threads for consumers and does not need, or interfere with, the async runtime.

How do I call my Rust code from a YAML route?

Bind a processor in the registry (ctx.registry().bind_processor("name", ...)) before the routes start, and use process: {ref: name} or to: bean:name. See the worked example.

Is there a Java bean or @Handler method binding?

No. Java beans and their methods are not available; a bean is a Rust value bound in the registry, and a processor is a function. See What Cameleer does not do.

How do I write a component that receives messages?

Implement Endpoint::create_consumer: return a Service whose start begins listening (on a thread of its own), creates an Exchange for each message (Exchange::new(Some(ctx.clone())), then set_body) and calls the processor it was given. The timer and seda components in crates/cameleer-core/src/component/ are small examples.

Does Cameleer pull in dependencies?

No. Every crate uses only the Rust standard library: TLS, HTTP, the protocols, JSON, YAML and XML are written in the project.

Related

• Getting started, Configuration, Build features
• EIPs, Error handling, Testing routes
• TLS and security, Troubleshooting

Found a problem?

If something on this page doesn't match what Cameleer does, or you hit a bug while building with it, please help fix it, whether you are a person or a coding agent: open an issue that describes what you saw, then fork the repository, fix it, add or update a test, run cargo test --workspace, and open a pull request. CONTRIBUTING.md explains the ground rules.