coreon-core 0.1.0

//! Processor — async transformation of an Exchange.

use crate::{error::Result, exchange::Exchange};
use async_trait::async_trait;
use std::sync::Arc;

/// The central abstraction. Processors are composed into pipelines; they
/// transform the Exchange in place (or route it elsewhere) and return `Ok(())`
/// to let the pipeline continue, or an error to trigger the error handler.
#[async_trait]
pub trait Processor: Send + Sync {
    async fn process(&self, exchange: &mut Exchange) -> Result<()>;
}

/// Adapter that lifts a closure into a `Processor`.
///
/// Closures must be `Send + Sync + 'static` and return `Result<()>`. Async
/// closures aren't stable yet, so this takes a sync closure; async work
/// belongs in dedicated `Processor` impls.
pub struct FnProcessor<F>(F);

impl<F> FnProcessor<F>
where
    F: Fn(&mut Exchange) -> Result<()> + Send + Sync + 'static,
{
    pub fn new(f: F) -> Arc<Self> {
        Arc::new(Self(f))
    }
}

#[async_trait]
impl<F> Processor for FnProcessor<F>
where
    F: Fn(&mut Exchange) -> Result<()> + Send + Sync + 'static,
{
    async fn process(&self, exchange: &mut Exchange) -> Result<()> {
        (self.0)(exchange)
    }
}

/// Predicate — a boolean test over an Exchange. Separate from Processor
/// because EIPs like Filter/Choice compose predicates, not processors.
pub trait Predicate: Send + Sync {
    fn matches(&self, exchange: &Exchange) -> bool;
}

impl<F> Predicate for F
where
    F: Fn(&Exchange) -> bool + Send + Sync,
{
    fn matches(&self, exchange: &Exchange) -> bool {
        self(exchange)
    }
}