use std::collections::{HashMap, VecDeque, hash_map::Entry};
use std::sync::Arc;
use tokio::sync::{Mutex, Notify};
use camel_api::component_metadata::ComponentMetadata;
use camel_component_api::UriConfig;
use camel_component_api::parse_uri;
use camel_component_api::{CamelError, Component, Endpoint};
use tracing::debug;
const DEFAULT_MAX_RETAINED: usize = 10_000;
#[derive(Clone, Debug)]
pub struct MockConfig {
pub max_retained: usize,
pub copy_on_exchange: bool,
pub fail_fast: bool,
pub assert_period_ms: u64,
pub any_order: bool,
}
#[derive(Debug, Clone, UriConfig)]
#[allow(dead_code)]
#[uri_scheme = "mock"]
#[uri_config(
skip_impl,
metadata(
scheme = "mock",
description = "Records exchanges for test assertions",
producer
),
crate = "camel_component_api"
)]
struct MockUriConfig {
#[uri_param(name = "retain")]
pub _retain: Option<String>,
#[uri_param(name = "copy")]
pub _copy: Option<String>,
#[uri_param(name = "failFast")]
pub _fail_fast: Option<String>,
#[uri_param(name = "expectedCount")]
pub _expected_count: Option<String>,
#[uri_param(name = "anyOrder")]
pub _any_order: Option<String>,
}
impl Default for MockConfig {
fn default() -> Self {
Self {
max_retained: DEFAULT_MAX_RETAINED,
copy_on_exchange: false,
fail_fast: false,
assert_period_ms: 0,
any_order: false,
}
}
}
impl MockConfig {
pub fn new(max_retained: usize) -> Self {
Self {
max_retained,
..Self::default()
}
}
pub fn metadata() -> ComponentMetadata {
MockUriConfig::metadata()
}
}
mod assert;
mod expectations;
mod inner;
pub use assert::MockAssertionError;
pub use expectations::MockExpectations;
pub use inner::{ExchangeAssert, MockEndpoint, MockEndpointInner};
#[derive(Clone)]
pub struct MockComponent {
registry: Arc<std::sync::Mutex<HashMap<String, Arc<MockEndpointInner>>>>,
config: MockConfig,
}
impl MockComponent {
pub fn new() -> Self {
Self::with_config(MockConfig::default())
}
pub fn with_config(config: MockConfig) -> Self {
Self {
registry: Arc::new(std::sync::Mutex::new(HashMap::new())),
config,
}
}
pub fn get_endpoint(&self, name: &str) -> Option<Arc<MockEndpointInner>> {
let registry = self
.registry
.lock()
.expect("mutex poisoned: another thread panicked while holding this lock"); registry.get(name).cloned()
}
}
impl Default for MockComponent {
fn default() -> Self {
Self::new()
}
}
fn parse_usize_param(uri_value: &str, name: &str) -> Result<usize, CamelError> {
uri_value.parse::<usize>().map_err(|_| {
CamelError::EndpointCreationFailed(format!(
"mock: invalid value for URI parameter '{name}': '{uri_value}' is not a non-negative integer"
))
})
}
fn parse_bool_param(uri_value: &str, name: &str) -> Result<bool, CamelError> {
match uri_value.to_ascii_lowercase().as_str() {
"true" => Ok(true),
"false" => Ok(false),
_ => Err(CamelError::EndpointCreationFailed(format!(
"mock: invalid value for URI parameter '{name}': '{uri_value}' is not a boolean (true|false)"
))),
}
}
impl Component for MockComponent {
fn scheme(&self) -> &str {
"mock"
}
fn metadata(&self) -> ComponentMetadata {
MockConfig::metadata()
}
fn create_endpoint(
&self,
uri: &str,
_ctx: &dyn camel_component_api::ComponentContext,
) -> Result<Box<dyn Endpoint>, CamelError> {
let parts = parse_uri(uri)?;
if parts.scheme != "mock" {
return Err(CamelError::InvalidUri(format!(
"expected scheme 'mock', got '{}'",
parts.scheme
)));
}
let name = parts.path;
if name.is_empty() {
return Err(CamelError::InvalidUri(
"mock endpoint name must be non-empty (use 'mock:<name>')".to_string(),
));
}
let max_retained = match parts.params.get("retain") {
Some(v) => {
let n = parse_usize_param(v, "retain")?;
if n == 0 {
return Err(CamelError::EndpointCreationFailed(
"mock: URI parameter 'retain' must be >= 1, got 0".to_string(),
));
}
n
}
None => self.config.max_retained,
};
let copy_on_exchange = match parts.params.get("copy") {
Some(v) => parse_bool_param(v, "copy")?,
None => self.config.copy_on_exchange,
};
let fail_fast = match parts.params.get("failFast") {
Some(v) => parse_bool_param(v, "failFast")?,
None => self.config.fail_fast,
};
let any_order = match parts.params.get("anyOrder") {
Some(v) => parse_bool_param(v, "anyOrder")?,
None => self.config.any_order,
};
let expected_count = match parts.params.get("expectedCount") {
Some(v) => Some(parse_usize_param(v, "expectedCount")?),
None => None,
};
let mut registry = self.registry.lock().map_err(|e| {
CamelError::EndpointCreationFailed(format!("mock registry lock poisoned: {e}"))
})?;
let assert_period_ms = self.config.assert_period_ms;
let (inner, fresh) = match registry.entry(name.clone()) {
Entry::Vacant(vacant) => {
let created = vacant.insert(Arc::new(MockEndpointInner {
uri: uri.to_string(),
name,
received: Arc::new(Mutex::new(VecDeque::new())),
notify: Arc::new(Notify::new()),
max_retained,
copy_on_exchange,
fail_fast,
fail_fast_error: Arc::new(std::sync::Mutex::new(None)),
assert_period_ms,
any_order,
expectations: Arc::new(std::sync::Mutex::new(MockExpectations::new())),
}));
(Arc::clone(created), true)
}
Entry::Occupied(occupied) => (Arc::clone(occupied.get()), false),
};
if fresh && let Some(n) = expected_count {
inner.expect_count(n);
}
debug!(endpoint_name = %inner.name, "mock endpoint created");
Ok(Box::new(MockEndpoint(inner)))
}
}
#[cfg(test)]
mod tests;