camel-processor

Message processors for rust-camel (EIP implementations)

Overview

camel-processor provides implementations of Enterprise Integration Patterns (EIP) as processors that transform, filter, and route messages in rust-camel pipelines. These are the building blocks you use when defining routes.

Features

• Filter: Conditional message routing based on predicates
• Choice: Content-based routing with when/otherwise branches
• Log: Logging processor for debugging and monitoring
• SetHeader / DynamicSetHeader: Set message headers
• SetBody / MapBody: Transform message bodies
• Splitter: Split messages into multiple fragments
• Aggregator: Aggregate multiple messages into one
• Multicast: Send to multiple destinations in parallel
• RecipientList: Dynamically resolve endpoints from expression at runtime
• Dynamic Router: Route to endpoints determined at runtime by a closure; includes same-destination loop detection
• WireTap: Fire-and-forget message copying
• Circuit Breaker: Resilience pattern implementation
• Error Handler: Centralized error handling
• Stop: Stop processing immediately
• Script: Execute mutating expressions via ScriptMutator; changes to headers, properties, and body propagate back with atomic rollback on error
• Delayer: Delay message processing with fixed or dynamic (header-based) duration
• Stream Handling: Processors that consume streams replace the body with a JSON placeholder {"placeholder": true}
• Security Policy Layer: Tower middleware that enforces authorization before forwarding to the inner service
• Marshal / Unmarshal: Serialize/deserialize message bodies using pluggable data formats (JSON, XML)

Installation

Add to your Cargo.toml:

[dependencies]
camel-processor = "*"

Usage

Using Processors Directly

use camel_processor::{
    FilterService, LogProcessor, LogLevel, SetHeader, SetBody,
    MapBody, SplitterService, MulticastService, DelayerService
};
use camel_api::{Exchange, Message, Body, Value, BoxProcessor, DelayConfig};

// Filter exchanges
let filter = FilterService::new(
    |ex: &Exchange| ex.input.body.as_text().map(|t| t.len() > 5).unwrap_or(false),
    my_processor
);

// Log messages
let logger = LogProcessor::new(LogLevel::Info, "Processing message".to_string());

// Set a header
let set_header = SetHeader::new(identity, "source", Value::String("api".into()));

// Map body
let upper = MapBody::new(identity, |body: Body| {
    body.as_text().map(|t| Body::Text(t.to_uppercase())).unwrap_or(body)
});

// Delay with fixed duration
let delayer = DelayerService::new(DelayConfig::new(500));

// Delay with dynamic header
let dynamic_delayer = DelayerService::new(
    DelayConfig::new(1000).with_dynamic_header("CamelDelayMs")
);

With RouteBuilder (Recommended)

use camel_builder::RouteBuilder;

let route = RouteBuilder::from("timer:tick")
    .set_header("processed", Value::Bool(true))
    .filter(|ex| ex.input.body.as_text().is_some())
        .log("Processing text message", LogLevel::Info)
        .map_body(|b| Body::Text(b.as_text().unwrap_or("").to_uppercase()))
    .end_filter()
    .to("mock:result")
    .build()
    .unwrap();

Available Processors

Security Policy Layer

SecurityPolicyLayer is a Tower Layer that intercepts every Exchange and evaluates a SecurityPolicy before forwarding to the inner service. It is the standard way to add authorization to any Camel pipeline.

How it works

The layer produces a SecurityPolicyService that calls policy.evaluate(&mut exchange) for each exchange and branches on the AuthorizationDecision:

Usage

use std::sync::Arc;
use camel_processor::SecurityPolicyLayer;
use camel_api::security_policy::SecurityPolicy;

// Suppose you have a SecurityPolicy implementation:
let policy: Arc<dyn SecurityPolicy> = Arc::new(MyPolicy);

// Wrap it as a Tower layer:
let layer = SecurityPolicyLayer::new(policy);

// Compose in a Tower pipeline:
let service = layer.layer(my_inner_service);

When composed with other processors, the security check runs before the inner service sees the exchange:

use tower::ServiceBuilder;
use camel_processor::{SecurityPolicyLayer, LogProcessor};

let service = ServiceBuilder::new()
    .layer(SecurityPolicyLayer::new(my_policy))
    .service(LogProcessor::new(LogLevel::Info, "Authorized request".into()));

Exchanges that are denied or encounter an evaluation error never reach the inner service.

Data Formats

The DataFormat trait defines serialization/deserialization for message bodies:

Usage with RouteBuilder

use camel_builder::RouteBuilder;

let route = RouteBuilder::from("direct:in")
    .unmarshal("json")
    .marshal("xml")
    .to("mock:out")
    .build()
    .unwrap();

YAML DSL

steps:
  - unmarshal: json
  - marshal: xml

Documentation

• API Documentation
• Repository

License

Apache-2.0

Contributing

Contributions are welcome! Please see the main repository for details.